feat(lake): publish gpci_proposed + primary_care_service_code to DuckLake (closes #630)
This commit is contained in:
@@ -26,6 +26,7 @@ SCHEMAS: dict[str, tuple[str, ...]] = {
|
||||
"carrier_locality",
|
||||
"clinical_labor",
|
||||
"gpci",
|
||||
"gpci_proposed",
|
||||
"medical_equipment",
|
||||
"medical_supply",
|
||||
"physician_work_time",
|
||||
@@ -36,6 +37,19 @@ SCHEMAS: dict[str, tuple[str, ...]] = {
|
||||
"cms": ("ingest_log",),
|
||||
}
|
||||
|
||||
# (monolith source, lake destination) — tables whose lake name differs
|
||||
# from their monolith name. The MSSP designated-primary-care-service
|
||||
# list (42 CFR 425.400(c), used for ACO beneficiary assignment) lives
|
||||
# monolith-side as a tuva/dbt intermediate; publish it under a stable
|
||||
# reference name the notebooks can cite (#630). Published whenever the
|
||||
# destination's schema is selected via --schemas.
|
||||
RENAMED: tuple[tuple[str, str], ...] = (
|
||||
(
|
||||
"cms_provider_attribution._primary_care_hcpcs_codes",
|
||||
"cms.primary_care_service_code",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
@@ -66,19 +80,28 @@ def main() -> int:
|
||||
)
|
||||
|
||||
src = connect.duckdb("aco") # read-only; resolves to the replica
|
||||
|
||||
def _publish(source: str, dest: str) -> bool:
|
||||
t = time.time()
|
||||
df = src.execute(f"SELECT * FROM {source}").pl() # noqa: S608
|
||||
ctx.save(dest, nw.from_native(df), mode="replace")
|
||||
back = nw.to_native(ctx.load(dest))
|
||||
ok = back.height == df.height
|
||||
label = source if source == dest else f"{source} → {dest}"
|
||||
print(
|
||||
f" {label}: {df.height} rows → lake "
|
||||
f"({back.height} read back) [{'OK' if ok else 'MISMATCH'}] "
|
||||
f"in {time.time() - t:.1f}s"
|
||||
)
|
||||
return ok
|
||||
|
||||
for schema in args.schemas:
|
||||
for table in SCHEMAS[schema]:
|
||||
t = time.time()
|
||||
df = src.execute(f"SELECT * FROM {schema}.{table}").pl() # noqa: S608
|
||||
ctx.save(f"{schema}.{table}", nw.from_native(df), mode="replace")
|
||||
back = nw.to_native(ctx.load(f"{schema}.{table}"))
|
||||
status = "OK" if back.height == df.height else "MISMATCH"
|
||||
print(
|
||||
f" {schema}.{table}: {df.height} rows → lake "
|
||||
f"({back.height} read back) [{status}] in {time.time() - t:.1f}s"
|
||||
)
|
||||
if status != "OK":
|
||||
if not _publish(f"{schema}.{table}", f"{schema}.{table}"):
|
||||
return 1
|
||||
for source, dest in RENAMED:
|
||||
if dest.split(".", 1)[0] in args.schemas and not _publish(source, dest):
|
||||
return 1
|
||||
src.close()
|
||||
print("lake publish complete")
|
||||
return 0
|
||||
|
||||
Reference in New Issue
Block a user