← back to Unclaimed Property Platform
services/common/sqlite_repo.py
306 lines
"""Concrete SQLite-backed Repository + filesystem ObjectStore for the local prototype.
Implements the same interface the production PostgreSQL repository and cloud object store
will implement, so services/ingestion/ingest.py is written once and runs unchanged against
either. Everything here is local and $0.
Idempotency is enforced two ways:
1. ingestion_batch UNIQUE(jurisdiction_id, checksum) — the same source FILE never
produces two batches.
2. property UNIQUE(jurisdiction_id, source_property_id) — a re-delivered RECORD upserts
in place; it never creates a duplicate property row.
"""
from __future__ import annotations
import os
import sqlite3
import uuid
from datetime import datetime, timezone
from pathlib import Path
_SCHEMA_PATH = Path(__file__).resolve().parents[2] / "db" / "schema.sql"
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
class FileObjectStore:
"""Filesystem stand-in for cloud object storage (immutable raw archive)."""
def __init__(self, root: str | os.PathLike) -> None:
self.root = Path(root)
self.root.mkdir(parents=True, exist_ok=True)
def read_bytes(self, uri: str) -> bytes:
return (self.root / uri).read_bytes()
def write_bytes(self, uri: str, data: bytes) -> None:
target = self.root / uri
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(data)
class SqliteRepository:
def __init__(self, db_path: str = ":memory:") -> None:
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
self.conn.execute("PRAGMA foreign_keys = ON")
self._init_schema()
def _init_schema(self) -> None:
# Load the PostgreSQL schema, downgrading the few types SQLite doesn't share.
ddl = _SCHEMA_PATH.read_text()
ddl = ddl.replace("NUMERIC", "REAL").replace("TIMESTAMP", "TEXT").replace("BOOLEAN", "INTEGER")
self.conn.executescript(ddl)
# Minimal jurisdiction seed so FK-free prototype inserts succeed.
self.conn.execute(
"INSERT OR IGNORE INTO jurisdiction(jurisdiction_id, name, status) VALUES (?,?,?)",
("SAMPLE", "Synthetic Sample State", "prototype"),
)
self.conn.commit()
# --- ingestion Repository protocol -------------------------------------
def batch_exists(self, jurisdiction: str, checksum: str) -> bool:
# Only a SUCCESSFULLY-completed batch blocks a re-run. A 'failed'/'running' row must
# NOT gate a retry, or a partially-ingested file could never be reprocessed (C1).
row = self.conn.execute(
"""SELECT 1 FROM ingestion_batch
WHERE jurisdiction_id=? AND checksum=?
AND status IN ('completed', 'completed_with_errors')""",
(jurisdiction, checksum),
).fetchone()
return row is not None
def begin_batch(self, jurisdiction: str, source_uri: str, checksum: str,
parser_version: str) -> str:
batch_id = str(uuid.uuid4())
self.conn.execute(
"INSERT OR IGNORE INTO jurisdiction(jurisdiction_id, name, status) VALUES (?,?,?)",
(jurisdiction, f"Jurisdiction {jurisdiction}", "prototype"),
)
self.conn.execute(
"""INSERT INTO ingestion_batch
(batch_id, jurisdiction_id, source_uri, checksum, parser_version,
received_at, status)
VALUES (?,?,?,?,?,?, 'running')""",
(batch_id, jurisdiction, source_uri, checksum, parser_version, _now()),
)
self.conn.commit()
return batch_id
def upsert_property(self, batch_id: str, record) -> None:
"""Idempotent upsert of one canonical property + its owner + search doc."""
jur = record.jurisdiction
spid = record.source_property_id
existing = self.conn.execute(
"SELECT property_id FROM property WHERE jurisdiction_id=? AND source_property_id=?",
(jur, spid),
).fetchone()
amount = float(record.amount) if record.amount is not None else None
if existing:
property_id = existing["property_id"]
self.conn.execute(
"UPDATE property SET holder_name_raw=?, property_type=?, amount=? WHERE property_id=?",
(record.holder_name_raw, record.property_type, amount, property_id),
)
else:
property_id = str(uuid.uuid4())
self.conn.execute(
"""INSERT INTO property
(property_id, jurisdiction_id, source_property_id, holder_name_raw,
property_type, amount, status)
VALUES (?,?,?,?,?,?, 'active')""",
(property_id, jur, spid, record.holder_name_raw, record.property_type, amount),
)
# Non-destructive version history: close the prior open version, then append.
# Setting effective_to makes point-in-time reconstruction possible (M3).
now = _now()
self.conn.execute(
"UPDATE property_version SET effective_to=? WHERE property_id=? AND effective_to IS NULL",
(now, property_id),
)
self.conn.execute(
"""INSERT INTO property_version
(property_version_id, property_id, batch_id, effective_from, raw_payload,
raw_record_hash)
VALUES (?,?,?,?,?,?)""",
(str(uuid.uuid4()), property_id, batch_id, now,
record.raw_payload, record.raw_record_hash),
)
# Owner: UPDATE the existing row in place, preserving owner_id so entity_link
# references survive a re-delivery. DELETE+INSERT would (a) break the entity_link
# FK and (b) throw away entity-resolution work every refresh (C4 / Cody #3).
existing_owner = self.conn.execute(
"SELECT owner_id FROM owner WHERE property_id=?", (property_id,)
).fetchone()
if existing_owner:
self.conn.execute(
"""UPDATE owner SET owner_type=?, owner_name_raw=?, owner_name_normalized=?,
city_normalized=?, region=?, postal_code=? WHERE property_id=?""",
(record.owner_type, record.owner_name_raw, record.owner_name_normalized,
record.city_normalized, record.region, record.postal_code, property_id),
)
else:
self.conn.execute(
"""INSERT INTO owner
(owner_id, property_id, owner_type, owner_name_raw, owner_name_normalized,
city_normalized, region, postal_code)
VALUES (?,?,?,?,?,?,?,?)""",
(str(uuid.uuid4()), property_id, record.owner_type, record.owner_name_raw,
record.owner_name_normalized, record.city_normalized, record.region,
record.postal_code),
)
# Search publication state (masked projection only).
self.conn.execute(
"""INSERT INTO search_document_state
(property_id, index_version, is_public, is_suppressed, owner_name_masked, amount_band)
VALUES (?,?,?,?,?,?)
ON CONFLICT(property_id) DO UPDATE SET
index_version = index_version + 1,
owner_name_masked = excluded.owner_name_masked,
amount_band = excluded.amount_band""",
(property_id, 0, 1, 0, record.owner_name_masked, record.amount_band),
)
self.conn.commit()
def finish_batch(self, batch_id: str, accepted: int, rejected: int, status: str,
expected: int | None = None, reconciliation: str | None = None) -> None:
self.conn.execute(
"""UPDATE ingestion_batch
SET accepted_count=?, rejected_count=?, status=?, expected_count=?, reconciliation=?
WHERE batch_id=?""",
(accepted, rejected, status, expected, reconciliation, batch_id),
)
self.conn.commit()
def batch_reconciliation(self, batch_id: str) -> dict | None:
r = self.conn.execute(
"""SELECT accepted_count, rejected_count, expected_count, reconciliation, status
FROM ingestion_batch WHERE batch_id=?""",
(batch_id,),
).fetchone()
return dict(r) if r else None
# --- convenience reads for tests / review tools ------------------------
def count_properties(self, jurisdiction: str | None = None) -> int:
if jurisdiction:
return self.conn.execute(
"SELECT COUNT(*) FROM property WHERE jurisdiction_id=?", (jurisdiction,)
).fetchone()[0]
return self.conn.execute("SELECT COUNT(*) FROM property").fetchone()[0]
def count_versions(self, jurisdiction: str | None = None) -> int:
if jurisdiction:
return self.conn.execute(
"""SELECT COUNT(*) FROM property_version pv
JOIN property p ON p.property_id = pv.property_id
WHERE p.jurisdiction_id=?""",
(jurisdiction,),
).fetchone()[0]
return self.conn.execute("SELECT COUNT(*) FROM property_version").fetchone()[0]
def masked_search(self, name_query: str, limit: int = 20) -> list[dict]:
"""In-DB fallback for the OpenSearch masked search (prototype only).
Returns the SAME allowlisted projection as the production API and runs each row
through assert_public_safe — so a raw-PII leak fails a test instead of shipping.
"""
from services.common.normalize import normalize_search
from services.common.public_projection import assert_public_safe
# Fully-collapsed alphanumeric key on BOTH sides fixes apostrophe/hyphen recall
# ("Oneil" finds "O'Neil"). REPLACE(...,' ','') matches the stored space-collapsed
# normalized name. (Prototype: full scan; production = analyzer-backed index.)
needle = normalize_search(name_query)
# Reject empty / all-punctuation queries — an empty needle would LIKE '%' and walk
# the whole table, a mass-enumeration primitive (m2).
if not needle:
raise ValueError("search query must contain at least one alphanumeric token")
rows = self.conn.execute(
"""SELECT s.owner_name_masked, s.amount_band, p.jurisdiction_id,
p.property_id, p.holder_name_raw, p.property_type, o.city_normalized
FROM owner o
JOIN property p ON p.property_id = o.property_id
JOIN search_document_state s ON s.property_id = p.property_id
WHERE s.is_public=1 AND s.is_suppressed=0
AND REPLACE(o.owner_name_normalized, ' ', '') LIKE ?
LIMIT ?""",
(f"%{needle}%", limit),
).fetchall()
# Build ONLY the public projection — holder_name_raw/property_id/city_normalized
# are read internally but never returned; note holder name is a masked-off public
# field per the schema design, so we surface it under the allowlisted key.
out = []
for r in rows:
out.append(assert_public_safe({
"public_reference": r["property_id"],
"jurisdiction": r["jurisdiction_id"],
"owner_name_masked": r["owner_name_masked"],
"owner_city": (r["city_normalized"] or None),
"holder_name": r["holder_name_raw"],
"property_type": r["property_type"],
"amount_band": r["amount_band"],
}))
return out
# --- state suppression / reinstatement (state-integration API surface) ---------------
# A state instructs "remove from public search" (privacy / fraud takedown). We flip the
# search-publication flag — the record leaves anonymous search IMMEDIATELY (masked_search
# filters is_suppressed=0) while ALL underlying data + version history is retained for
# audit. Every action writes an immutable audit_event. In production this is authenticated
# to a state actor; here it's a repository method the state-integration layer would call.
def _property_id(self, jurisdiction: str, source_property_id: str) -> str | None:
row = self.conn.execute(
"SELECT property_id FROM property WHERE jurisdiction_id=? AND source_property_id=?",
(jurisdiction, source_property_id),
).fetchone()
return row["property_id"] if row else None
def _set_suppression(self, jurisdiction: str, source_property_id: str,
suppressed: bool, actor: str, reason: str = "") -> bool:
pid = self._property_id(jurisdiction, source_property_id)
if pid is None:
return False
before = self.conn.execute(
"SELECT is_suppressed FROM search_document_state WHERE property_id=?", (pid,)
).fetchone()
before_val = None if before is None else before["is_suppressed"]
self.conn.execute(
"UPDATE search_document_state SET is_suppressed=? WHERE property_id=?",
(1 if suppressed else 0, pid),
)
self.conn.execute(
"""INSERT INTO audit_event
(audit_event_id, actor, action, resource, before_hash, after_hash, created_at)
VALUES (?,?,?,?,?,?,?)""",
(str(uuid.uuid4()), actor, "suppress" if suppressed else "reinstate",
pid, str(before_val), "1" if suppressed else "0", _now()),
)
self.conn.commit()
return True
def suppress_property(self, jurisdiction: str, source_property_id: str,
actor: str, reason: str = "") -> bool:
"""Remove a record from public search (retains all data + history). Returns False
if the record isn't found."""
return self._set_suppression(jurisdiction, source_property_id, True, actor, reason)
def reinstate_property(self, jurisdiction: str, source_property_id: str,
actor: str, reason: str = "") -> bool:
"""Return a previously-suppressed record to public search."""
return self._set_suppression(jurisdiction, source_property_id, False, actor, reason)
def audit_trail(self, resource_property_id: str) -> list[dict]:
rows = self.conn.execute(
"SELECT actor, action, created_at FROM audit_event WHERE resource=? ORDER BY created_at",
(resource_property_id,),
).fetchall()
return [dict(r) for r in rows]