feat(ingest): sqlite sql-source connector + typed cell rendering + dispatch (I4)
read_sql resolves connection_ref from the environment to a sqlite path, opens it read-only (mode=ro so a write in the query fails at the DB), and converts cells to their spec-5 text form: INTEGER plain decimal, REAL shortest round-trip (repr), TEXT verbatim, SQL NULL empty string, BLOB/other IngestError (never silent coercion). materialize now dispatches file->read_csv, sql->read_sql; http stays I6-refused. Pure stdlib (sqlite3/os/contextlib) — MAF-free context-seam guard intact. No spec change (frozen); the SQL type/number rules were delegated to I4. Refs: shared/ingest-spec.md 4/5/8 · sesjonsplan I4
This commit is contained in:
parent
fdcbaffc88
commit
d7e5f2fec7
1 changed files with 110 additions and 21 deletions
|
|
@ -1,4 +1,4 @@
|
|||
"""Ingest layer — deterministic connectors + OKF-bundle materialization (I2, `file`/CSV).
|
||||
"""Ingest layer — deterministic connectors + OKF-bundle materialization (I2 `file`/CSV, I4 `sql`).
|
||||
|
||||
Implements the framework-neutral contract in ``shared/ingest-spec.md`` (normative, frozen):
|
||||
one JSON **manifest** per source coupling (§4), schema-validated fail-fast BEFORE any source
|
||||
|
|
@ -20,11 +20,14 @@ Invariants carried by this module (guarded by ``tests/test_ingest_loadbearing.py
|
|||
* **Determinism (§5, §10, §11):** explicit required ``ingested_at`` (no wall-clock default),
|
||||
byte-stable rendering, idempotent re-materialization.
|
||||
|
||||
Decisions on spec-silent points are pinned here and in the golden fixture (plan Assumptions):
|
||||
CSV cells are text-verbatim (the §5 number rules bite typed ``sql`` values, I4); relative
|
||||
``root`` resolves against the manifest file's directory; CSV files are read ``utf-8-sig``
|
||||
(a BOM never leaks into the first header cell); header cells are escaped identically to data
|
||||
cells; empty/ragged CSV and missing ``root``/``query`` fail fast as ``IngestError``.
|
||||
Decisions on spec-silent points are pinned here and in the golden fixtures (plan Assumptions):
|
||||
CSV cells are text-verbatim; relative ``root`` resolves against the manifest file's directory;
|
||||
CSV files are read ``utf-8-sig`` (a BOM never leaks into the first header cell); header cells
|
||||
are escaped identically to data cells; empty/ragged CSV and missing ``root``/``query`` fail
|
||||
fast as ``IngestError``. For ``sql`` (I4): ``connection_ref`` names an env var whose value is a
|
||||
filesystem path to a sqlite database opened read-only; the §5 numeric rules bite the typed
|
||||
values (``_sql_value_to_text``) — INTEGER→plain decimal, REAL→``repr`` (shortest round-trip),
|
||||
TEXT verbatim, SQL NULL→empty string, BLOB/other→``IngestError`` (never silent coercion).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -33,10 +36,13 @@ import csv
|
|||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
from contextlib import closing
|
||||
from pathlib import Path
|
||||
from typing import Literal
|
||||
from urllib.parse import urlsplit
|
||||
from urllib.parse import quote, urlsplit
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator, model_validator
|
||||
|
||||
|
|
@ -197,6 +203,81 @@ def read_csv(root: str | Path, query: str, *, max_rows: int) -> tuple[list[str],
|
|||
return header, rows
|
||||
|
||||
|
||||
def _sql_value_to_text(value: object) -> str:
|
||||
"""Convert one sqlite cell value to its §5 text form (BEFORE the §5 table escaping).
|
||||
|
||||
§5 for ``sql``: SQL NULL → empty string; integers in plain decimal; non-integral numbers
|
||||
in shortest round-trip decimal form; text verbatim; any other value type MUST fail —
|
||||
never silent coercion. sqlite3's default typing yields ``None``/``int``/``float``/``str``/
|
||||
``bytes``, so BLOB (bytes) and anything unexpected raise. ``bool`` is an ``int`` subclass
|
||||
sqlite never emits — guarded explicitly so a stray one can never render as ``str(True)``."""
|
||||
if value is None:
|
||||
return "" # SQL NULL → empty string
|
||||
if isinstance(value, bool):
|
||||
raise IngestError(
|
||||
f"unsupported SQL cell type bool ({value!r}) — never silent coercion (spec §5)"
|
||||
)
|
||||
if isinstance(value, int):
|
||||
return str(value) # plain decimal (arbitrary precision)
|
||||
if isinstance(value, float):
|
||||
# repr is Python's shortest round-trip decimal form; extreme magnitudes use E-notation.
|
||||
return repr(value)
|
||||
if isinstance(value, str):
|
||||
return value # verbatim; table escaping happens in render_table
|
||||
raise IngestError(
|
||||
f"unsupported SQL cell type {type(value).__name__} — ingest spec §5 requires "
|
||||
"integer/float/text/NULL; any other type MUST fail (never silent coercion)"
|
||||
)
|
||||
|
||||
|
||||
def read_sql(
|
||||
connection_ref: str, query: str, *, max_rows: int
|
||||
) -> tuple[list[str], list[list[str]]]:
|
||||
"""Execute a ``sql``-source extraction: a read-only SELECT against the sqlite database
|
||||
whose path is resolved at run time from the env var ``connection_ref`` names (§4).
|
||||
|
||||
Pinned I4 decision: ``connection_ref`` names an environment variable whose value is a
|
||||
filesystem PATH to a sqlite database (the I4 sqlite-fixture reading of §4's "connection
|
||||
string or database path"). Opened read-only (``mode=ro``) so a write in ``query`` fails
|
||||
at the DB — §4's "SHOULD enforce read-only" honoured robustly, not by fragile string
|
||||
parsing; ``Connection.execute`` runs exactly one statement (§4). Cells are converted to
|
||||
their §5 text form (``_sql_value_to_text``); the streaming ``max_rows`` cap is an ERROR
|
||||
the moment it is exceeded, never a silent truncation (§8). Env-unset, missing file, and
|
||||
any ``sqlite3`` error (syntax, write attempt, corruption) raise ``IngestError`` fail-fast.
|
||||
Pure: no writes, no logging (materialization owns the §8 log)."""
|
||||
dsn = os.environ.get(connection_ref)
|
||||
if not dsn:
|
||||
raise IngestError(
|
||||
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 IngestError(
|
||||
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; the
|
||||
# absolute-path form file:/abs/path 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 IngestError(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 IngestError(
|
||||
f"extraction {query!r} exceeds max_rows={max_rows} (error, never "
|
||||
"silent truncation — ingest spec §8)"
|
||||
)
|
||||
rows.append([_sql_value_to_text(value) for value in row])
|
||||
except (sqlite3.Error, sqlite3.Warning) as exc:
|
||||
raise IngestError(f"sql extraction failed for query {query!r}: {exc}") from exc
|
||||
return header, rows
|
||||
|
||||
|
||||
def _escape_cell(value: str) -> str:
|
||||
# §5 escape order is load-bearing: `\` FIRST (else the backslash introduced by
|
||||
# pipe-escaping gets double-escaped), then `|`, then newlines → single space with
|
||||
|
|
@ -208,9 +289,10 @@ def _escape_cell(value: str) -> str:
|
|||
def render_table(header: list[str], rows: list[list[str]]) -> str:
|
||||
"""Render extraction rows as the §5 markdown-table body (LF-only, one trailing newline).
|
||||
|
||||
Cells are text-verbatim after escaping (pinned decision: on the ``file``/CSV path every
|
||||
cell is a string; §5's integer/float/NULL clauses bite typed ``sql`` values in I4).
|
||||
Header cells are escaped identically to data cells (pinned decision)."""
|
||||
Cells are strings by the time they reach here: the ``file``/CSV path is text-verbatim, and
|
||||
``sql`` values are converted to their §5 text form in ``read_sql`` (``_sql_value_to_text``:
|
||||
integers/floats/NULL) BEFORE this escaping. Header cells are escaped identically to data
|
||||
cells (pinned decision)."""
|
||||
|
||||
def line(cells: list[str]) -> str:
|
||||
return "| " + " | ".join(_escape_cell(cell) for cell in cells) + " |"
|
||||
|
|
@ -308,25 +390,32 @@ def materialize(
|
|||
)
|
||||
manifest_file = Path(manifest_path)
|
||||
manifest, stamp = load_manifest(manifest_file)
|
||||
if not isinstance(manifest.source, FileSource):
|
||||
source = manifest.source
|
||||
if isinstance(source, HttpSource):
|
||||
raise IngestError(
|
||||
f"source type {manifest.source.type!r} has no connector in this implementation "
|
||||
"yet (sql arrives in I4, http is a gated extension point for I6) — the schema "
|
||||
"validates it, execution defers"
|
||||
f"source type {source.type!r} is a gated extension point (I6) with no connector "
|
||||
"in this implementation — the schema validates it, execution defers"
|
||||
)
|
||||
# Pinned decision: a relative root resolves against the manifest file's directory —
|
||||
# never the process cwd, or the extraction would not be reproducible.
|
||||
root = Path(manifest.source.root)
|
||||
if not root.is_absolute():
|
||||
root = manifest_file.parent / root
|
||||
# Pinned decision (file): a relative root resolves against the manifest file's directory —
|
||||
# never the process cwd, or the extraction would not be reproducible. (sql resolves its
|
||||
# source at run time from the env var connection_ref names, inside read_sql.)
|
||||
if isinstance(source, FileSource):
|
||||
root = Path(source.root)
|
||||
if not root.is_absolute():
|
||||
root = manifest_file.parent / root
|
||||
|
||||
# Stage everything in memory BEFORE any disk mutation.
|
||||
staged: list[tuple[str, str]] = []
|
||||
for extraction in manifest.extractions:
|
||||
header, rows = read_csv(root, extraction.query, max_rows=extraction.max_rows)
|
||||
if isinstance(source, FileSource):
|
||||
header, rows = read_csv(root, extraction.query, max_rows=extraction.max_rows)
|
||||
else: # SqlSource — http already refused above
|
||||
header, rows = read_sql(
|
||||
source.connection_ref, extraction.query, max_rows=extraction.max_rows
|
||||
)
|
||||
_LOGGER.info(
|
||||
"source call: source=%s ingested_at=%s rows=%d",
|
||||
manifest.source.id,
|
||||
source.id,
|
||||
ingested_at,
|
||||
len(rows),
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue