af9c088b84
Some checks failed
CI / test (pull_request) Failing after 2m21s
Infra CI / zotero (pull_request) Successful in 17s
Infra CI / docs (pull_request) Successful in 14s
Infra CI / notebooks (pull_request) Successful in 59s
Infra CI / api (pull_request) Successful in 20s
Infra CI / mc (pull_request) Failing after 35s
Infra CI / bcda (pull_request) Successful in 16s
Infra CI / llm (pull_request) Successful in 15s
CI / lint (push) Successful in 39s
CI / test (push) Failing after 2m52s
Infra CI / notebooks (push) Successful in 1m3s
Infra CI / bcda (push) Successful in 1m46s
Infra CI / mc (push) Failing after 46s
CI / notebooks-smoke (push) Successful in 1m45s
Infra CI / zotero (push) Successful in 16s
Infra CI / api (push) Successful in 1m9s
Infra CI / docs (push) Successful in 1m44s
Infra CI / llm (push) Successful in 52s
CI / lint (pull_request) Successful in 35s
CI / notebooks-smoke (pull_request) Successful in 1m36s
A job recorded before manifests existed has no transactionTime. The first version fell back to the instant the `complete` row was written and called it "safe, slightly overlapping" — I had that backwards: a _since later than the server's true cut-off is a gap, not an overlap. The real store showed it: the sandbox stamps transactionTime as midnight UTC of the run day, so a baseline that finished at 16:53 was being credited with a 16:53 cut-off and every same-day delta (00:00) was read as predating it. The fallback is now the job's first ledger row (the request instant), which can only overlap; Store.job_started_at() reads it. `stack bcda resume <key>` still replaces the fallback with the server's value — done for the four pre-manifest jobs in the real store, whose watermarks now read 2026-10-25T00:00:00Z. refs #1139
1100 lines
41 KiB
Python
1100 lines
41 KiB
Python
"""Incremental BCDA pulls: choose ``_since``, refuse the wrong one, reconcile the result.
|
||
|
||
Design: ``docs/superpowers/specs/2026-10-25-bcda-since-reconciliation-design.md``.
|
||
|
||
The API's contract (bcda.cms.gov, vendored ``bcda-app``): ``_since``
|
||
filters on ``meta.lastUpdated``; ``/Group/all`` deltas also carry *all*
|
||
resources of newly attributed enrollees (backfill); the next ``_since``
|
||
is the previous job's ``transactionTime``; each claim *version* has a
|
||
new FHIR id, so claims are tracked by control number; nothing in a delta
|
||
can express deletion. Three layers follow from that:
|
||
|
||
1. **First order** — which row is current per key
|
||
(:data:`KEY_EXPR`, :func:`_rank_order`): newest ``last_updated``,
|
||
then NCH/DDPS over SharedSystem, then later ``transactionTime``,
|
||
later applied sequence, larger FHIR id. Candidates are the newest
|
||
applied baseline of a type plus the deltas at or after it.
|
||
2. **Second order** — what a job did to the store: ``new`` /
|
||
``replaced`` / ``unchanged`` / ``stale_ignored`` / ``backfill`` /
|
||
``late`` / ``tombstoned`` / ``adjusted`` / ``retired`` / orphans,
|
||
written to the ledger per job and table.
|
||
3. **Third order** — invariants over the whole store after every
|
||
apply (:func:`_invariants`): unique keys, count conservation,
|
||
monotone watermarks, item↔header integrity; replay determinism and
|
||
idempotency are properties of the recompute and are tested.
|
||
|
||
Temporality checks for a proposed ``_since`` live in :func:`check_since`;
|
||
:func:`next_since` formats the watermark. Current tables land where the
|
||
loader already reads them (``flat/<table>.parquet``); per-job partitions
|
||
are kept under ``flat/history/<job_key>/`` and the ledger in
|
||
``flat/sync_ledger.parquet``. Reconciliation runs DuckDB over local
|
||
Parquet, so this module expects a local store root.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
from dataclasses import asdict, dataclass, field
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import TYPE_CHECKING, Any
|
||
|
||
import pyarrow as pa
|
||
import pyarrow.parquet as pq
|
||
|
||
if TYPE_CHECKING:
|
||
from bcda.store import Store
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
# No v3 resource has lastUpdated before this (How to Filter Claims Data);
|
||
# an earlier _since is a full pull in disguise.
|
||
V3_FLOOR = datetime(2026, 3, 26, tzinfo=timezone.utc)
|
||
# Past this, the baseline's files are long gone from the server and only a
|
||
# fresh baseline can retire rows.
|
||
BASELINE_MAX_AGE = timedelta(days=30)
|
||
RESOURCE_TYPES = ("Patient", "Coverage", "ExplanationOfBenefit")
|
||
TABLE_OF = {
|
||
"Patient": "patient",
|
||
"Coverage": "coverage",
|
||
"ExplanationOfBenefit": "explanation_of_benefit",
|
||
}
|
||
TYPE_OF = {v: k for k, v in TABLE_OF.items()}
|
||
# Identity per table. A claim version gets a new FHIR id; the control
|
||
# number is the claim. Patient.id is BENE_SK; Coverage ids are composite.
|
||
KEY_EXPR = {
|
||
"patient": "coalesce(bene_id, id)",
|
||
"coverage": "id",
|
||
"explanation_of_benefit": "coalesce(clm_cntl_num, clm_id, id)",
|
||
}
|
||
# Beneficiary of a row, for backfill/orphan classification.
|
||
BENE_EXPR = {
|
||
"patient": "coalesce(bene_id, id)",
|
||
"coverage": "regexp_replace(beneficiary_reference, '^Patient/', '')",
|
||
"explanation_of_benefit": "bene_id",
|
||
}
|
||
COUNT_FIELDS = (
|
||
"rows",
|
||
"new",
|
||
"replaced",
|
||
"unchanged",
|
||
"stale_ignored",
|
||
"backfill",
|
||
"late",
|
||
"tombstoned",
|
||
"adjusted",
|
||
"retired",
|
||
"orphans",
|
||
"current",
|
||
)
|
||
|
||
_INSTANT = re.compile(
|
||
r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?(Z|[+-]\d{2}:\d{2})$"
|
||
)
|
||
|
||
|
||
class SyncError(Exception):
|
||
"""A reconciliation could not proceed."""
|
||
|
||
|
||
class SyncInvariantError(SyncError):
|
||
"""A third-order invariant failed; current tables were left untouched."""
|
||
|
||
|
||
# ── FHIR instants ─────────────────────────────────────────────
|
||
|
||
|
||
def parse_instant(value: str) -> datetime:
|
||
"""A FHIR instant (date, time *and* offset) as an aware datetime.
|
||
|
||
Raises ``ValueError`` for anything the server would 400 on
|
||
(``validateSinceParameter`` parses RFC 3339 and requires the offset).
|
||
"""
|
||
text = (value or "").strip()
|
||
if not _INSTANT.match(text):
|
||
raise ValueError(
|
||
f"{value!r} is not a FHIR instant (YYYY-MM-DDThh:mm:ss[.sss]±hh:mm or Z)"
|
||
)
|
||
return datetime.fromisoformat(text.replace("Z", "+00:00"))
|
||
|
||
|
||
def format_instant(dt: datetime) -> str:
|
||
"""UTC FHIR instant with milliseconds, the shape BCDA documents."""
|
||
if dt.tzinfo is None:
|
||
raise ValueError("naive datetime; an instant needs an offset")
|
||
u = dt.astimezone(timezone.utc)
|
||
return u.strftime("%Y-%m-%dT%H:%M:%S.") + f"{u.microsecond // 1000:03d}+00:00"
|
||
|
||
|
||
def _maybe_instant(value: Any) -> datetime | None:
|
||
if value in (None, ""):
|
||
return None
|
||
if isinstance(value, datetime):
|
||
return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
|
||
try:
|
||
return parse_instant(str(value))
|
||
except ValueError:
|
||
try:
|
||
dt = datetime.fromisoformat(str(value))
|
||
except ValueError:
|
||
return None
|
||
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
|
||
|
||
|
||
# ── Jobs, scopes, watermarks ──────────────────────────────────
|
||
|
||
|
||
def split_types(types_str: str) -> tuple[list[str], str]:
|
||
"""``"Patient,Coverage|_typeFilter=…"`` → (types, type_filter)."""
|
||
types_part, _, filt = (types_str or "").partition("|_typeFilter=")
|
||
types = [t for t in types_part.split(",") if t]
|
||
return (types or list(RESOURCE_TYPES)), filt
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class JobInfo:
|
||
key: str
|
||
endpoint: str
|
||
types: tuple[str, ...]
|
||
type_filter: str
|
||
since: str
|
||
transaction_time: datetime
|
||
transaction_source: str # job_row | manifest | started_at
|
||
ts: datetime
|
||
|
||
@property
|
||
def kind(self) -> str:
|
||
return "delta" if self.since else "baseline"
|
||
|
||
@property
|
||
def population(self) -> str:
|
||
"""``runout`` (last year's enrollees) or ``current`` (``Group/all`` and ``Patient``)."""
|
||
return population_of(self.endpoint)
|
||
|
||
@property
|
||
def scope(self) -> tuple[str, str]:
|
||
return (self.population, self.type_filter)
|
||
|
||
|
||
def population_of(endpoint: str) -> str:
|
||
return "runout" if endpoint == "Group/runout" else "current"
|
||
|
||
|
||
def job_info(store: Store, job: dict) -> JobInfo:
|
||
"""A complete job row as :class:`JobInfo`, resolving its cut-off instant.
|
||
|
||
``transactionTime`` comes from the job row (new ledgers), else the
|
||
saved manifest, else — for jobs recorded before either existed — the
|
||
instant the job was *started* (its first ledger row). The server's
|
||
snapshot cannot be newer than the moment it began processing the
|
||
request, so a ``_since`` at the start overlaps and never gaps; the
|
||
completion instant would be *later* than the true cut-off and could
|
||
skip updates made while the job ran. ``Store.resume`` replaces the
|
||
fallback with the server's value whenever the job is still on the
|
||
server.
|
||
"""
|
||
types, filt = split_types(job.get("types") or "")
|
||
tt = _maybe_instant(job.get("transaction_time"))
|
||
source = "job_row"
|
||
if tt is None:
|
||
tt = _maybe_instant(store.manifest(job["key"]).get("transactionTime"))
|
||
source = "manifest"
|
||
if tt is None:
|
||
tt = store.job_started_at(job["key"]) or _maybe_instant(job.get("ts"))
|
||
source = "started_at"
|
||
return JobInfo(
|
||
key=job["key"],
|
||
endpoint=job.get("endpoint") or "Group/all",
|
||
types=tuple(types),
|
||
type_filter=filt,
|
||
since=job.get("since") or "",
|
||
transaction_time=tt,
|
||
transaction_source=source,
|
||
ts=_maybe_instant(job.get("ts")) or tt,
|
||
)
|
||
|
||
|
||
def complete_jobs(store: Store) -> list[JobInfo]:
|
||
"""Complete jobs whose files all downloaded, oldest cut-off first.
|
||
|
||
A delta with nothing to deliver is still a job: the server answers
|
||
200 with an empty ``output`` list, so it has no file rows but a
|
||
manifest saying so, and it must advance the watermark. A complete
|
||
row with neither files nor manifest is indistinguishable from an
|
||
interrupted download and is left out (``Store.resume`` fills in the
|
||
manifest).
|
||
"""
|
||
out: list[JobInfo] = []
|
||
for job in store.list_jobs(status="complete"):
|
||
files = store.get_files(job["key"])
|
||
if files:
|
||
if any(f["status"] != "complete" for f in files):
|
||
continue
|
||
elif "output" not in store.manifest(job["key"]):
|
||
continue
|
||
out.append(job_info(store, job))
|
||
out.sort(key=lambda j: (j.transaction_time, j.ts, j.key))
|
||
return out
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Watermark:
|
||
job_key: str
|
||
transaction_time: datetime
|
||
kind: str
|
||
since: str
|
||
|
||
|
||
def watermarks(
|
||
jobs: list[JobInfo], *, endpoint: str | None = None, type_filter: str | None = None
|
||
) -> dict[tuple[str, str, str], Watermark]:
|
||
"""``{(population, type_filter, resource_type): newest cut-off}`` over *jobs*.
|
||
|
||
``Group/all`` and ``Patient`` describe the same population (currently
|
||
attributed enrollees) and share watermarks; ``Group/runout`` is last
|
||
year's enrollees and is kept apart. A ``_typeFilter`` narrows which
|
||
claims were observed, so each filter keeps its own watermark.
|
||
"""
|
||
out: dict[tuple[str, str, str], Watermark] = {}
|
||
pop = population_of(endpoint) if endpoint is not None else None
|
||
for j in jobs:
|
||
if pop is not None and j.population != pop:
|
||
continue
|
||
if type_filter is not None and j.type_filter != type_filter:
|
||
continue
|
||
for t in j.types:
|
||
k = (j.population, j.type_filter, t)
|
||
if k not in out or j.transaction_time >= out[k].transaction_time:
|
||
out[k] = Watermark(j.key, j.transaction_time, j.kind, j.since)
|
||
return out
|
||
|
||
|
||
def baselines(jobs: list[JobInfo]) -> dict[tuple[str, str, str], JobInfo]:
|
||
"""Newest baseline per ``(population, type_filter, resource_type)``."""
|
||
out: dict[tuple[str, str, str], JobInfo] = {}
|
||
for j in jobs:
|
||
if j.kind != "baseline":
|
||
continue
|
||
for t in j.types:
|
||
k = (j.population, j.type_filter, t)
|
||
if k not in out or j.transaction_time >= out[k].transaction_time:
|
||
out[k] = j
|
||
return out
|
||
|
||
|
||
def next_since(
|
||
store: Store,
|
||
*,
|
||
endpoint: str = "Group/all",
|
||
types: list[str] | None = None,
|
||
type_filter: str = "",
|
||
overlap_seconds: int = 0,
|
||
) -> dict[str, str | None]:
|
||
"""The ``_since`` to pass per resource type: the watermark, as an instant.
|
||
|
||
``None`` for a type with no complete job in this scope — run a
|
||
baseline first. *overlap_seconds* widens the window backwards;
|
||
reconciliation makes overlap free, so it only costs download time.
|
||
"""
|
||
wm = watermarks(complete_jobs(store), endpoint=endpoint, type_filter=type_filter)
|
||
out: dict[str, str | None] = {}
|
||
for t in types or RESOURCE_TYPES:
|
||
w = wm.get((population_of(endpoint), type_filter, t))
|
||
out[t] = (
|
||
format_instant(w.transaction_time - timedelta(seconds=overlap_seconds))
|
||
if w
|
||
else None
|
||
)
|
||
return out
|
||
|
||
|
||
# ── Temporality checks ────────────────────────────────────────
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Check:
|
||
level: str # ok | warn | block
|
||
code: str
|
||
message: str
|
||
resource_type: str = ""
|
||
|
||
|
||
def blocking(checks: list[Check]) -> bool:
|
||
return any(c.level == "block" for c in checks)
|
||
|
||
|
||
def check_since(
|
||
store: Store,
|
||
since: str,
|
||
*,
|
||
endpoint: str = "Group/all",
|
||
types: list[str] | None = None,
|
||
type_filter: str = "",
|
||
now: datetime | None = None,
|
||
attribution: dict | None = None,
|
||
) -> list[Check]:
|
||
"""Everything that can be known about a proposed ``_since`` before the job starts.
|
||
|
||
*attribution* is the ``attribution_status`` body when the caller has
|
||
one (``ingestion_dates`` with ``last_attribution_update``).
|
||
"""
|
||
now = now or datetime.now(tz=timezone.utc)
|
||
types = list(types or RESOURCE_TYPES)
|
||
checks: list[Check] = []
|
||
try:
|
||
dt = parse_instant(since)
|
||
except ValueError as exc:
|
||
return [Check("block", "format", str(exc))]
|
||
if dt > now:
|
||
checks.append(
|
||
Check(
|
||
"block",
|
||
"future",
|
||
f"{since} is in the future; the server rejects it (400)",
|
||
)
|
||
)
|
||
if dt < V3_FLOOR:
|
||
checks.append(
|
||
Check(
|
||
"warn",
|
||
"floor",
|
||
f"{since} predates every v3 lastUpdated ({V3_FLOOR.date()}): this is a full pull",
|
||
)
|
||
)
|
||
|
||
jobs = complete_jobs(store)
|
||
base = baselines(jobs)
|
||
wm = watermarks(jobs)
|
||
pop = population_of(endpoint)
|
||
populations = {j.population for j in jobs if j.kind == "baseline"}
|
||
if populations and pop not in populations:
|
||
checks.append(
|
||
Check(
|
||
"block",
|
||
"endpoint_mix",
|
||
f"the store's baselines describe the {', '.join(sorted(populations))} population; "
|
||
f"a {endpoint} delta ({pop}) cannot extend them",
|
||
)
|
||
)
|
||
for t in types:
|
||
b = base.get((pop, type_filter, t))
|
||
if b is None:
|
||
other_filter = [
|
||
bb
|
||
for (pp, f, tt), bb in base.items()
|
||
if pp == pop and tt == t and f != type_filter
|
||
]
|
||
if not other_filter:
|
||
checks.append(
|
||
Check(
|
||
"block",
|
||
"no_baseline",
|
||
f"{t}: no complete baseline (unfiltered pull) for the {pop} population to extend",
|
||
t,
|
||
)
|
||
)
|
||
continue
|
||
b = max(other_filter, key=lambda bb: bb.transaction_time)
|
||
checks.append(
|
||
Check(
|
||
"warn",
|
||
"scope_filter",
|
||
f"{t}: baseline {b.key} was pulled with _typeFilter={b.type_filter or '<default>'}, "
|
||
f"this delta asks for {type_filter or '<default>'}; its rows are reconciled, but claims "
|
||
"outside its filter are not refreshed and nothing retires until the next baseline",
|
||
t,
|
||
)
|
||
)
|
||
if endpoint == "Patient" and b.endpoint == "Group/all":
|
||
checks.append(
|
||
Check(
|
||
"warn",
|
||
"backfill_scope",
|
||
f"{t}: a /Patient delta over a Group/all baseline omits newly attributed enrollees' "
|
||
"history (only Group/all backfills them)",
|
||
t,
|
||
)
|
||
)
|
||
w = wm.get((pop, type_filter, t)) or max(
|
||
(ww for (pp, _f, tt), ww in wm.items() if pp == pop and tt == t),
|
||
key=lambda ww: ww.transaction_time,
|
||
)
|
||
if dt > w.transaction_time:
|
||
checks.append(
|
||
Check(
|
||
"block",
|
||
"gap",
|
||
f"{t}: {since} is after the watermark {format_instant(w.transaction_time)} (job {w.job_key}); "
|
||
"resources updated in between would never be fetched",
|
||
t,
|
||
)
|
||
)
|
||
elif dt < w.transaction_time:
|
||
checks.append(
|
||
Check(
|
||
"ok",
|
||
"overlap",
|
||
f"{t}: {since} is before the watermark {format_instant(w.transaction_time)}; "
|
||
"the overlap is re-delivered and reconciles to no change",
|
||
t,
|
||
)
|
||
)
|
||
else:
|
||
checks.append(
|
||
Check(
|
||
"ok",
|
||
"watermark",
|
||
f"{t}: _since equals the watermark (job {w.job_key})",
|
||
t,
|
||
)
|
||
)
|
||
if now - b.transaction_time > BASELINE_MAX_AGE:
|
||
checks.append(
|
||
Check(
|
||
"warn",
|
||
"stale_baseline",
|
||
f"{t}: baseline {b.key} is {(now - b.transaction_time).days} days old; "
|
||
"enrollees who left attribution since then stay until a new baseline",
|
||
t,
|
||
)
|
||
)
|
||
if attribution:
|
||
last_attr = None
|
||
for d in attribution.get("ingestion_dates", []):
|
||
if d.get("type") == "last_attribution_update":
|
||
last_attr = _maybe_instant(d.get("timestamp"))
|
||
if last_attr and last_attr > b.transaction_time:
|
||
checks.append(
|
||
Check(
|
||
"warn",
|
||
"attribution_newer",
|
||
f"{t}: attribution updated {format_instant(last_attr)}, after baseline {b.key}; "
|
||
"expect full-history backfill for new enrollees, and no signal for those who left",
|
||
t,
|
||
)
|
||
)
|
||
# in-flight and unapplied work
|
||
for job in store.list_jobs():
|
||
if job["status"] in ("pending", "polling", "downloading"):
|
||
jt, jf = split_types(job.get("types") or "")
|
||
if population_of(job.get("endpoint") or "Group/all") == pop and set(
|
||
jt
|
||
) & set(types):
|
||
checks.append(
|
||
Check(
|
||
"warn",
|
||
"inflight",
|
||
f"job {job['key']} ({job['status']}) covers {', '.join(sorted(set(jt) & set(types)))}; "
|
||
"start this one after it lands so the watermark is right",
|
||
)
|
||
)
|
||
applied = {r["job_key"] for r in read_ledger(store) if r["ok"]}
|
||
pending = [j.key for j in jobs if j.key not in applied]
|
||
if pending:
|
||
checks.append(
|
||
Check(
|
||
"warn",
|
||
"unapplied",
|
||
f"{len(pending)} downloaded job(s) not yet reconciled: {', '.join(pending[:6])}",
|
||
)
|
||
)
|
||
return checks
|
||
|
||
|
||
# ── Ledger ────────────────────────────────────────────────────
|
||
|
||
LEDGER_SCHEMA = pa.schema(
|
||
[
|
||
("applied_seq", pa.int64()),
|
||
("job_key", pa.string()),
|
||
("kind", pa.string()),
|
||
("endpoint", pa.string()),
|
||
("type_filter", pa.string()),
|
||
("types", pa.string()),
|
||
("since", pa.string()),
|
||
("transaction_time", pa.string()),
|
||
("transaction_source", pa.string()),
|
||
("applied_at", pa.string()),
|
||
("ok", pa.bool_()),
|
||
("counts", pa.string()), # json {table: {field: n}}
|
||
("notes", pa.string()), # json [str]
|
||
("error", pa.string()),
|
||
]
|
||
)
|
||
|
||
|
||
def _root(store: Store) -> Path:
|
||
return Path(str(store.root))
|
||
|
||
|
||
def _ledger_path(store: Store) -> Path:
|
||
return _root(store) / "flat" / "sync_ledger.parquet"
|
||
|
||
|
||
def read_ledger(store: Store) -> list[dict]:
|
||
p = _ledger_path(store)
|
||
if not p.exists():
|
||
return []
|
||
return pq.read_table(p, schema=LEDGER_SCHEMA).to_pylist()
|
||
|
||
|
||
def _append_ledger(store: Store, row: dict) -> None:
|
||
p = _ledger_path(store)
|
||
p.parent.mkdir(parents=True, exist_ok=True)
|
||
existing = (
|
||
pq.read_table(p, schema=LEDGER_SCHEMA)
|
||
if p.exists()
|
||
else LEDGER_SCHEMA.empty_table()
|
||
)
|
||
new = pa.Table.from_pylist([row], schema=LEDGER_SCHEMA)
|
||
tmp = p.with_suffix(".tmp")
|
||
pq.write_table(pa.concat_tables([existing, new]), tmp)
|
||
os.replace(tmp, p)
|
||
|
||
|
||
# ── Reconciliation ────────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class JobReport:
|
||
job_key: str
|
||
kind: str
|
||
endpoint: str
|
||
type_filter: str
|
||
types: list[str]
|
||
since: str
|
||
transaction_time: str
|
||
counts: dict[str, dict[str, int]] = field(default_factory=dict)
|
||
notes: list[str] = field(default_factory=list)
|
||
ok: bool = True
|
||
error: str = ""
|
||
|
||
def as_dict(self) -> dict:
|
||
return asdict(self)
|
||
|
||
|
||
@dataclass
|
||
class SyncReport:
|
||
applied: list[JobReport] = field(default_factory=list)
|
||
skipped: list[str] = field(default_factory=list)
|
||
|
||
@property
|
||
def ok(self) -> bool:
|
||
return all(r.ok for r in self.applied)
|
||
|
||
|
||
def _history_dir(store: Store, key: str) -> Path:
|
||
return _root(store) / "flat" / "history" / key
|
||
|
||
|
||
def _flatten_job(store: Store, job: JobInfo) -> dict[str, int]:
|
||
"""Flatten one job's ndjson into its history partition (idempotent)."""
|
||
from bcda.express.flatten import flatten_export
|
||
|
||
export_dir = _root(store) / "exports" / job.key
|
||
if not export_dir.is_dir():
|
||
raise SyncError(f"job {job.key}: no export directory at {export_dir}")
|
||
return flatten_export([export_dir], _root(store), out_dir=f"flat/history/{job.key}")
|
||
|
||
|
||
def _rank_order(table: str) -> str:
|
||
src = (
|
||
"case when system_type = 'SharedSystem' then 1 else 0 end"
|
||
if table == "explanation_of_benefit"
|
||
else "0"
|
||
)
|
||
return (
|
||
"try_cast(last_updated as timestamptz) desc nulls last, "
|
||
f"{src} asc, _transaction_time desc, _applied_seq desc, id desc"
|
||
)
|
||
|
||
|
||
def _candidates(
|
||
table: str, applied: list[JobInfo], seqs: dict[str, int]
|
||
) -> tuple[list[JobInfo], JobInfo | None, list[str]]:
|
||
"""Jobs whose partitions feed *table*'s current state, plus notes.
|
||
|
||
The newest applied baseline carrying the type is authoritative;
|
||
deltas of the same scope at or after its cut-off extend it. Deltas
|
||
older than the baseline, or in another scope, are left out (the
|
||
baseline already reflects them / they describe another population).
|
||
"""
|
||
rtype = TYPE_OF[table]
|
||
with_type = [j for j in applied if rtype in j.types]
|
||
base = [j for j in with_type if j.kind == "baseline"]
|
||
notes: list[str] = []
|
||
if not base:
|
||
deltas = [j for j in with_type if j.kind == "delta"]
|
||
if deltas:
|
||
notes.append(
|
||
f"{table}: {len(deltas)} delta(s) applied without a baseline; current state is partial"
|
||
)
|
||
return deltas, None, notes
|
||
auth = max(base, key=lambda j: (j.transaction_time, seqs[j.key]))
|
||
chosen = [auth]
|
||
for j in with_type:
|
||
if j.key == auth.key or j.kind != "delta":
|
||
continue
|
||
if j.population != auth.population:
|
||
notes.append(
|
||
f"{table}: delta {j.key} ignored — {j.population} population, baseline {auth.key} is {auth.population}"
|
||
)
|
||
elif j.transaction_time < auth.transaction_time:
|
||
notes.append(
|
||
f"{table}: delta {j.key} ignored — cut-off {format_instant(j.transaction_time)} predates baseline {auth.key}"
|
||
)
|
||
else:
|
||
if j.type_filter != auth.type_filter:
|
||
notes.append(
|
||
f"{table}: delta {j.key} reconciled with _typeFilter={j.type_filter or '<default>'} "
|
||
f"(baseline {auth.key}: {auth.type_filter or '<default>'}) — scope drift, rows outside "
|
||
"its filter were not refreshed"
|
||
)
|
||
chosen.append(j)
|
||
return chosen, auth, notes
|
||
|
||
|
||
def _partition(store: Store, key: str, table: str) -> Path | None:
|
||
p = _history_dir(store, key) / f"{table}.parquet"
|
||
return p if p.exists() else None
|
||
|
||
|
||
def _sql_literal(s: str) -> str:
|
||
return "'" + s.replace("'", "''") + "'"
|
||
|
||
|
||
def _union_sql(
|
||
store: Store, table: str, jobs: list[JobInfo], seqs: dict[str, int]
|
||
) -> str | None:
|
||
parts = []
|
||
for j in jobs:
|
||
p = _partition(store, j.key, table)
|
||
if p is None:
|
||
continue
|
||
parts.append(
|
||
f"select *, {_sql_literal(j.key)} as _job_key, "
|
||
f"{_sql_literal(format_instant(j.transaction_time))}::timestamptz as _transaction_time, "
|
||
f"{_sql_literal(j.since)} as _since, {_sql_literal(j.kind)} as _kind, "
|
||
f"{seqs[j.key]} as _applied_seq "
|
||
f"from read_parquet({_sql_literal(str(p))})"
|
||
)
|
||
if not parts:
|
||
return None
|
||
return " union all by name ".join(parts)
|
||
|
||
|
||
def _current_path(store: Store, table: str) -> Path:
|
||
return _root(store) / "flat" / f"{table}.parquet"
|
||
|
||
|
||
def apply(
|
||
store: Store,
|
||
keys: list[str] | None = None,
|
||
*,
|
||
force: bool = False,
|
||
jobs: list[JobInfo] | None = None,
|
||
) -> SyncReport:
|
||
"""Reconcile downloaded jobs into the current tables, one job at a time.
|
||
|
||
Unapplied complete jobs are applied oldest cut-off first (or just
|
||
*keys*). Each apply re-derives the current tables from the full set
|
||
of applied partitions, so the result is a function of that set and
|
||
not of the order; ``force`` re-applies already-applied jobs (a
|
||
no-op by construction, useful after a flattener change).
|
||
"""
|
||
import duckdb
|
||
|
||
jobs = jobs if jobs is not None else complete_jobs(store)
|
||
ledger = read_ledger(store)
|
||
applied_keys = {r["job_key"] for r in ledger if r["ok"]}
|
||
seqs: dict[str, int] = {r["job_key"]: r["applied_seq"] for r in ledger if r["ok"]}
|
||
next_seq = max((r["applied_seq"] for r in ledger), default=0) + 1
|
||
by_key = {j.key: j for j in jobs}
|
||
report = SyncReport()
|
||
if keys is not None:
|
||
unknown = [k for k in keys if k not in by_key]
|
||
if unknown:
|
||
raise SyncError(f"not complete jobs in this store: {', '.join(unknown)}")
|
||
todo = [by_key[k] for k in keys]
|
||
else:
|
||
todo = [j for j in jobs if force or j.key not in applied_keys]
|
||
todo.sort(key=lambda j: (j.transaction_time, j.ts, j.key))
|
||
for j in jobs:
|
||
if j.key in applied_keys and j not in todo:
|
||
report.skipped.append(j.key)
|
||
|
||
for job in todo:
|
||
rep = JobReport(
|
||
job_key=job.key,
|
||
kind=job.kind,
|
||
endpoint=job.endpoint,
|
||
type_filter=job.type_filter,
|
||
types=list(job.types),
|
||
since=job.since,
|
||
transaction_time=format_instant(job.transaction_time),
|
||
)
|
||
rep.notes.append(f"transactionTime from {job.transaction_source}")
|
||
con = duckdb.connect()
|
||
try:
|
||
_flatten_job(store, job)
|
||
seqs[job.key] = seqs.get(job.key, next_seq)
|
||
applied_now = [by_key[k] for k in applied_keys if k in by_key]
|
||
if job.key not in applied_keys:
|
||
applied_now.append(job)
|
||
_recompute(con, store, job, applied_now, seqs, rep)
|
||
next_seq = max(next_seq, seqs[job.key]) + 1
|
||
applied_keys.add(job.key)
|
||
except Exception as exc: # noqa: BLE001 — recorded, then re-raised
|
||
rep.ok = False
|
||
rep.error = f"{type(exc).__name__}: {exc}"
|
||
log.error("sync apply %s failed: %s", job.key, rep.error)
|
||
_append_ledger(store, _ledger_row(rep, seqs.get(job.key, next_seq), job))
|
||
report.applied.append(rep)
|
||
raise
|
||
finally:
|
||
con.close()
|
||
_append_ledger(store, _ledger_row(rep, seqs[job.key], job))
|
||
report.applied.append(rep)
|
||
log.info(
|
||
"sync applied %s (%s): %s",
|
||
job.key,
|
||
job.kind,
|
||
{t: c.get("current") for t, c in rep.counts.items()},
|
||
)
|
||
return report
|
||
|
||
|
||
def _ledger_row(rep: JobReport, seq: int, job: JobInfo) -> dict:
|
||
return {
|
||
"applied_seq": seq,
|
||
"job_key": rep.job_key,
|
||
"kind": rep.kind,
|
||
"endpoint": rep.endpoint,
|
||
"type_filter": rep.type_filter,
|
||
"types": ",".join(rep.types),
|
||
"since": rep.since,
|
||
"transaction_time": rep.transaction_time,
|
||
"transaction_source": job.transaction_source,
|
||
"applied_at": format_instant(datetime.now(tz=timezone.utc)),
|
||
"ok": rep.ok,
|
||
"counts": json.dumps(rep.counts, sort_keys=True),
|
||
"notes": json.dumps(rep.notes),
|
||
"error": rep.error,
|
||
}
|
||
|
||
|
||
def _recompute(
|
||
con: Any,
|
||
store: Store,
|
||
job: JobInfo,
|
||
applied: list[JobInfo],
|
||
seqs: dict[str, int],
|
||
rep: JobReport,
|
||
) -> None:
|
||
"""Derive every current table from the applied partitions and classify *job*'s effect."""
|
||
staged: dict[str, Path] = {}
|
||
prev_counts: dict[str, int] = {}
|
||
prev_patients: str | None = None
|
||
|
||
# previous current tables (for classification and conservation)
|
||
for table in TABLE_OF.values():
|
||
p = _current_path(store, table)
|
||
if p.exists():
|
||
con.execute(
|
||
f"create table prev_{table} as select {KEY_EXPR[table]} as _key, id, last_updated, "
|
||
f"{BENE_EXPR[table]} as _bene from read_parquet({_sql_literal(str(p))})"
|
||
)
|
||
else:
|
||
con.execute(
|
||
f"create table prev_{table} (_key varchar, id varchar, last_updated varchar, _bene varchar)"
|
||
)
|
||
prev_counts[table] = con.execute(
|
||
f"select count(*) from prev_{table}"
|
||
).fetchone()[0]
|
||
prev_patients = "select _bene from prev_patient"
|
||
|
||
for table in TABLE_OF.values():
|
||
rtype = TYPE_OF[table]
|
||
cand_jobs, auth, notes = _candidates(table, applied, seqs)
|
||
rep.notes.extend(notes)
|
||
sql = _union_sql(store, table, cand_jobs, seqs)
|
||
counts = dict.fromkeys(COUNT_FIELDS, 0)
|
||
if sql is None:
|
||
# nothing applied carries this type; keep whatever is current
|
||
if rtype in job.types:
|
||
rep.counts[table] = counts
|
||
continue
|
||
con.execute(f"create or replace table cand_{table} as {sql}")
|
||
con.execute(
|
||
f"create or replace table cur_{table} as "
|
||
f"select * exclude (_rn) from (select *, row_number() over "
|
||
f"(partition by {KEY_EXPR[table]} order by {_rank_order(table)}) as _rn "
|
||
f"from cand_{table}) where _rn = 1"
|
||
)
|
||
staged[table] = _write_current(con, store, table)
|
||
counts["current"] = con.execute(f"select count(*) from cur_{table}").fetchone()[
|
||
0
|
||
]
|
||
|
||
if rtype in job.types and _partition(store, job.key, table) is not None:
|
||
if any(j.key == job.key for j in cand_jobs):
|
||
_classify(con, store, job, table, counts, prev_patients)
|
||
if job.kind == "baseline":
|
||
counts["retired"] = _retired(con, table)
|
||
else:
|
||
# Not a candidate (older than the authoritative baseline, or
|
||
# another scope): it changes nothing, every row is stale.
|
||
counts["rows"] = counts["stale_ignored"] = con.execute(
|
||
f"select count(*) from read_parquet({_sql_literal(str(_partition(store, job.key, table)))})"
|
||
).fetchone()[0]
|
||
if rtype in job.types:
|
||
rep.counts[table] = counts
|
||
|
||
# eob_item follows its header: the winning header's lines from the same job
|
||
if "explanation_of_benefit" in staged:
|
||
eob_jobs, _, _ = _candidates("explanation_of_benefit", applied, seqs)
|
||
isql = _union_sql(store, "eob_item", eob_jobs, seqs)
|
||
if isql:
|
||
con.execute(f"create or replace table cand_eob_item as {isql}")
|
||
con.execute(
|
||
"create or replace table cur_eob_item as select i.* from cand_eob_item i "
|
||
"join cur_explanation_of_benefit h on i.clm_id = h.clm_id and i._job_key = h._job_key"
|
||
)
|
||
staged["eob_item"] = _write_current(con, store, "eob_item")
|
||
con.execute(
|
||
"create or replace table cur_eob_supersession as "
|
||
"select clm_cntl_num, clm_orig_cntl_num, id, last_updated, final_action, _job_key, _transaction_time "
|
||
"from cur_explanation_of_benefit where clm_orig_cntl_num is not null and clm_orig_cntl_num <> ''"
|
||
)
|
||
staged["eob_supersession"] = _write_current(con, store, "eob_supersession")
|
||
if "explanation_of_benefit" in rep.counts:
|
||
rep.counts["explanation_of_benefit"]["orphans"] = (
|
||
con.execute(
|
||
"select count(*) from cur_explanation_of_benefit e where bene_id is not null and not exists "
|
||
"(select 1 from cur_patient p where coalesce(p.bene_id, p.id) = e.bene_id)"
|
||
).fetchone()[0]
|
||
if _table_exists(con, "cur_patient")
|
||
else 0
|
||
)
|
||
if (
|
||
"coverage" in rep.counts
|
||
and _table_exists(con, "cur_patient")
|
||
and _table_exists(con, "cur_coverage")
|
||
):
|
||
rep.counts["coverage"]["orphans"] = con.execute(
|
||
"select count(*) from cur_coverage c where not exists (select 1 from cur_patient p "
|
||
"where coalesce(p.bene_id, p.id) = regexp_replace(c.beneficiary_reference, '^Patient/', ''))"
|
||
).fetchone()[0]
|
||
|
||
_invariants(con, store, job, rep, prev_counts, seqs)
|
||
|
||
# all good: publish the staged files atomically, one by one
|
||
for table, tmp in staged.items():
|
||
os.replace(tmp, _current_path(store, table))
|
||
|
||
|
||
def _table_exists(con: Any, name: str) -> bool:
|
||
return bool(
|
||
con.execute(
|
||
"select count(*) from information_schema.tables where table_name = ?",
|
||
[name],
|
||
).fetchone()[0]
|
||
)
|
||
|
||
|
||
def _retired(con: Any, table: str) -> int:
|
||
return con.execute(
|
||
f"select count(*) from prev_{table} p where not exists "
|
||
f"(select 1 from (select {KEY_EXPR[table]} as _key from cur_{table}) c where c._key = p._key)"
|
||
).fetchone()[0]
|
||
|
||
|
||
def _write_current(con: Any, store: Store, table: str) -> Path:
|
||
out = _current_path(store, table)
|
||
out.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = out.with_name(f".{table}.parquet.tmp")
|
||
con.execute(f"copy cur_{table} to {_sql_literal(str(tmp))} (format parquet)")
|
||
return tmp
|
||
|
||
|
||
def _classify(
|
||
con: Any,
|
||
store: Store,
|
||
job: JobInfo,
|
||
table: str,
|
||
counts: dict[str, int],
|
||
prev_patients: str,
|
||
) -> None:
|
||
"""Second-order counts for *job*'s partition of *table* against the previous current state."""
|
||
key = KEY_EXPR[table]
|
||
p = _partition(store, job.key, table)
|
||
con.execute(
|
||
f"create or replace table inc_{table} as select {key} as _key, id, last_updated, "
|
||
f"{BENE_EXPR[table]} as _bene, try_cast(last_updated as timestamptz) as _lu "
|
||
f"from read_parquet({_sql_literal(str(p))})"
|
||
)
|
||
con.execute(
|
||
f"create or replace table win_{table} as select {key} as _key, id, last_updated, _job_key "
|
||
f"from cur_{table}"
|
||
)
|
||
q = lambda sql: con.execute(sql).fetchone()[0] # noqa: E731
|
||
counts["rows"] = q(f"select count(*) from inc_{table}")
|
||
counts["new"] = q(
|
||
f"select count(distinct _key) from inc_{table} i where not exists (select 1 from prev_{table} p where p._key = i._key)"
|
||
)
|
||
counts["unchanged"] = q(
|
||
f"select count(distinct i._key) from inc_{table} i join prev_{table} p on p._key = i._key "
|
||
"where p.id is not distinct from i.id and p.last_updated is not distinct from i.last_updated"
|
||
)
|
||
counts["replaced"] = q(
|
||
f"select count(distinct i._key) from inc_{table} i join prev_{table} p on p._key = i._key "
|
||
f"join win_{table} w on w._key = i._key "
|
||
f"where not (p.id is not distinct from i.id and p.last_updated is not distinct from i.last_updated) "
|
||
f"and w._job_key = {_sql_literal(job.key)}"
|
||
)
|
||
counts["stale_ignored"] = q(
|
||
f"select count(distinct i._key) from inc_{table} i join prev_{table} p on p._key = i._key "
|
||
f"join win_{table} w on w._key = i._key "
|
||
f"where not (p.id is not distinct from i.id and p.last_updated is not distinct from i.last_updated) "
|
||
f"and w._job_key <> {_sql_literal(job.key)}"
|
||
)
|
||
if job.kind == "delta":
|
||
# Only rows that actually changed the store count: a re-delivered
|
||
# duplicate or a stale older version dated before _since is noise,
|
||
# not late data.
|
||
since_lit = _sql_literal(format_instant(parse_instant(job.since)))
|
||
effective = (
|
||
f"from inc_{table} i join win_{table} w on w._key = i._key "
|
||
f"left join prev_{table} p on p._key = i._key "
|
||
f"where w._job_key = {_sql_literal(job.key)} and w.id is not distinct from i.id "
|
||
"and not (p._key is not null and p.id is not distinct from i.id "
|
||
"and p.last_updated is not distinct from i.last_updated) "
|
||
f"and i._lu < {since_lit}::timestamptz "
|
||
)
|
||
counts["backfill"] = q(
|
||
f"select count(*) {effective} and i._bene not in ({prev_patients} where _bene is not null)"
|
||
)
|
||
counts["late"] = q(
|
||
f"select count(*) {effective} and i._bene in ({prev_patients} where _bene is not null)"
|
||
)
|
||
if table == "explanation_of_benefit":
|
||
counts["tombstoned"] = q(
|
||
f"select count(*) from cur_{table} where _job_key = {_sql_literal(job.key)} "
|
||
"and final_action = 'NotFinalAction'"
|
||
)
|
||
counts["adjusted"] = q(
|
||
f"select count(*) from cur_{table} where _job_key = {_sql_literal(job.key)} "
|
||
"and clm_orig_cntl_num is not null and clm_orig_cntl_num <> ''"
|
||
)
|
||
|
||
|
||
def _invariants(
|
||
con: Any,
|
||
store: Store,
|
||
job: JobInfo,
|
||
rep: JobReport,
|
||
prev_counts: dict[str, int],
|
||
seqs: dict[str, int],
|
||
) -> None:
|
||
"""Third-order checks; raise :class:`SyncInvariantError` before anything is published."""
|
||
problems: list[str] = []
|
||
for table in TABLE_OF.values():
|
||
if not _table_exists(con, f"cur_{table}"):
|
||
continue
|
||
n, d = con.execute(
|
||
f"select count(*), count(distinct {KEY_EXPR[table]}) from cur_{table}"
|
||
).fetchone()
|
||
if n != d:
|
||
problems.append(f"{table}: {n} rows but {d} distinct keys")
|
||
c = rep.counts.get(table)
|
||
if c is not None and TYPE_OF[table] in job.types:
|
||
expected = prev_counts[table] + c["new"] - c["retired"]
|
||
if c["current"] != expected:
|
||
problems.append(
|
||
f"{table}: count conservation failed — {prev_counts[table]} + {c['new']} new − {c['retired']} retired "
|
||
f"= {expected}, but current is {c['current']}"
|
||
)
|
||
if _table_exists(con, "cur_eob_item") and _table_exists(
|
||
con, "cur_explanation_of_benefit"
|
||
):
|
||
stray = con.execute(
|
||
"select count(*) from cur_eob_item i where not exists "
|
||
"(select 1 from cur_explanation_of_benefit h where h.clm_id = i.clm_id)"
|
||
).fetchone()[0]
|
||
if stray:
|
||
problems.append(f"eob_item: {stray} line(s) without a current header")
|
||
# watermark monotone: this job's cut-off vs the newest already-applied job of the same scope/type
|
||
for r in read_ledger(store):
|
||
if not r["ok"] or r["job_key"] == job.key:
|
||
continue
|
||
if (population_of(r["endpoint"]), r["type_filter"]) != job.scope:
|
||
continue
|
||
if not set(r["types"].split(",")) & set(job.types):
|
||
continue
|
||
prev_tt = _maybe_instant(r["transaction_time"])
|
||
if prev_tt and job.transaction_time < prev_tt and job.kind == "baseline":
|
||
rep.notes.append(
|
||
f"baseline {job.key} cut-off {format_instant(job.transaction_time)} is older than applied "
|
||
f"{r['job_key']} ({r['transaction_time']}); newest baseline stays authoritative"
|
||
)
|
||
if problems:
|
||
raise SyncInvariantError("; ".join(problems))
|
||
|
||
|
||
# ── Status ────────────────────────────────────────────────────
|
||
|
||
|
||
def status(store: Store) -> dict[str, Any]:
|
||
"""Watermarks, applied/pending jobs and current counts, for the UI and CLI."""
|
||
jobs = complete_jobs(store)
|
||
ledger = read_ledger(store)
|
||
applied = {r["job_key"] for r in ledger if r["ok"]}
|
||
wm = watermarks(jobs)
|
||
current: dict[str, int] = {}
|
||
for table in (*TABLE_OF.values(), "eob_item", "eob_supersession"):
|
||
p = _current_path(store, table)
|
||
if p.exists():
|
||
current[table] = pq.read_metadata(p).num_rows
|
||
return {
|
||
"watermarks": [
|
||
{
|
||
"population": e,
|
||
"type_filter": f,
|
||
"resource_type": t,
|
||
"job_key": w.job_key,
|
||
"transaction_time": format_instant(w.transaction_time),
|
||
"kind": w.kind,
|
||
"applied": w.job_key in applied,
|
||
}
|
||
for (e, f, t), w in sorted(wm.items())
|
||
],
|
||
"jobs": [
|
||
{
|
||
"key": j.key,
|
||
"kind": j.kind,
|
||
"endpoint": j.endpoint,
|
||
"population": j.population,
|
||
"type_filter": j.type_filter,
|
||
"types": ",".join(j.types),
|
||
"since": j.since,
|
||
"transaction_time": format_instant(j.transaction_time),
|
||
"transaction_source": j.transaction_source,
|
||
"applied": j.key in applied,
|
||
}
|
||
for j in jobs
|
||
],
|
||
"pending": [j.key for j in jobs if j.key not in applied],
|
||
"ledger": [
|
||
{
|
||
**r,
|
||
"counts": json.loads(r["counts"] or "{}"),
|
||
"notes": json.loads(r["notes"] or "[]"),
|
||
}
|
||
for r in ledger
|
||
],
|
||
"current": current,
|
||
}
|