feat(connectors): add the sql connector (read-only sqlite, env-resolved)
TDD step 4: read_sql executes one read-only SELECT against the sqlite database whose path is resolved at run time from the env var named by connection_ref (credentials never in the manifest). Read-only enforced by the connection mode (file:...?mode=ro), single statement via Connection.execute, §5 cell text conversion, streaming max_rows cap, and every sqlite failure wrapped in SourceError. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QeqhJpYQyghASjiJo5EhGg
This commit is contained in:
parent
af1849f0b3
commit
c94ed6d525
2 changed files with 153 additions and 0 deletions
|
|
@ -8,9 +8,13 @@ from __future__ import annotations
|
|||
|
||||
import csv
|
||||
import os
|
||||
import sqlite3
|
||||
from contextlib import closing
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote
|
||||
|
||||
from .errors import SourceError
|
||||
from .render import sql_value_to_text
|
||||
|
||||
|
||||
def _safe_resolve(root: Path, relative: str) -> Path:
|
||||
|
|
@ -65,3 +69,50 @@ def read_csv(root: str | Path, query: str, *, max_rows: int) -> tuple[list[str],
|
|||
)
|
||||
rows.append(row)
|
||||
return header, rows
|
||||
|
||||
|
||||
def read_sql(
|
||||
connection_ref: str, query: str, *, max_rows: int
|
||||
) -> tuple[list[str], list[list[str]]]:
|
||||
"""Execute a `sql`-source extraction: one read-only SELECT against sqlite.
|
||||
|
||||
`connection_ref` names an environment variable whose value is the
|
||||
database path — resolved at run time, never from the manifest (§4).
|
||||
Opened read-only (mode=ro) so a write in `query` fails at the DB —
|
||||
§4's "SHOULD enforce read-only" honoured robustly, not by string
|
||||
parsing; Connection.execute runs exactly one statement. Cells are
|
||||
converted to their §5 text form; the streaming max_rows cap is an
|
||||
ERROR the moment it is exceeded (§8). Env-unset, missing file, and
|
||||
any sqlite3 error raise typed errors fail-fast.
|
||||
"""
|
||||
dsn = os.environ.get(connection_ref)
|
||||
if not dsn:
|
||||
raise SourceError(
|
||||
f"sql source connection_ref {connection_ref!r} is not set in the environment "
|
||||
"(paths/credentials resolve at run time, never from the manifest — §4)"
|
||||
)
|
||||
db_file = Path(dsn)
|
||||
if not db_file.is_file():
|
||||
raise SourceError(
|
||||
f"sql connection_ref {connection_ref!r} points at a missing database file: {db_file}"
|
||||
)
|
||||
# quote keeps '/' but encodes spaces/'?'/'#' so an odd path can't corrupt
|
||||
# the URI; file:{abs path}?mode=ro is sqlite's documented read-only open.
|
||||
uri = f"file:{quote(str(db_file))}?mode=ro"
|
||||
try:
|
||||
with closing(sqlite3.connect(uri, uri=True)) as conn:
|
||||
cursor = conn.execute(query)
|
||||
if cursor.description is None: # a SELECT always has columns; defensive
|
||||
raise SourceError(f"sql extraction returned no columns: {query!r}")
|
||||
header = [column[0] for column in cursor.description]
|
||||
rows: list[list[str]] = []
|
||||
for row in cursor:
|
||||
if len(rows) >= max_rows:
|
||||
raise SourceError(
|
||||
f"extraction {query!r} exceeds max_rows={max_rows} "
|
||||
"(error, never silent truncation — spec §8)"
|
||||
)
|
||||
rows.append([sql_value_to_text(value) for value in row])
|
||||
except (sqlite3.Error, sqlite3.Warning) as exc:
|
||||
raise SourceError(f"sql extraction failed for query {query!r}: {exc}") from exc
|
||||
return header, rows
|
||||
|
|
|
|||
102
tests/test_sql_connector.py
Normal file
102
tests/test_sql_connector.py
Normal file
|
|
@ -0,0 +1,102 @@
|
|||
"""The `sql` connector (ingest-spec §4, §8).
|
||||
|
||||
sqlite via connection_ref env-var resolution (the ref names the variable;
|
||||
its value is the database path), read-only open, single statement,
|
||||
streaming max_rows cap, typed errors for every sqlite failure.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from llm_ingestion_okf.connectors import read_sql
|
||||
from llm_ingestion_okf.errors import IngestError, SourceError
|
||||
|
||||
REF = "OKF_TEST_DB"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
path = tmp_path / "fixture.db"
|
||||
with sqlite3.connect(path) as conn:
|
||||
conn.execute("CREATE TABLE t (id INTEGER, name TEXT, score REAL, note TEXT)")
|
||||
conn.executemany(
|
||||
"INSERT INTO t VALUES (?, ?, ?, ?)",
|
||||
[(1, "alpha", 1.5, None), (2, "beta", 0.1, "x|y")],
|
||||
)
|
||||
monkeypatch.setenv(REF, str(path))
|
||||
return path
|
||||
|
||||
|
||||
# --- happy path: §5 text form, source order ---
|
||||
|
||||
|
||||
def test_reads_header_and_rows_as_text(db_path: Path) -> None:
|
||||
header, rows = read_sql(REF, "SELECT id, name, score, note FROM t ORDER BY id", max_rows=10)
|
||||
assert header == ["id", "name", "score", "note"]
|
||||
# NULL -> empty string; int plain decimal; float shortest round-trip; text verbatim.
|
||||
assert rows == [["1", "alpha", "1.5", ""], ["2", "beta", "0.1", "x|y"]]
|
||||
|
||||
|
||||
# --- env-var resolution (credentials never in the manifest, §4) ---
|
||||
|
||||
|
||||
def test_unset_connection_ref_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv(REF, raising=False)
|
||||
with pytest.raises(SourceError):
|
||||
read_sql(REF, "SELECT 1", max_rows=10)
|
||||
|
||||
|
||||
def test_empty_connection_ref_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv(REF, "")
|
||||
with pytest.raises(SourceError):
|
||||
read_sql(REF, "SELECT 1", max_rows=10)
|
||||
|
||||
|
||||
def test_missing_database_file_rejected(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv(REF, str(tmp_path / "nope.db"))
|
||||
with pytest.raises(SourceError):
|
||||
read_sql(REF, "SELECT 1", max_rows=10)
|
||||
|
||||
|
||||
# --- read-only, single statement, typed sqlite errors ---
|
||||
|
||||
|
||||
def test_write_statement_fails_at_the_database(db_path: Path) -> None:
|
||||
"""Read-only is enforced by the connection mode, not string parsing."""
|
||||
with pytest.raises(SourceError):
|
||||
read_sql(REF, "DELETE FROM t", max_rows=10)
|
||||
|
||||
|
||||
def test_multiple_statements_rejected(db_path: Path) -> None:
|
||||
with pytest.raises(SourceError):
|
||||
read_sql(REF, "SELECT 1; SELECT 2", max_rows=10)
|
||||
|
||||
|
||||
def test_syntax_error_wrapped_in_typed_error(db_path: Path) -> None:
|
||||
with pytest.raises(SourceError):
|
||||
read_sql(REF, "SELEC nonsense", max_rows=10)
|
||||
|
||||
|
||||
def test_blob_cell_fails_typed(db_path: Path, tmp_path: Path) -> None:
|
||||
"""Never silent coercion (§5): a BLOB cell is a typed IngestError."""
|
||||
with sqlite3.connect(tmp_path / "fixture.db") as conn:
|
||||
conn.execute("INSERT INTO t VALUES (3, 'blob', 0.0, x'DEADBEEF')")
|
||||
with pytest.raises(IngestError):
|
||||
read_sql(REF, "SELECT note FROM t WHERE id = 3", max_rows=10)
|
||||
|
||||
|
||||
# --- max_rows cap (§8): error, never silent truncation ---
|
||||
|
||||
|
||||
def test_row_count_at_cap_is_allowed(db_path: Path) -> None:
|
||||
_, rows = read_sql(REF, "SELECT id FROM t ORDER BY id", max_rows=2)
|
||||
assert len(rows) == 2
|
||||
|
||||
|
||||
def test_exceeding_cap_is_an_error(db_path: Path) -> None:
|
||||
with pytest.raises(SourceError):
|
||||
read_sql(REF, "SELECT id FROM t ORDER BY id", max_rows=1)
|
||||
Loading…
Add table
Add a link
Reference in a new issue