tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(api): split providers + clearhouse + config routers

Browse Source 
Checkpoint 2b. Extracts 7 more endpoints (1 + 2 + 4) into three
new routers:

- providers.py: GET /api/providers (list distinct providers from
  claim rows; npi/state filter; NDJSON or paginated JSON).
- clearhouse.py: GET /api/clearhouse + POST /api/clearhouse/submit.
  Carries the two serialize-from-raw helpers (_serialize_claim_for_submit,
  _serialize_claim_from_raw) since they're only used here.
- config.py: GET /api/config/providers + GET /api/config/providers/{npi}
  + GET /api/config/payers + GET /api/config/payers/{payer_id}/configs.
  Payer configs endpoint merges YAML-loaded blocks with any DB-overridden
  live blocks per payer.

api.py shrank from 2474 to 2310 lines (-164) and no longer has any
single resource's full request/response shape — every remaining route
now lives in a dedicated router module.

Verifies:
- Full pytest: 8 failed / 735 passed / 16 skipped — identical to
  clean main baseline (the 8 are pre-existing env/secret/sqlcipher).
- Live smoke: /api/health, /api/providers, /api/clearhouse,
  /api/config/payers, /api/config/payers/co_medicaid/configs all 200.
  /api/config/providers/{npi-not-seeded} returns 404 as expected.

