"""SQLite artifact extractor (Section 45). Extracts configured tables of a SQLite database into JSON Lines artifacts under the output root: //.jsonl Requirements implemented here (Section 45.4 - 45.6): * read-only source access via snapshot (FR-034): the database file and its WAL sidecar are copied to a temporary location before reading, so a live SCADA system writing in WAL mode is never disturbed and a partially written transaction is never captured; * deterministic output (FR-031): rows ordered by primary key (rowid fallback), stable JSON encoding, no run-dependent values; * stale artifact removal (FR-033): artifact files for tables that no longer exist in the database are deleted. """ from __future__ import annotations import shutil import sqlite3 import tempfile from contextlib import contextmanager from pathlib import Path from git_auto_tracker.errors import CollectionError from git_auto_tracker.extraction.base import ( ExtractionError, ExtractionResult, ExtractionSource, artifact_filename, artifact_folder, remove_if_exists, to_jsonl, write_if_changed, ) _USER_TABLES_SQL = ( "SELECT name FROM sqlite_master " "WHERE type = 'table' AND name NOT LIKE 'sqlite_%' " "ORDER BY name" ) @contextmanager def _snapshot_database(source_path: Path): """Copy the database (and WAL sidecar) to temp; open the copy (45.5). Opening the snapshot instead of the live file guarantees read-only access to the source (FR-034) and a consistent view even when the source is being written concurrently (Section 20). The connection is closed before the temporary directory is removed; otherwise Windows keeps the file locked and cleanup fails. """ with tempfile.TemporaryDirectory(prefix="git-auto-tracker-extract-") as tmp: snapshot = Path(tmp) / source_path.name shutil.copy2(source_path, snapshot) for sidecar in ("-wal", "-shm"): live = source_path.with_name(source_path.name + sidecar) if live.exists(): shutil.copy2(live, snapshot.with_name(snapshot.name + sidecar)) connection = sqlite3.connect(f"file:{snapshot}?mode=ro", uri=True) try: yield connection finally: connection.close() def _order_by(table: str, columns: list[str]) -> str: """Deterministic ORDER BY clause: primary key columns, rowid fallback.""" pk_columns = [c for c in columns if not c.startswith("_rowid_")] quoted = ", ".join(f'"{c}"' for c in pk_columns) if pk_columns else "rowid" return quoted def _primary_key_columns(cursor: sqlite3.Cursor, table: str) -> list[str]: """Primary key column names for a table, in index order; [] if none.""" pk_columns: list[str] = [] for row in cursor.execute(f'PRAGMA table_info("{table}")'): # row: (cid, name, type, notnull, dflt_value, pk) if row[5]: # pk index (1-based ordinal) while len(pk_columns) < row[5]: pk_columns.append("") pk_columns[row[5] - 1] = str(row[1]) return pk_columns def _sanitize(identifier: str) -> str: """Quote an identifier for SQL; rejects embedded quotes (FR-029).""" if '"' in identifier: raise ExtractionError(f"invalid table name: {identifier!r}") return f'"{identifier}"' class SqliteExtractor: """Extracts SQLite tables to JSON Lines artifacts (Section 45).""" name = "sqlite" def extract( self, source: ExtractionSource, repository_root: Path, output_root: Path, ) -> ExtractionResult: source_path = repository_root / Path(source.path) if not source_path.is_file(): raise ExtractionError(f"source database not found: {source.path}") folder = output_root / artifact_folder(source) result = ExtractionResult(source=source.path) try: with _snapshot_database(source_path) as connection: connection.row_factory = sqlite3.Row cursor = connection.cursor() tables = self._resolve_tables(cursor, source.tables) for table in tables: artifact = folder / artifact_filename(table) rel = self._relative(repository_root, artifact) content = self._dump_table(cursor, table) if write_if_changed(artifact, content): result.artifacts.append(rel) else: result.unchanged = True result.removed = self._remove_stale(folder, tables, repository_root) except ExtractionError: raise except sqlite3.Error as exc: raise ExtractionError(f"failed to extract {source.path}: {exc}") from exc except OSError as exc: raise ExtractionError(f"cannot read source {source.path}: {exc}") from exc if not result.artifacts and not result.removed and not result.unchanged: # Source had no user tables at all result.unchanged = True return result @staticmethod def _relative(repository_root: Path, path: Path) -> str: try: return path.resolve().relative_to(repository_root.resolve()).as_posix() except ValueError: return path.as_posix() def _resolve_tables(self, cursor: sqlite3.Cursor, wanted: tuple[str, ...]) -> list[str]: """Configured tables validated against the database; empty = all.""" available = [ str(row[0]) for row in cursor.execute(_USER_TABLES_SQL) ] if not wanted: return available known = set(available) for table in wanted: if table not in known: raise ExtractionError( f"configured table {table!r} does not exist; " f"available: {', '.join(available) or 'none'}" ) _sanitize(table) return [t for t in available if t in set(wanted)] def _dump_table(self, cursor: sqlite3.Cursor, table: str) -> str: """One JSON object per row, ordered by primary key (Section 45.4).""" pk = _primary_key_columns(cursor, table) order = ", ".join(f'"{c}"' for c in pk) if pk else "rowid" rows = [ {str(key): value for key, value in dict(row).items()} for row in cursor.execute( f"SELECT * FROM {_sanitize(table)} ORDER BY {order}" ) ] return to_jsonl(rows) def _remove_stale( self, folder: Path, current_tables: list[str], repository_root: Path, ) -> list[str]: """Delete artifact files whose table no longer exists (FR-033).""" expected = {artifact_filename(table) for table in current_tables} removed: list[str] = [] if not folder.is_dir(): return removed for existing in sorted(folder.iterdir()): if existing.is_file() and existing.name not in expected and remove_if_exists(existing): removed.append(self._relative(repository_root, existing)) return removed def extractor_for(name: str) -> SqliteExtractor: """Registry lookup for configured extractor names (Section 45.3).""" registry = {SqliteExtractor.name: SqliteExtractor} factory = registry.get(name) if factory is None: raise CollectionError( f"unknown extraction source format: {name!r}; available: {list(registry)}" ) return factory()