(self, db_path: str | Path)
| 144 | """SQLite-backed code knowledge graph.""" |
| 145 | |
| 146 | def __init__(self, db_path: str | Path) -> None: |
| 147 | self.db_path = Path(db_path) |
| 148 | self.db_path.parent.mkdir(parents=True, exist_ok=True) |
| 149 | self._conn = sqlite3.connect( |
| 150 | str(self.db_path), timeout=30, check_same_thread=False, |
| 151 | isolation_level=None, # Disable implicit transactions (#135) |
| 152 | ) |
| 153 | self._conn.row_factory = sqlite3.Row |
| 154 | self._conn.execute("PRAGMA journal_mode=WAL") |
| 155 | self._conn.execute("PRAGMA busy_timeout=5000") |
| 156 | self._init_schema() |
| 157 | # Ensure schema_version is set, then run pending migrations |
| 158 | if get_schema_version(self._conn) < 1: |
| 159 | # Fresh DB — metadata table just created by _init_schema |
| 160 | self._conn.execute( |
| 161 | "INSERT OR IGNORE INTO metadata (key, value) " |
| 162 | "VALUES ('schema_version', '1')" |
| 163 | ) |
| 164 | self._conn.commit() |
| 165 | run_migrations(self._conn) |
| 166 | self._nxg_cache: nx.DiGraph | None = None |
| 167 | self._cache_lock = threading.Lock() |
| 168 | |
| 169 | def __enter__(self) -> "GraphStore": |
| 170 | return self |
nothing calls this directly
no test coverage detected