See /tmp/refactor-cyclone.md for the full plan and progress.
This commit is contained in:
Tyler
2026-06-21 02:28:56 -06:00
parent bbf89c9dd8
commit 45cafbdb32
4 changed files with 250 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/api.py
+11 -175
View File
@@ -141,11 +141,21 @@ app.add_middleware(
# and are wired in here. (Kept as a top-level package rather than nested
# under `cyclone.api` so the existing ``cyclone.api`` module path keeps
# working — Python prefers packages over same-named modules.)
from cyclone.api_routers import acks, health, ta1_acks # noqa: E402
from cyclone.api_routers import ( # noqa: E402
acks,
clearhouse,
config,
health,
providers,
ta1_acks,
)
app.include_router(health.router)
app.include_router(acks.router)
app.include_router(ta1_acks.router)
app.include_router(providers.router)
app.include_router(clearhouse.router)
app.include_router(config.router)
def _resolve_payer(name: str) -> PayerConfig:
@@ -1803,36 +1813,6 @@ def get_remittance(remittance_id: str) -> dict:
return body
@app.get("/api/providers")
def list_providers(
request: Request,
npi: str | None = Query(None),
state: str | None = Query(None),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
) -> Any:
items = store.distinct_providers()
if npi is not None:
items = [p for p in items if p["npi"] == npi]
if state is not None:
items = [p for p in items if p.get("state") == state]
paged = items[offset:offset + limit]
total = len(items)
returned = len(paged)
has_more = total > offset + returned
if _wants_ndjson(request):
return StreamingResponse(
_ndjson_stream_list(paged, total, returned, has_more),
media_type="application/x-ndjson",
)
return {
"items": paged,
"total": total,
"returned": returned,
"has_more": has_more,
}
@app.get("/api/activity")
def list_activity(
request: Request,
@@ -2089,125 +2069,8 @@ async def post_eligibility_parse_271(
}
# ---------------------------------------------------------------------------
# SP9: providers / payers / clearhouse endpoints
# ---------------------------------------------------------------------------
@app.get("/api/clearhouse")
def get_clearhouse():
"""Return the singleton clearhouse config (dzinesco's identity, SFTP block, filename block)."""
ch = store.get_clearhouse()
if ch is None:
raise HTTPException(status_code=404, detail="clearhouse not seeded")
return json.loads(ch.model_dump_json())
@app.post("/api/clearhouse/submit")
def submit_to_clearhouse(body: dict):
"""Submit a batch of claims to the clearhouse (SFTP). SP9: stub.
Body: ``{"claim_ids": [...], "payer_id": "CO_TXIX"}``
Stub behavior: serializes each claim via the SP7 serializer, builds
an HCPF-compliant outbound filename, and copies the result to
``{staging_dir}/{outbound_path}/{filename}`` instead of opening a
real SFTP connection. Returns a receipt per claim.
"""
from cyclone.clearhouse import make_client
from cyclone.edi.filenames import build_outbound_filename
claim_ids = body.get("claim_ids", [])
payer_id = body.get("payer_id")
if not claim_ids:
raise HTTPException(status_code=400, detail="claim_ids required")
if not payer_id:
raise HTTPException(status_code=400, detail="payer_id required")
ch = store.get_clearhouse()
if ch is None:
raise HTTPException(status_code=500, detail="clearhouse not seeded")
client = make_client(ch.sftp_block)
results = []
for cid in claim_ids:
try:
x12_text = _serialize_claim_for_submit(cid)
except Exception as exc: # noqa: BLE001
results.append({"claim_id": cid, "ok": False, "error": str(exc)})
continue
filename = build_outbound_filename(ch.tpid, "837P")
remote = f"{ch.sftp_block.paths['outbound']}/{filename}"
staging_path = client.write_file(remote, x12_text.encode("utf-8"))
results.append({
"claim_id": cid,
"ok": True,
"filename": filename,
"staging_path": str(staging_path),
"remote_path": remote,
})
# SP11: audit trail for each successful clearhouse submission.
with db.SessionLocal()() as audit_s:
append_event(audit_s, AuditEvent(
event_type="clearhouse.submitted",
entity_type="claim",
entity_id=cid,
payload={
"filename": filename,
"remote_path": remote,
"tpid": ch.tpid,
"stub": ch.sftp_block.stub,
},
actor="clearhouse-submit",
))
audit_s.commit()
return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub}
def _serialize_claim_for_submit(claim_id: str) -> str:
"""Serialize a claim to X12 for SFTP submission. Lazy import of the
serializer to avoid pulling FastAPI machinery at module import time."""
from cyclone.parsers.serialize_837 import serialize_837
from cyclone import db
with db.SessionLocal()() as s:
row = s.get(db.Claim, claim_id)
if row is None:
raise ValueError(f"claim {claim_id!r} not found")
# Re-parse the stored raw_json to get a ClaimOutput
from cyclone.parsers.models import ClaimOutput, Envelope, Subscriber, Payer, BillingProvider
from cyclone.parsers.parse_837 import parse
raw = row.raw_json or {}
# Reconstruct minimal ClaimOutput from raw_json; this is best-effort.
return _serialize_claim_from_raw(row, raw)
def _serialize_claim_from_raw(claim_row, raw: dict) -> str:
"""Best-effort serializer that uses the stored raw_json to emit a fresh 837.
For SP9 this delegates to the existing serialize_837 helper if the
claim has a complete raw_segments array. Otherwise it returns a
minimal placeholder.
"""
from cyclone.parsers.serialize_837 import serialize_837
from cyclone.parsers.parse_837 import parse
# Re-parse the original batch text (need to re-derive from store).
# SP9 stub: if the claim has a `raw_json` with `x12_text`, use that.
if isinstance(raw, dict) and raw.get("x12_text"):
result = parse(raw["x12_text"])
if result.claims:
return serialize_837(result.claims[0])
# Fallback: raise so the caller sees an error.
raise RuntimeError(
f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text"
)
@app.get("/api/config/providers")
def list_configured_providers(is_active: bool | None = Query(default=True)):
"""List the configured provider rows (3 NPIs for SP9)."""
return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)]
# --------------------------------------------------------------------------- #
# SP11: tamper-evident audit log (admin)
@@ -2430,35 +2293,8 @@ def rotate_db_key_endpoint(body: dict | None = None) -> Any:
_db_rotate_lock.release()
@app.get("/api/config/providers/{npi}")
def get_configured_provider(npi: str):
p = store.get_provider(npi)
if p is None:
raise HTTPException(status_code=404, detail=f"provider {npi!r} not found")
return json.loads(p.model_dump_json())
@app.get("/api/config/payers")
def list_configured_payers(is_active: bool | None = Query(default=True)):
return [json.loads(p.model_dump_json()) for p in store.list_payers(is_active=is_active)]
@app.get("/api/config/payers/{payer_id}/configs")
def list_payer_configs(payer_id: str):
"""List all (transaction_type, config_json) blocks for a payer."""
from cyclone import payers as payer_loader
configs = [
{"transaction_type": tx, "config_json": block, "source": "yaml"}
for (pid, tx), block in payer_loader.all_configs().items()
if pid == payer_id
]
# Also check the DB for runtime-overridden configs
for tx in ("837P", "835", "277CA", "999", "TA1"):
live = store.get_payer_config(payer_id, tx)
if live is not None:
configs.append({"transaction_type": tx, "config_json": live, "source": "db"})
return configs
@app.post("/api/admin/reload-config")
def reload_config():
backend/src/cyclone/api_routers/clearhouse.py
+134
View File
@@ -0,0 +1,134 @@
"""``/api/clearhouse`` — SP9 single-payer SFTP submit surface.
Two endpoints:
- ``GET /api/clearhouse`` — return the singleton clearhouse config
(dzinesco's identity, SFTP block, filename block). 404 when the
config hasn't been seeded yet.
- ``POST /api/clearhouse/submit`` — submit a batch of claims to the
clearhouse (SFTP). SP9 ships a stub that copies to ``staging_dir``
instead of opening a real SFTP connection; SP13 wires the real
paramiko-backed client behind the same ``make_client()`` factory.
For each claim the endpoint serializes a fresh 837 via
:func:`_serialize_claim_for_submit`, builds an HCPF-compliant outbound
filename, and records an audit-log row on success. Failures are
captured per-claim so the operator sees exactly which claim_ids
didn't go through.
"""
from __future__ import annotations
import json
from fastapi import APIRouter, HTTPException
from cyclone import db
from cyclone.audit_log import AuditEvent, append_event
from cyclone.store import store
router = APIRouter()
@router.get("/api/clearhouse")
def get_clearhouse():
"""Return the singleton clearhouse config (dzinesco's identity, SFTP block, filename block)."""
ch = store.get_clearhouse()
if ch is None:
raise HTTPException(status_code=404, detail="clearhouse not seeded")
return json.loads(ch.model_dump_json())
@router.post("/api/clearhouse/submit")
def submit_to_clearhouse(body: dict):
"""Submit a batch of claims to the clearhouse (SFTP). SP9: stub.
Body: ``{"claim_ids": [...], "payer_id": "CO_TXIX"}``
Stub behavior: serializes each claim via the SP7 serializer, builds
an HCPF-compliant outbound filename, and copies the result to
``{staging_dir}/{outbound_path}/{filename}`` instead of opening a
real SFTP connection. Returns a receipt per claim.
"""
from cyclone.clearhouse import make_client
from cyclone.edi.filenames import build_outbound_filename
claim_ids = body.get("claim_ids", [])
payer_id = body.get("payer_id")
if not claim_ids:
raise HTTPException(status_code=400, detail="claim_ids required")
if not payer_id:
raise HTTPException(status_code=400, detail="payer_id required")
ch = store.get_clearhouse()
if ch is None:
raise HTTPException(status_code=500, detail="clearhouse not seeded")
client = make_client(ch.sftp_block)
results = []
for cid in claim_ids:
try:
x12_text = _serialize_claim_for_submit(cid)
except Exception as exc: # noqa: BLE001
results.append({"claim_id": cid, "ok": False, "error": str(exc)})
continue
filename = build_outbound_filename(ch.tpid, "837P")
remote = f"{ch.sftp_block.paths['outbound']}/{filename}"
staging_path = client.write_file(remote, x12_text.encode("utf-8"))
results.append({
"claim_id": cid,
"ok": True,
"filename": filename,
"staging_path": str(staging_path),
"remote_path": remote,
})
# SP11: audit trail for each successful clearhouse submission.
with db.SessionLocal()() as audit_s:
append_event(audit_s, AuditEvent(
event_type="clearhouse.submitted",
entity_type="claim",
entity_id=cid,
payload={
"filename": filename,
"remote_path": remote,
"tpid": ch.tpid,
"stub": ch.sftp_block.stub,
},
actor="clearhouse-submit",
))
audit_s.commit()
return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub}
def _serialize_claim_for_submit(claim_id: str) -> str:
"""Serialize a claim to X12 for SFTP submission. Lazy import of the
serializer to avoid pulling FastAPI machinery at module import time."""
from cyclone import db as _db
with _db.SessionLocal()() as s:
row = s.get(_db.Claim, claim_id)
if row is None:
raise ValueError(f"claim {claim_id!r} not found")
# Re-parse the stored raw_json to get a ClaimOutput
raw = row.raw_json or {}
# Reconstruct minimal ClaimOutput from raw_json; this is best-effort.
return _serialize_claim_from_raw(row, raw)
def _serialize_claim_from_raw(claim_row, raw: dict) -> str:
"""Best-effort serializer that uses the stored raw_json to emit a fresh 837.
For SP9 this delegates to the existing serialize_837 helper if the
claim has a complete raw_segments array. Otherwise it returns a
minimal placeholder.
"""
from cyclone.parsers.serialize_837 import serialize_837
from cyclone.parsers.parse_837 import parse
# Re-parse the original batch text (need to re-derive from store).
# SP9 stub: if the claim has a `raw_json` with `x12_text`, use that.
if isinstance(raw, dict) and raw.get("x12_text"):
result = parse(raw["x12_text"])
if result.claims:
return serialize_837(result.claims[0])
# Fallback: raise so the caller sees an error.
raise RuntimeError(
f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text"
)
backend/src/cyclone/api_routers/config.py
+56
View File
@@ -0,0 +1,56 @@
"""``/api/config/*`` — configured-provider / configured-payer lookups.
The ``config`` prefix is the operator-facing read-only window into the
``config/payers.yaml`` static config and the seeded providers table.
Used by the frontend Settings page to show what's available.
The payers endpoint at ``/api/config/payers/{payer_id}/configs`` is
slightly different: it returns both the YAML-loaded config and any
DB-overridden config (so a runtime override wins over the static
fallback). 404 / 200 / array shape match the rest of the API.
"""
from __future__ import annotations
import json
from fastapi import APIRouter, HTTPException, Query
from cyclone import payers as payer_loader
from cyclone.store import store
router = APIRouter()
@router.get("/api/config/providers")
def list_configured_providers(is_active: bool | None = Query(default=True)):
"""List the configured provider rows (3 NPIs for SP9)."""
return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)]
@router.get("/api/config/providers/{npi}")
def get_configured_provider(npi: str):
p = store.get_provider(npi)
if p is None:
raise HTTPException(status_code=404, detail=f"provider {npi!r} not found")
return json.loads(p.model_dump_json())
@router.get("/api/config/payers")
def list_configured_payers(is_active: bool | None = Query(default=True)):
return [json.loads(p.model_dump_json()) for p in store.list_payers(is_active=is_active)]
@router.get("/api/config/payers/{payer_id}/configs")
def list_payer_configs(payer_id: str):
"""List all (transaction_type, config_json) blocks for a payer."""
configs = [
{"transaction_type": tx, "config_json": block, "source": "yaml"}
for (pid, tx), block in payer_loader.all_configs().items()
if pid == payer_id
]
# Also check the DB for runtime-overridden configs
for tx in ("837P", "835", "277CA", "999", "TA1"):
live = store.get_payer_config(payer_id, tx)
if live is not None:
configs.append({"transaction_type": tx, "config_json": live, "source": "db"})
return configs
backend/src/cyclone/api_routers/providers.py
+49
View File
@@ -0,0 +1,49 @@
"""``GET /api/providers`` — distinct providers observed across claims.
The list is built by ``store.distinct_providers()`` which scans claim
rows and returns one entry per unique ``billing_provider_npi``. The
``npi`` and ``state`` query params filter client-side. ``limit`` /
``offset`` paginate. Accepts ``application/x-ndjson`` like the other
list endpoints — see ``api_helpers.ndjson_stream_list``.
"""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Query, Request
from fastapi.responses import StreamingResponse
from cyclone.api_helpers import ndjson_stream_list, wants_ndjson
from cyclone.store import store
router = APIRouter()
@router.get("/api/providers")
def list_providers(
request: Request,
npi: str | None = Query(None),
state: str | None = Query(None),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
) -> Any:
items = store.distinct_providers()
if npi is not None:
items = [p for p in items if p["npi"] == npi]
if state is not None:
items = [p for p in items if p.get("state") == state]
paged = items[offset:offset + limit]
total = len(items)
returned = len(paged)
has_more = total > offset + returned
if wants_ndjson(request):
return StreamingResponse(
ndjson_stream_list(paged, total, returned, has_more),
media_type="application/x-ndjson",
)
return {
"items": paged,
"total": total,
"returned": returned,
"has_more": has_more,
}