homelab/stack
1
0
Fork 0
Code Issues 111 Pull Requests 1 Actions Packages Projects 1 Releases Wiki Activity

fix(pfs): CPT lineage/extraction sources tolerate a replica without pfs.cpt_* tables (refs #685 #686)

Browse Source
This commit is contained in:
kert
2026-09-09 20:19:07 -04:00
parent c30173c5b4
commit b33fd3e150
6 changed files with 146 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/cli/pfs.py
View File

@@ -23,6 +23,7 @@ import typer
from pfs.codetables import (
cpt_years,
ensure_tables,
is_missing_table_error,
read_all_elements,
read_all_events,
read_cpt_codes,
@@ -45,10 +46,12 @@ app = typer.Typer(no_args_is_help=True)
#: A replica that hasn't had `elements`/`lineage --write` (or an initial
#: `families --write`) run yet is missing `pfs.code_element`/
#: `pfs.code_event`/`pfs.code_family` — not a bug, just nothing derived
#: yet (I4). DuckDB's "table does not exist" errors are Catalog errors.
#: yet (I4). DuckDB's "table does not exist" errors are Catalog errors —
#: the predicate itself lives in `pfs.codetables` so `pfs.lineage`/
#: `pfs.extract` can reuse it for the same replica shape (a CPT-table-
#: free replica is not a bug either).
def _missing_table(exc: Exception) -> bool:
msg = str(exc)
return "Catalog" in msg or "does not exist" in msg
return is_missing_table_error(exc)
# ── indirections the tests monkeypatch ───────────────────────────────

13
src/pfs/codetables.py
View File

@@ -216,6 +216,19 @@ CREATE TABLE IF NOT EXISTS pfs.cpt_code_alt (
"""
def is_missing_table_error(exc: Exception) -> bool:
"""True for a DuckDB "table/schema does not exist" Catalog error — a
replica that hasn't had ``cpt-ingest``/``elements``/``lineage
--write``/``families --write`` run yet is missing
``pfs.cpt_*``/``pfs.code_element``/``pfs.code_event``/
``pfs.code_family``, which is not a bug, just nothing derived or
ingested yet (I4). Shared by ``src/cli/pfs.py``, ``pfs.lineage`` and
``pfs.extract`` so every read-only CPT-table lookup treats a missing
table the same way."""
msg = str(exc)
return "Catalog" in msg or "does not exist" in msg
def ensure_tables(con: Any) -> None:
for stmt in _DDL.strip().split(";"):
if stmt.strip():

77
src/pfs/extract.py
View File

@@ -16,7 +16,7 @@ from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Callable, Sequence
from pfs.codetables import ElementRow, ReviewRow
from pfs.codetables import ElementRow, ReviewRow, is_missing_table_error
from pfs.descriptors import (
DescriptorRun,
Para,
@@ -92,6 +92,14 @@ def extract_text(code: str, text: str, *, year: int, source: str) -> Extraction:
return Extraction(code, tuple(rows), ())
def _cpt_row(
code: str, edition_year: int, item_key: str, e: Element, text: str
) -> ElementRow:
return ElementRow(
code, edition_year, e.type.value, e.value, e.detail, text, item_key, 0, 0, "cpt"
)
def _cpt_elements(
con: Any, code: str, *, classify: Classifier | None = None
) -> tuple[list[ElementRow], list[ReviewRow]]:
@@ -102,12 +110,20 @@ def _cpt_elements(
to *classify* if given, else to the review queue. The ``parent``
code's stem/elements already apply to add-on rows (the CPT parser
inherits them onto the child's own ``pfs.cpt_code`` row), so nothing
extra is done here for add-ons."""
row = con.execute(
"SELECT edition_year, item_key, stem, elements, tail FROM pfs.cpt_code "
"WHERE code = ? ORDER BY edition_year DESC LIMIT 1",
[code],
).fetchone()
extra is done here for add-ons. A replica with no ``pfs.cpt_code``
table yet (CPT ingestion hasn't run) is not a bug — it just yields no
CPT rows/reviews (I4, shared with the CLI's read-only paths via
``is_missing_table_error``)."""
try:
row = con.execute(
"SELECT edition_year, item_key, stem, elements, tail FROM pfs.cpt_code "
"WHERE code = ? ORDER BY edition_year DESC LIMIT 1",
[code],
).fetchone()
except Exception as exc:
if is_missing_table_error(exc):
return [], []
raise
if row is None:
return [], []
edition_year, item_key, stem, elements, tail = row
@@ -115,20 +131,7 @@ def _cpt_elements(
tail_text = (tail or "").strip()
rows: list[ElementRow] = []
for e in parse_descriptor(f"{stem_text} {tail_text}".strip()):
rows.append(
ElementRow(
code,
edition_year,
e.type.value,
e.value,
e.detail,
stem_text,
item_key,
0,
0,
"cpt",
)
)
rows.append(_cpt_row(code, edition_year, item_key, e, stem_text))
reviews: list[ReviewRow] = []
for raw in elements or ():
text = raw.strip()
@@ -136,41 +139,15 @@ def _cpt_elements(
text = text[:-1].strip()
if not text:
continue
found = parse_descriptor(raw)
found = parse_descriptor(text)
if found:
for e in found:
rows.append(
ElementRow(
code,
edition_year,
e.type.value,
e.value,
e.detail,
text,
item_key,
0,
0,
"cpt",
)
)
rows.append(_cpt_row(code, edition_year, item_key, e, text))
continue
choice = classify(text, _CHOICES) if classify else None
if choice is not None and choice in _TYPE_OF:
e = Element(_TYPE_OF[choice], choice)
rows.append(
ElementRow(
code,
edition_year,
e.type.value,
e.value,
e.detail,
text,
item_key,
0,
0,
"cpt",
)
)
rows.append(_cpt_row(code, edition_year, item_key, e, text))
else:
reviews.append(ReviewRow(code, text, "", "", item_key, 0))
return rows, reviews

76
src/pfs/lineage.py
View File

@@ -25,7 +25,7 @@ import dataclasses
import re
from typing import Any
from pfs.codetables import EventRow, cpt_years
from pfs.codetables import EventRow, cpt_years, is_missing_table_error
from pfs.descriptors import rule_year_of
from pfs.families import FR_CITE_RE, find_codes
@@ -265,59 +265,53 @@ def fr_events(store: Any, code: str) -> list[EventRow]:
return sorted(out, key=lambda e: (e.year, FR_KINDS.index(e.kind), e.p_id))
def _cpt_ev(code: str, year: int, kind: str, item_key: str, note: str) -> EventRow:
return EventRow(code, year, kind, "", "", item_key, 0, 0, "cpt", True, note)
def cpt_events(con: Any, code: str) -> list[EventRow]:
"""CPT codebook lineage for *code*: one ``cpt_changed`` event per year
the newest ingested edition's ``cpt-changes`` reference row names
(``cpt_present`` is deliberately not an event — too noisy), or a
single ``cpt_deleted`` event when *code* is present in an older
edition's ``pfs.cpt_code`` but absent from the newest one."""
edition's ``pfs.cpt_code`` but absent from the newest one. A replica
with no ``pfs.cpt_*`` tables yet (CPT ingestion hasn't run) is not a
bug — it just yields no CPT events (I4, shared with the CLI's
read-only paths via ``is_missing_table_error``)."""
code = code.upper()
years = cpt_years(con)
if not years:
return []
newest = max(years)
ref = con.execute(
"SELECT item_key, years FROM pfs.cpt_reference "
"WHERE edition_year = ? AND code = ? AND kind = 'cpt-changes'",
[newest, code],
).fetchone()
if ref is not None:
item_key, yrs = ref
return [
EventRow(
code,
int(y),
"cpt_changed",
"",
"",
item_key,
0,
0,
"cpt",
True,
f"CPT Changes {int(y)}",
)
for y in sorted({int(y) for y in (yrs or [])})
]
present = con.execute(
"SELECT edition_year, item_key FROM pfs.cpt_code WHERE code = ? "
"ORDER BY edition_year",
[code],
).fetchall()
try:
years = cpt_years(con)
if not years:
return []
newest = max(years)
ref = con.execute(
"SELECT item_key, years FROM pfs.cpt_reference "
"WHERE edition_year = ? AND code = ? AND kind = 'cpt-changes'",
[newest, code],
).fetchone()
if ref is not None:
item_key, yrs = ref
return [
_cpt_ev(code, int(y), "cpt_changed", item_key, f"CPT Changes {int(y)}")
for y in sorted({int(y) for y in (yrs or [])})
]
present = con.execute(
"SELECT edition_year, item_key FROM pfs.cpt_code WHERE code = ? "
"ORDER BY edition_year",
[code],
).fetchall()
except Exception as exc:
if is_missing_table_error(exc):
return []
raise
if present and int(present[-1][0]) < newest:
last_year, last_key = present[-1]
return [
EventRow(
_cpt_ev(
code,
int(last_year) + 1,
"cpt_deleted",
"",
"",
last_key,
0,
0,
"cpt",
True,
f"absent from CPT {newest}",
)
]

31
tests/pfs/test_extract.py
View File

@@ -194,6 +194,24 @@ def _duckdb_con():
return c
def _duckdb_con_no_cpt():
"""A replica shape that hasn't been through `cpt-ingest` yet: `pfs.rvu`
and `terminology.hcpcs_level_2` exist, but no `pfs.cpt_*` table does —
the read-only path deliberately never calls `ensure_tables` (I4)."""
c = duckdb.connect(":memory:")
c.execute("CREATE SCHEMA pfs")
c.execute(
"CREATE TABLE pfs.rvu (hcpcs VARCHAR, mod VARCHAR, description VARCHAR, "
"status_code VARCHAR, non_fac_total DOUBLE, year INTEGER)"
)
c.execute("CREATE SCHEMA terminology")
c.execute(
"CREATE TABLE terminology.hcpcs_level_2 (hcpcs VARCHAR, long_description VARCHAR, "
"seqnum VARCHAR, recid VARCHAR)"
)
return c
def _insert_cpt_code(con, year, item_key, code, *, stem, elements, tail):
con.execute(
"INSERT INTO pfs.cpt_code VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
@@ -268,6 +286,19 @@ class TestExtractCodeCpt:
assert not any(r.source == "cpt" for r in x.rows)
assert x.reviews == ()
def test_replica_with_no_cpt_tables_at_all_does_not_raise(self, store):
# #685/#686 review finding: pfs.cpt_code may not exist yet (an
# older replica, or one that hasn't been through cpt-ingest) —
# the read-only `elements --dry-run` path never calls
# ensure_tables, so this must not raise CatalogException.
con = _duckdb_con_no_cpt()
try:
x = extract_code(store, con, "99490")
finally:
con.close()
assert not any(r.source == "cpt" for r in x.rows)
assert x.reviews == ()
def test_fr_wins_over_cpt_on_duplicate(self, con):
store_con = _sqlite_store_con()
try:

34
tests/pfs/test_lineage.py
View File

@@ -299,8 +299,42 @@ class TestCptEvents:
def test_no_cpt_tables_yet_is_empty(self, con):
assert cpt_events(con, "99490") == []
def test_no_cpt_tables_at_all_does_not_raise(self):
# #685/#686 review finding: a replica that has never had
# cpt-ingest run has no pfs.cpt_* schema at all (not merely empty
# tables) — pfs.cpt_section itself doesn't exist, so cpt_years()
# raises CatalogException. cpt_events must swallow that (I4), the
# same way the CLI's read-only paths already tolerate a missing
# pfs.code_element/pfs.code_event.
c = duckdb.connect(":memory:")
c.execute("CREATE SCHEMA pfs")
c.execute(f"CREATE TABLE pfs.rvu ({RVU_COLS})")
try:
assert cpt_events(c, "99490") == []
finally:
c.close()
class TestCrossCheck:
def test_lineage_with_no_cpt_tables_at_all_does_not_raise(self, store):
c = duckdb.connect(":memory:")
c.execute("CREATE SCHEMA pfs")
c.execute(f"CREATE TABLE pfs.rvu ({RVU_COLS})")
c.executemany(
"INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
[
("00000", None, "filler", "A", 1.0, 2019),
("00000", None, "filler", "A", 1.0, 2021),
("G2058", None, "Ccm add 20min", "A", 1.0, 2020),
],
)
try:
ev = lineage(c, store, "G2058")
finally:
c.close()
assert not any(e.source == "cpt" for e in ev)
assert any(e.kind == "appeared" for e in ev)
def test_rvu_event_anchored_by_cpt_changed_within_one_year(self, con, store):
# 99487 has no FR paragraph in `store`, but a cpt_changed at 2016
# (Y-1) must still anchor the 2017 status_change.