Spaces:
Running
Running
File size: 1,545 Bytes
b7c46f4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | """
Connection to the intel store. Phase 1 runs on a local SQLite file (ADR-0020);
the schema in backend/sql/intel_schema.sql is portable so the same tables move
to Postgres/Supabase later.
INTEL_DB_PATH selects the file. When it's unset the store is "off": the CLI
refuses to run and hooks from other systems (the news pipeline's event writer)
silently do nothing — so the deployed backend, which has no intel store yet,
behaves exactly as before.
"""
import os
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
SCHEMA_PATH = Path(__file__).resolve().parent.parent / "sql" / "intel_schema.sql"
def db_path() -> str | None:
return os.environ.get("INTEL_DB_PATH") or None
def enabled() -> bool:
return db_path() is not None
def connect(path: str | None = None) -> sqlite3.Connection:
path = path or db_path()
if not path:
raise RuntimeError("INTEL_DB_PATH is not set — the intel store is off (see backend/intel/README.md).")
if path != ":memory:":
Path(path).parent.mkdir(parents=True, exist_ok=True)
con = sqlite3.connect(path)
con.row_factory = sqlite3.Row
con.execute("pragma foreign_keys = on")
if path != ":memory:":
con.execute("pragma journal_mode = wal")
return con
def init(con: sqlite3.Connection) -> None:
con.executescript(SCHEMA_PATH.read_text(encoding="utf8"))
from intel.registry import seed
seed(con)
con.commit()
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|