feat(pfs): code_element / code_element_review / code_event / code_family tables + writers (refs #684 #686 #687)

src/pfs/codetables.py

@@ -0,0 +1,144 @@
+"""DDL and writers for the code-family tables in the ``pfs`` schema.
+
+``pfs.code_element``        one row per (code, element) with its FR anchor
+``pfs.code_element_review`` element lines the classifier could not place
+``pfs.code_event``          dated lineage events (created, replaced, …)
+``pfs.code_family``         derived family membership with roles
+
+Writers are delete-then-insert per code (events, elements) or full
+replace (families) so a re-run is idempotent. Callers own the
+connection (``conf.connect.duckdb_batch`` for writes) and republish the
+replica afterwards; nothing here opens a database.
+"""
+
+from __future__ import annotations
+
+from dataclasses import astuple, dataclass
+from typing import Any, Sequence
+
+
+@dataclass(frozen=True)
+class ElementRow:
+    code: str
+    year: int
+    type: str
+    value: str
+    detail: str
+    text: str
+    item_key: str
+    p_id: int
+    page: int
+    source: str  # "fr" | "hcpcs" | "rvu"
+
+
+@dataclass(frozen=True)
+class ReviewRow:
+    code: str
+    text: str
+    proposed_type: str
+    proposed_value: str
+    item_key: str
+    p_id: int
+
+
+@dataclass(frozen=True)
+class EventRow:
+    code: str
+    year: int
+    kind: str
+    from_codes: str  # space-joined
+    to_codes: str
+    item_key: str
+    p_id: int
+    page: int
+    source: str  # "fr" | "rvu"
+    anchored: bool
+    note: str
+
+
+@dataclass(frozen=True)
+class FamilyRow:
+    key: str
+    name: str
+    code: str
+    role: str  # base | add-on | predecessor | successor | bundle | member
+    since: int | None
+    until: int | None
+    item_key: str
+    p_id: int
+
+
+_DDL = """
+CREATE SCHEMA IF NOT EXISTS pfs;
+CREATE TABLE IF NOT EXISTS pfs.code_element (
+    code VARCHAR, year INTEGER, type VARCHAR, value VARCHAR, detail VARCHAR,
+    text VARCHAR, item_key VARCHAR, p_id INTEGER, page INTEGER, source VARCHAR);
+CREATE TABLE IF NOT EXISTS pfs.code_element_review (
+    code VARCHAR, text VARCHAR, proposed_type VARCHAR, proposed_value VARCHAR,
+    item_key VARCHAR, p_id INTEGER);
+CREATE TABLE IF NOT EXISTS pfs.code_event (
+    code VARCHAR, year INTEGER, kind VARCHAR, from_codes VARCHAR, to_codes VARCHAR,
+    item_key VARCHAR, p_id INTEGER, page INTEGER, source VARCHAR, anchored BOOLEAN, note VARCHAR);
+CREATE TABLE IF NOT EXISTS pfs.code_family (
+    key VARCHAR, name VARCHAR, code VARCHAR, role VARCHAR, since INTEGER, until INTEGER,
+    item_key VARCHAR, p_id INTEGER);
+"""
+
+
+def ensure_tables(con: Any) -> None:
+    for stmt in _DDL.strip().split(";"):
+        if stmt.strip():
+            con.execute(stmt)
+
+
+def _insert(con: Any, table: str, rows: Sequence[Any]) -> int:
+    if not rows:
+        return 0
+    width = len(astuple(rows[0]))
+    marks = ",".join("?" * width)
+    con.executemany(f"INSERT INTO {table} VALUES ({marks})", [astuple(r) for r in rows])
+    return len(rows)
+
+
+def write_elements(
+    con: Any, code: str, rows: Sequence[ElementRow], reviews: Sequence[ReviewRow]
+) -> int:
+    ensure_tables(con)
+    con.execute("DELETE FROM pfs.code_element WHERE code = ?", [code])
+    con.execute("DELETE FROM pfs.code_element_review WHERE code = ?", [code])
+    _insert(con, "pfs.code_element_review", reviews)
+    return _insert(con, "pfs.code_element", rows)
+
+
+def write_events(con: Any, code: str, rows: Sequence[EventRow]) -> int:
+    ensure_tables(con)
+    con.execute("DELETE FROM pfs.code_event WHERE code = ?", [code])
+    return _insert(con, "pfs.code_event", rows)
+
+
+def write_families(con: Any, rows: Sequence[FamilyRow]) -> int:
+    ensure_tables(con)
+    con.execute("DELETE FROM pfs.code_family")
+    return _insert(con, "pfs.code_family", rows)
+
+
+def read_elements(con: Any, code: str) -> list[ElementRow]:
+    rows = con.execute(
+        "SELECT * FROM pfs.code_element WHERE code = ? ORDER BY year, type, value, detail",
+        [code],
+    ).fetchall()
+    return [ElementRow(*r) for r in rows]
+
+
+def read_events(con: Any, code: str) -> list[EventRow]:
+    rows = con.execute(
+        "SELECT * FROM pfs.code_event WHERE code = ? ORDER BY year, kind", [code]
+    ).fetchall()
+    return [EventRow(*r) for r in rows]
+
+
+def read_families(con: Any) -> list[FamilyRow]:
+    rows = con.execute(
+        "SELECT * FROM pfs.code_family ORDER BY key, role, code"
+    ).fetchall()
+    return [FamilyRow(*r) for r in rows]

tests/pfs/test_codetables.py

@@ -0,0 +1,133 @@
+"""pfs.codetables — DDL + writers for the code element/event/family tables."""
+
+from __future__ import annotations
+
+import duckdb
+import pytest
+
+from pfs.codetables import (
+    ElementRow,
+    EventRow,
+    FamilyRow,
+    ReviewRow,
+    ensure_tables,
+    read_elements,
+    read_events,
+    read_families,
+    write_elements,
+    write_events,
+    write_families,
+)
+
+
+@pytest.fixture
+def con():
+    c = duckdb.connect(":memory:")
+    ensure_tables(c)
+    yield c
+    c.close()
+
+
+def _el(code="99490", value="consent", year=2015):
+    return ElementRow(
+        code, year, "activity", value, "", "Consent;", "DE2VH9PD", 1245, 67716, "fr"
+    )
+
+
+class TestDDL:
+    def test_idempotent(self, con):
+        ensure_tables(con)
+        names = {
+            r[0]
+            for r in con.execute(
+                "SELECT table_name FROM information_schema.tables WHERE table_schema='pfs'"
+            ).fetchall()
+        }
+        assert {
+            "code_element",
+            "code_element_review",
+            "code_event",
+            "code_family",
+        } <= names
+
+
+class TestElements:
+    def test_write_replaces_per_code(self, con):
+        assert (
+            write_elements(
+                con,
+                "99490",
+                [_el(), _el(value="24-7-access")],
+                [ReviewRow("99490", "odd line", "activity", "", "DE2VH9PD", 1246)],
+            )
+            == 2
+        )
+        assert write_elements(con, "99490", [_el()], []) == 1
+        rows = read_elements(con, "99490")
+        assert [r.value for r in rows] == ["consent"]
+        assert (
+            con.execute(
+                "SELECT count(*) FROM pfs.code_element_review WHERE code='99490'"
+            ).fetchone()[0]
+            == 0
+        )
+
+    def test_other_codes_untouched(self, con):
+        write_elements(con, "99490", [_el()], [])
+        write_elements(con, "G0556", [_el(code="G0556")], [])
+        write_elements(con, "99490", [], [])
+        assert read_elements(con, "99490") == []
+        assert len(read_elements(con, "G0556")) == 1
+
+
+class TestEvents:
+    def test_roundtrip_sorted_by_year(self, con):
+        rows = [
+            EventRow(
+                "99439",
+                2021,
+                "replaces",
+                "G2058",
+                "99439",
+                "YBM4IZUS",
+                1578,
+                84639,
+                "fr",
+                True,
+                "",
+            ),
+            EventRow("99439", 2021, "appeared", "", "", "", 0, 0, "rvu", True, ""),
+        ]
+        assert write_events(con, "99439", rows) == 2
+        got = read_events(con, "99439")
+        assert [e.kind for e in got] == ["appeared", "replaces"]
+        assert got[1].anchored is True
+
+
+class TestFamilies:
+    def test_full_replace(self, con):
+        write_families(
+            con,
+            [
+                FamilyRow(
+                    "CCM", "Chronic Care Management", "99490", "base", 2015, None, "", 0
+                )
+            ],
+        )
+        write_families(
+            con,
+            [
+                FamilyRow(
+                    "CCM",
+                    "Chronic Care Management",
+                    "99439",
+                    "add-on",
+                    2021,
+                    None,
+                    "YBM4IZUS",
+                    1578,
+                )
+            ],
+        )
+        fams = read_families(con)
+        assert len(fams) == 1 and fams[0].code == "99439" and fams[0].role == "add-on"