refactor(api): split claims router (list + stream + detail + serialize + line-recon)
Extracts the 5 /api/claims endpoints from api.py into
cyclone.api_routers.claims:
GET /api/claims (paginated list)
GET /api/claims/stream (NDJSON live-tail)
GET /api/claims/{claim_id} (drawer detail)
GET /api/claims/{claim_id}/serialize-837 (X12 regen)
GET /api/claims/{claim_id}/line-reconciliation (per-line view)
Plus the two private projection helpers _claim_line_dict and
_svc_to_dict which are only used by line-reconciliation.
claims_stream is re-exported at the cyclone.api module level so
test_api_stream_live.py's direct import keeps working (same pattern
as the activity_stream re-export).
Route ordering preserved: /stream is declared before /{claim_id} in
the router source so FastAPI's first-match routing doesn't swallow
the literal 'stream' segment as a claim id.
api.py: 2201 -> 1841 (-360). Net diff: +432 / -389.
Pytest: 8 failed / 735 passed / 16 skipped — identical to baseline.
57/57 claims-targeted tests pass.
Live smoke: all 5 routes return expected status codes (200 for
list/stream, 404 for missing-claim detail/serialize/line-recon).
This commit is contained in:
Tyler
+4
-366
@@ -145,6 +145,7 @@ from cyclone.api_routers import ( # noqa: E402
|
|||||||
acks,
|
acks,
|
||||||
activity,
|
activity,
|
||||||
batches,
|
batches,
|
||||||
|
claims,
|
||||||
clearhouse,
|
clearhouse,
|
||||||
config,
|
config,
|
||||||
health,
|
health,
|
||||||
@@ -160,12 +161,14 @@ app.include_router(clearhouse.router)
|
|||||||
app.include_router(config.router)
|
app.include_router(config.router)
|
||||||
app.include_router(activity.router)
|
app.include_router(activity.router)
|
||||||
app.include_router(batches.router)
|
app.include_router(batches.router)
|
||||||
|
app.include_router(claims.router)
|
||||||
|
|
||||||
# Backwards-compat re-exports: test_api_stream_live.py imports these
|
# Backwards-compat re-exports: test_api_stream_live.py imports these
|
||||||
# names directly from ``cyclone.api`` to invoke the endpoint coroutines
|
# names directly from ``cyclone.api`` to invoke the endpoint coroutines
|
||||||
# in-process. Re-pointing at the router keeps the public surface stable
|
# in-process. Re-pointing at the router keeps the public surface stable
|
||||||
# while the bodies live in the routers package.
|
# while the bodies live in the routers package.
|
||||||
activity_stream = activity.activity_stream_endpoint
|
activity_stream = activity.activity_stream_endpoint
|
||||||
|
claims_stream = claims.claims_stream_endpoint
|
||||||
|
|
||||||
|
|
||||||
def _resolve_payer(name: str) -> PayerConfig:
|
def _resolve_payer(name: str) -> PayerConfig:
|
||||||
@@ -1164,372 +1167,7 @@ def inbox_export_csv(lane: str):
|
|||||||
# --- /api/batches lives in cyclone.api_routers.batches ---
|
# --- /api/batches lives in cyclone.api_routers.batches ---
|
||||||
|
|
||||||
|
|
||||||
@app.get("/api/claims")
|
# --- /api/claims lives in cyclone.api_routers.claims ---
|
||||||
def list_claims(
|
|
||||||
request: Request,
|
|
||||||
batch_id: str | None = Query(None),
|
|
||||||
status: str | None = Query(None),
|
|
||||||
provider_npi: str | None = Query(None),
|
|
||||||
payer: str | None = Query(None),
|
|
||||||
date_from: str | None = Query(None),
|
|
||||||
date_to: str | None = Query(None),
|
|
||||||
sort: str | None = Query(None),
|
|
||||||
order: str = Query("desc"),
|
|
||||||
limit: int = Query(100, ge=1, le=1000),
|
|
||||||
offset: int = Query(0, ge=0),
|
|
||||||
) -> Any:
|
|
||||||
common = dict(
|
|
||||||
batch_id=batch_id,
|
|
||||||
status=status,
|
|
||||||
provider_npi=provider_npi,
|
|
||||||
payer=payer,
|
|
||||||
date_from=date_from,
|
|
||||||
date_to=date_to,
|
|
||||||
)
|
|
||||||
items = list(store.iter_claims(
|
|
||||||
sort=sort, order=order, limit=limit, offset=offset, **common,
|
|
||||||
))
|
|
||||||
total = len(list(store.iter_claims(**common)))
|
|
||||||
returned = len(items)
|
|
||||||
has_more = total > offset + returned
|
|
||||||
if _wants_ndjson(request):
|
|
||||||
return StreamingResponse(
|
|
||||||
_ndjson_stream_list(items, total, returned, has_more),
|
|
||||||
media_type="application/x-ndjson",
|
|
||||||
)
|
|
||||||
return {
|
|
||||||
"items": items,
|
|
||||||
"total": total,
|
|
||||||
"returned": returned,
|
|
||||||
"has_more": has_more,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
# --------------------------------------------------------------------------- #
|
|
||||||
# Live-tail NDJSON streaming endpoints (Phase 3 — SP5)
|
|
||||||
# --------------------------------------------------------------------------- #
|
|
||||||
|
|
||||||
|
|
||||||
@app.get("/api/claims/stream")
|
|
||||||
async def claims_stream(
|
|
||||||
request: Request,
|
|
||||||
status: str | None = Query(None),
|
|
||||||
provider_npi: str | None = Query(None),
|
|
||||||
payer: str | None = Query(None),
|
|
||||||
date_from: str | None = Query(None),
|
|
||||||
date_to: str | None = Query(None),
|
|
||||||
sort: str | None = Query(None),
|
|
||||||
order: str = Query("desc"),
|
|
||||||
limit: int = Query(100, ge=1, le=1000),
|
|
||||||
) -> StreamingResponse:
|
|
||||||
"""Stream Claims as NDJSON: snapshot first, then live events.
|
|
||||||
|
|
||||||
Wire format:
|
|
||||||
* ``{"type":"item","data":<claim>}`` per snapshot row, then per
|
|
||||||
new ``claim_written`` event
|
|
||||||
* ``{"type":"snapshot_end","data":{"count":N}}`` after the snapshot
|
|
||||||
* ``{"type":"heartbeat","data":{"ts":<iso>}}`` every
|
|
||||||
``CYCLONE_TAIL_HEARTBEAT_S`` seconds when idle
|
|
||||||
|
|
||||||
Query params mirror :func:`list_claims` so a frontend can swap a
|
|
||||||
one-shot fetch for a tail with no URL surgery.
|
|
||||||
|
|
||||||
NOTE: registered before ``/api/claims/{claim_id}`` so the literal
|
|
||||||
``stream`` path segment doesn't get matched as a claim id.
|
|
||||||
"""
|
|
||||||
bus: EventBus = request.app.state.event_bus
|
|
||||||
|
|
||||||
async def gen() -> AsyncIterator[bytes]:
|
|
||||||
# 1. Snapshot (eager — iter_claims returns a list already).
|
|
||||||
rows = store.iter_claims(
|
|
||||||
status=status, provider_npi=provider_npi, payer=payer,
|
|
||||||
date_from=date_from, date_to=date_to,
|
|
||||||
sort=sort or "-submission_date", order=order, limit=limit,
|
|
||||||
)
|
|
||||||
for row in rows:
|
|
||||||
yield _ndjson_line({"type": "item", "data": row})
|
|
||||||
yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}})
|
|
||||||
|
|
||||||
# 2. Subscribe + heartbeats.
|
|
||||||
async for chunk in _tail_events(request, bus, ["claim_written"]):
|
|
||||||
yield chunk
|
|
||||||
|
|
||||||
return StreamingResponse(gen(), media_type="application/x-ndjson")
|
|
||||||
|
|
||||||
|
|
||||||
@app.get("/api/claims/{claim_id}")
|
|
||||||
def get_claim_detail_endpoint(claim_id: str) -> dict:
|
|
||||||
"""Return one claim with full drawer context (SP4).
|
|
||||||
|
|
||||||
Body shape is produced by :meth:`CycloneStore.get_claim_detail`:
|
|
||||||
header, state, service lines, diagnoses, parties, validation,
|
|
||||||
raw segments, ``stateHistory`` (most-recent-first, capped at 50),
|
|
||||||
and a populated ``matchedRemittance`` block when paired.
|
|
||||||
|
|
||||||
Path param is ``claim_id`` (matches the SP3 ``/api/acks/{ack_id}``
|
|
||||||
convention). Returns 404 — never 500 — on a missing claim so the
|
|
||||||
UI can distinguish "doesn't exist" from a transient fetch error.
|
|
||||||
"""
|
|
||||||
body = store.get_claim_detail(claim_id)
|
|
||||||
if body is None:
|
|
||||||
raise HTTPException(
|
|
||||||
status_code=404,
|
|
||||||
detail={
|
|
||||||
"error": "Not found",
|
|
||||||
"detail": f"Claim {claim_id} not found",
|
|
||||||
},
|
|
||||||
)
|
|
||||||
return body
|
|
||||||
|
|
||||||
|
|
||||||
@app.get("/api/claims/{claim_id}/serialize-837")
|
|
||||||
def serialize_claim_as_837(claim_id: str):
|
|
||||||
"""Return the claim as a regenerated X12 837P file (SP8).
|
|
||||||
|
|
||||||
Loads the ClaimOutput from the persisted ``raw_json`` and runs the
|
|
||||||
outbound serializer. Returns 404 if the claim doesn't exist, 422 if
|
|
||||||
the stored payload has no parseable ClaimOutput (data integrity
|
|
||||||
issue, not a transient failure).
|
|
||||||
"""
|
|
||||||
with db.SessionLocal()() as s:
|
|
||||||
row = s.get(Claim, claim_id)
|
|
||||||
if row is None:
|
|
||||||
return JSONResponse(
|
|
||||||
{"error": "Not found", "detail": f"Claim {claim_id} not found"},
|
|
||||||
status_code=404,
|
|
||||||
)
|
|
||||||
if not row.raw_json:
|
|
||||||
return JSONResponse(
|
|
||||||
{
|
|
||||||
"error": "Unprocessable",
|
|
||||||
"detail": f"Claim {claim_id} has no raw_json; cannot serialize",
|
|
||||||
},
|
|
||||||
status_code=422,
|
|
||||||
)
|
|
||||||
try:
|
|
||||||
claim_obj = ClaimOutput.model_validate(row.raw_json)
|
|
||||||
except Exception as exc:
|
|
||||||
return JSONResponse(
|
|
||||||
{
|
|
||||||
"error": "Unprocessable",
|
|
||||||
"detail": f"Claim {claim_id} raw_json is malformed: {exc}",
|
|
||||||
},
|
|
||||||
status_code=422,
|
|
||||||
)
|
|
||||||
|
|
||||||
try:
|
|
||||||
text = serialize_837(claim_obj)
|
|
||||||
except SerializeError837 as exc:
|
|
||||||
return JSONResponse(
|
|
||||||
{"error": "Unprocessable", "detail": str(exc)},
|
|
||||||
status_code=422,
|
|
||||||
)
|
|
||||||
|
|
||||||
return Response(
|
|
||||||
content=text,
|
|
||||||
media_type="text/x12",
|
|
||||||
headers={
|
|
||||||
"Content-Disposition": f'attachment; filename="claim-{claim_id}.x12"'
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@app.get("/api/claims/{claim_id}/line-reconciliation")
|
|
||||||
def get_claim_line_reconciliation(claim_id: str) -> dict:
|
|
||||||
"""Per-line reconciliation view for the ClaimDrawer tab.
|
|
||||||
|
|
||||||
Spec §5.1. Returns the 837 service lines and 835 SVC composites
|
|
||||||
side-by-side, with per-line CAS adjustments and a summary block.
|
|
||||||
|
|
||||||
Architecture note: 837 service lines live in ``Claim.raw_json``
|
|
||||||
(not a separate ORM table), so the 837-side rows are read from the
|
|
||||||
JSON blob; the 835-side rows come from ``ServiceLinePayment`` ORM.
|
|
||||||
``LineReconciliation.claim_service_line_number`` stores the 1-based
|
|
||||||
line number to join them.
|
|
||||||
"""
|
|
||||||
from sqlalchemy import select
|
|
||||||
from cyclone.db import (
|
|
||||||
LineReconciliation, ServiceLinePayment, CasAdjustment,
|
|
||||||
)
|
|
||||||
import json as _json
|
|
||||||
from decimal import Decimal
|
|
||||||
|
|
||||||
with db.SessionLocal()() as s:
|
|
||||||
claim = s.get(db.Claim, claim_id)
|
|
||||||
if claim is None:
|
|
||||||
raise HTTPException(
|
|
||||||
status_code=404,
|
|
||||||
detail={"error": "Not found", "detail": f"Claim {claim_id} not found"},
|
|
||||||
)
|
|
||||||
|
|
||||||
# 837 service lines: from raw_json.
|
|
||||||
raw = claim.raw_json or {}
|
|
||||||
claim_lines_raw = raw.get("service_lines") or []
|
|
||||||
# Normalize to dicts for the response.
|
|
||||||
claim_lines = [_claim_line_dict(d) for d in claim_lines_raw]
|
|
||||||
|
|
||||||
# 835 service payments: ORM rows from the matched remit.
|
|
||||||
remits = list(
|
|
||||||
s.execute(
|
|
||||||
select(db.Remittance).where(db.Remittance.claim_id == claim_id)
|
|
||||||
).scalars().all()
|
|
||||||
)
|
|
||||||
svc_payments: list[dict] = []
|
|
||||||
svc_ids: list[int] = []
|
|
||||||
if remits:
|
|
||||||
svc_rows = list(
|
|
||||||
s.execute(
|
|
||||||
select(ServiceLinePayment).where(
|
|
||||||
ServiceLinePayment.remittance_id.in_([r.id for r in remits])
|
|
||||||
).order_by(ServiceLinePayment.line_number)
|
|
||||||
).scalars().all()
|
|
||||||
)
|
|
||||||
for svc in svc_rows:
|
|
||||||
d = _svc_to_dict(svc)
|
|
||||||
svc_payments.append(d)
|
|
||||||
svc_ids.append(svc.id)
|
|
||||||
|
|
||||||
# LineReconciliation rows.
|
|
||||||
lrs = list(
|
|
||||||
s.execute(
|
|
||||||
select(LineReconciliation).where(LineReconciliation.claim_id == claim_id)
|
|
||||||
).scalars().all()
|
|
||||||
)
|
|
||||||
# Index by claim_service_line_number and service_line_payment_id.
|
|
||||||
lr_by_claim_num: dict[int, LineReconciliation] = {
|
|
||||||
lr.claim_service_line_number: lr for lr in lrs if lr.claim_service_line_number is not None
|
|
||||||
}
|
|
||||||
lr_by_svc: dict[int, LineReconciliation] = {
|
|
||||||
lr.service_line_payment_id: lr for lr in lrs if lr.service_line_payment_id is not None
|
|
||||||
}
|
|
||||||
|
|
||||||
# CAS rows grouped by svc id.
|
|
||||||
cas_by_svc: dict[int, list[CasAdjustment]] = {}
|
|
||||||
if svc_ids:
|
|
||||||
cas_rows = list(
|
|
||||||
s.execute(
|
|
||||||
select(CasAdjustment).where(CasAdjustment.service_line_payment_id.in_(svc_ids))
|
|
||||||
).scalars().all()
|
|
||||||
)
|
|
||||||
for c in cas_rows:
|
|
||||||
cas_by_svc.setdefault(c.service_line_payment_id, []).append(c)
|
|
||||||
|
|
||||||
# Build output lines array, preserving 837 order then 835-only.
|
|
||||||
svc_by_id: dict[int, dict] = {d["id"]: d for d in svc_payments}
|
|
||||||
lines_out: list[dict] = []
|
|
||||||
billed_total = Decimal("0")
|
|
||||||
paid_total = Decimal("0")
|
|
||||||
adjustment_total = Decimal("0")
|
|
||||||
matched_count = 0
|
|
||||||
used_svc_ids: set[int] = set()
|
|
||||||
|
|
||||||
for cl in claim_lines:
|
|
||||||
billed_total += Decimal(str(cl["charge"]))
|
|
||||||
lr = lr_by_claim_num.get(cl["line_number"])
|
|
||||||
if lr is None:
|
|
||||||
lines_out.append({
|
|
||||||
"claim_service_line": cl,
|
|
||||||
"service_line_payment": None,
|
|
||||||
"status": "unmatched_837_only",
|
|
||||||
"adjustments": [],
|
|
||||||
})
|
|
||||||
continue
|
|
||||||
svc_id = lr.service_line_payment_id
|
|
||||||
svc = svc_by_id.get(svc_id) if svc_id else None
|
|
||||||
if svc_id is not None:
|
|
||||||
used_svc_ids.add(svc_id)
|
|
||||||
cas_list = cas_by_svc.get(svc_id, []) if svc_id is not None else []
|
|
||||||
cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0"))
|
|
||||||
if svc:
|
|
||||||
paid_total += Decimal(str(svc["payment"]))
|
|
||||||
adjustment_total += cas_total
|
|
||||||
if lr.status == "matched":
|
|
||||||
matched_count += 1
|
|
||||||
lines_out.append({
|
|
||||||
"claim_service_line": cl,
|
|
||||||
"service_line_payment": svc,
|
|
||||||
"status": lr.status,
|
|
||||||
"adjustments": [
|
|
||||||
{"group_code": c.group_code, "reason_code": c.reason_code,
|
|
||||||
"amount": str(Decimal(str(c.amount)))}
|
|
||||||
for c in cas_list
|
|
||||||
],
|
|
||||||
})
|
|
||||||
|
|
||||||
# 835-only lines (no claim match).
|
|
||||||
for lr in lrs:
|
|
||||||
if lr.claim_service_line_number is not None:
|
|
||||||
continue
|
|
||||||
svc_id = lr.service_line_payment_id
|
|
||||||
if svc_id is None:
|
|
||||||
continue
|
|
||||||
if svc_id in used_svc_ids:
|
|
||||||
continue
|
|
||||||
svc = svc_by_id.get(svc_id)
|
|
||||||
cas_list = cas_by_svc.get(svc_id, [])
|
|
||||||
cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0"))
|
|
||||||
if svc:
|
|
||||||
paid_total += Decimal(str(svc["payment"]))
|
|
||||||
adjustment_total += cas_total
|
|
||||||
lines_out.append({
|
|
||||||
"claim_service_line": None,
|
|
||||||
"service_line_payment": svc,
|
|
||||||
"status": lr.status,
|
|
||||||
"adjustments": [
|
|
||||||
{"group_code": c.group_code, "reason_code": c.reason_code,
|
|
||||||
"amount": str(Decimal(str(c.amount)))}
|
|
||||||
for c in cas_list
|
|
||||||
],
|
|
||||||
})
|
|
||||||
|
|
||||||
return {
|
|
||||||
"claim_id": claim_id,
|
|
||||||
"summary": {
|
|
||||||
"billed_total": str(billed_total),
|
|
||||||
"paid_total": str(paid_total),
|
|
||||||
"adjustment_total": str(adjustment_total),
|
|
||||||
"matched_lines": matched_count,
|
|
||||||
"total_lines": len(claim_lines),
|
|
||||||
},
|
|
||||||
"lines": lines_out,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def _claim_line_dict(d: dict) -> dict:
|
|
||||||
"""Project an 837 service-line dict from ``Claim.raw_json`` to wire shape."""
|
|
||||||
from decimal import Decimal
|
|
||||||
proc = d.get("procedure") or {}
|
|
||||||
charge = d.get("charge")
|
|
||||||
units = d.get("units")
|
|
||||||
return {
|
|
||||||
"line_number": d.get("line_number"),
|
|
||||||
"procedure_qualifier": proc.get("qualifier", "HC"),
|
|
||||||
"procedure_code": proc.get("code", ""),
|
|
||||||
"modifiers": proc.get("modifiers") or [],
|
|
||||||
"charge": str(Decimal(str(charge))) if charge is not None else "0",
|
|
||||||
"units": str(Decimal(str(units))) if units is not None else None,
|
|
||||||
"unit_type": d.get("unit_type"),
|
|
||||||
"service_date": d.get("service_date"),
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def _svc_to_dict(svc) -> dict:
|
|
||||||
"""Project an ORM ``ServiceLinePayment`` to wire shape."""
|
|
||||||
import json as _json
|
|
||||||
from decimal import Decimal
|
|
||||||
return {
|
|
||||||
"id": svc.id,
|
|
||||||
"line_number": svc.line_number,
|
|
||||||
"procedure_qualifier": svc.procedure_qualifier,
|
|
||||||
"procedure_code": svc.procedure_code,
|
|
||||||
"modifiers": _json.loads(svc.modifiers_json or "[]"),
|
|
||||||
"charge": str(Decimal(str(svc.charge))),
|
|
||||||
"payment": str(Decimal(str(svc.payment))),
|
|
||||||
"units": str(Decimal(str(svc.units))) if svc.units is not None else None,
|
|
||||||
"unit_type": svc.unit_type,
|
|
||||||
"service_date": svc.service_date.isoformat() if svc.service_date else None,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
@app.get("/api/reconciliation/unmatched")
|
@app.get("/api/reconciliation/unmatched")
|
||||||
def get_reconciliation_unmatched() -> dict:
|
def get_reconciliation_unmatched() -> dict:
|
||||||
|
|||||||
@@ -0,0 +1,411 @@
|
|||||||
|
"""``/api/claims`` — list, live-tail, detail, X12 regenerate, line-reconciliation.
|
||||||
|
|
||||||
|
This is the largest single-resource router. Five endpoints:
|
||||||
|
|
||||||
|
* ``GET /api/claims`` — paginated list with filters
|
||||||
|
(status / batch_id / provider_npi / payer / date range / sort).
|
||||||
|
* ``GET /api/claims/stream`` — NDJSON snapshot + live tail via the
|
||||||
|
``claim_written`` EventBus kind.
|
||||||
|
* ``GET /api/claims/{claim_id}`` — drawer detail
|
||||||
|
(header / state / service lines / diagnoses / parties / validation /
|
||||||
|
raw segments / stateHistory / matchedRemittance).
|
||||||
|
* ``GET /api/claims/{claim_id}/serialize-837`` — regenerate X12 for
|
||||||
|
download (SP8).
|
||||||
|
* ``GET /api/claims/{claim_id}/line-reconciliation`` — per-line
|
||||||
|
reconciliation view for the ClaimDrawer tab (spec §5.1).
|
||||||
|
|
||||||
|
**Route ordering.** ``/stream`` is declared **before** ``/{claim_id}`` so
|
||||||
|
FastAPI's first-match routing doesn't swallow the literal ``stream``
|
||||||
|
segment as a claim id. Same order as the prior inline code.
|
||||||
|
|
||||||
|
The two projection helpers ``_claim_line_dict`` / ``_svc_to_dict`` are
|
||||||
|
private to this router; both only used by the line-reconciliation
|
||||||
|
endpoint.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from decimal import Decimal
|
||||||
|
from typing import Any, AsyncIterator
|
||||||
|
|
||||||
|
from fastapi import APIRouter, HTTPException, Query, Request
|
||||||
|
from fastapi.responses import JSONResponse, Response, StreamingResponse
|
||||||
|
|
||||||
|
from cyclone import db
|
||||||
|
from cyclone.api_helpers import (
|
||||||
|
ndjson_line,
|
||||||
|
ndjson_stream_list,
|
||||||
|
tail_events,
|
||||||
|
wants_ndjson,
|
||||||
|
)
|
||||||
|
from cyclone.db import Claim
|
||||||
|
from cyclone.parsers.models import ClaimOutput
|
||||||
|
from cyclone.parsers.serialize_837 import SerializeError as SerializeError837, serialize_837
|
||||||
|
from cyclone.pubsub import EventBus
|
||||||
|
from cyclone.store import store
|
||||||
|
|
||||||
|
router = APIRouter()
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/claims")
|
||||||
|
def list_claims_endpoint(
|
||||||
|
request: Request,
|
||||||
|
batch_id: str | None = Query(None),
|
||||||
|
status: str | None = Query(None),
|
||||||
|
provider_npi: str | None = Query(None),
|
||||||
|
payer: str | None = Query(None),
|
||||||
|
date_from: str | None = Query(None),
|
||||||
|
date_to: str | None = Query(None),
|
||||||
|
sort: str | None = Query(None),
|
||||||
|
order: str = Query("desc"),
|
||||||
|
limit: int = Query(100, ge=1, le=1000),
|
||||||
|
offset: int = Query(0, ge=0),
|
||||||
|
) -> Any:
|
||||||
|
"""Paginated claims list with filters; supports NDJSON streaming.
|
||||||
|
|
||||||
|
``has_more`` is computed as ``total > offset + returned`` so paginating
|
||||||
|
forward keeps returning true until the last page is reached.
|
||||||
|
"""
|
||||||
|
common = dict(
|
||||||
|
batch_id=batch_id,
|
||||||
|
status=status,
|
||||||
|
provider_npi=provider_npi,
|
||||||
|
payer=payer,
|
||||||
|
date_from=date_from,
|
||||||
|
date_to=date_to,
|
||||||
|
)
|
||||||
|
items = list(store.iter_claims(
|
||||||
|
sort=sort, order=order, limit=limit, offset=offset, **common,
|
||||||
|
))
|
||||||
|
total = len(list(store.iter_claims(**common)))
|
||||||
|
returned = len(items)
|
||||||
|
has_more = total > offset + returned
|
||||||
|
if wants_ndjson(request):
|
||||||
|
return StreamingResponse(
|
||||||
|
ndjson_stream_list(items, total, returned, has_more),
|
||||||
|
media_type="application/x-ndjson",
|
||||||
|
)
|
||||||
|
return {
|
||||||
|
"items": items,
|
||||||
|
"total": total,
|
||||||
|
"returned": returned,
|
||||||
|
"has_more": has_more,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/claims/stream")
|
||||||
|
async def claims_stream_endpoint(
|
||||||
|
request: Request,
|
||||||
|
status: str | None = Query(None),
|
||||||
|
provider_npi: str | None = Query(None),
|
||||||
|
payer: str | None = Query(None),
|
||||||
|
date_from: str | None = Query(None),
|
||||||
|
date_to: str | None = Query(None),
|
||||||
|
sort: str | None = Query(None),
|
||||||
|
order: str = Query("desc"),
|
||||||
|
limit: int = Query(100, ge=1, le=1000),
|
||||||
|
) -> StreamingResponse:
|
||||||
|
"""Stream Claims as NDJSON: snapshot first, then live events.
|
||||||
|
|
||||||
|
Wire format:
|
||||||
|
* ``{"type":"item","data":<claim>}`` per snapshot row, then per
|
||||||
|
new ``claim_written`` event
|
||||||
|
* ``{"type":"snapshot_end","data":{"count":N}}`` after the snapshot
|
||||||
|
* ``{"type":"heartbeat","data":{"ts":<iso>}}`` every
|
||||||
|
``CYCLONE_TAIL_HEARTBEAT_S`` seconds when idle
|
||||||
|
|
||||||
|
Query params mirror :func:`list_claims_endpoint` so a frontend can
|
||||||
|
swap a one-shot fetch for a tail with no URL surgery.
|
||||||
|
|
||||||
|
NOTE: registered before ``/api/claims/{claim_id}`` so the literal
|
||||||
|
``stream`` path segment doesn't get matched as a claim id.
|
||||||
|
"""
|
||||||
|
bus: EventBus = request.app.state.event_bus
|
||||||
|
|
||||||
|
async def gen() -> AsyncIterator[bytes]:
|
||||||
|
# 1. Snapshot (eager — iter_claims returns a list already).
|
||||||
|
rows = store.iter_claims(
|
||||||
|
status=status, provider_npi=provider_npi, payer=payer,
|
||||||
|
date_from=date_from, date_to=date_to,
|
||||||
|
sort=sort or "-submission_date", order=order, limit=limit,
|
||||||
|
)
|
||||||
|
for row in rows:
|
||||||
|
yield ndjson_line({"type": "item", "data": row})
|
||||||
|
yield ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}})
|
||||||
|
|
||||||
|
# 2. Subscribe + heartbeats.
|
||||||
|
async for chunk in tail_events(request, bus, ["claim_written"]):
|
||||||
|
yield chunk
|
||||||
|
|
||||||
|
return StreamingResponse(gen(), media_type="application/x-ndjson")
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/claims/{claim_id}")
|
||||||
|
def get_claim_detail_endpoint(claim_id: str) -> dict:
|
||||||
|
"""Return one claim with full drawer context (SP4).
|
||||||
|
|
||||||
|
Body shape is produced by :meth:`CycloneStore.get_claim_detail`:
|
||||||
|
header, state, service lines, diagnoses, parties, validation,
|
||||||
|
raw segments, ``stateHistory`` (most-recent-first, capped at 50),
|
||||||
|
and a populated ``matchedRemittance`` block when paired.
|
||||||
|
|
||||||
|
Returns 404 — never 500 — on a missing claim so the UI can
|
||||||
|
distinguish "doesn't exist" from a transient fetch error.
|
||||||
|
"""
|
||||||
|
body = store.get_claim_detail(claim_id)
|
||||||
|
if body is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=404,
|
||||||
|
detail={
|
||||||
|
"error": "Not found",
|
||||||
|
"detail": f"Claim {claim_id} not found",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
return body
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/claims/{claim_id}/serialize-837")
|
||||||
|
def serialize_claim_as_837(claim_id: str):
|
||||||
|
"""Return the claim as a regenerated X12 837P file (SP8).
|
||||||
|
|
||||||
|
Loads the ClaimOutput from the persisted ``raw_json`` and runs the
|
||||||
|
outbound serializer. Returns 404 if the claim doesn't exist, 422 if
|
||||||
|
the stored payload has no parseable ClaimOutput (data integrity
|
||||||
|
issue, not a transient failure).
|
||||||
|
"""
|
||||||
|
with db.SessionLocal()() as s:
|
||||||
|
row = s.get(Claim, claim_id)
|
||||||
|
if row is None:
|
||||||
|
return JSONResponse(
|
||||||
|
{"error": "Not found", "detail": f"Claim {claim_id} not found"},
|
||||||
|
status_code=404,
|
||||||
|
)
|
||||||
|
if not row.raw_json:
|
||||||
|
return JSONResponse(
|
||||||
|
{
|
||||||
|
"error": "Unprocessable",
|
||||||
|
"detail": f"Claim {claim_id} has no raw_json; cannot serialize",
|
||||||
|
},
|
||||||
|
status_code=422,
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
claim_obj = ClaimOutput.model_validate(row.raw_json)
|
||||||
|
except Exception as exc:
|
||||||
|
return JSONResponse(
|
||||||
|
{
|
||||||
|
"error": "Unprocessable",
|
||||||
|
"detail": f"Claim {claim_id} raw_json is malformed: {exc}",
|
||||||
|
},
|
||||||
|
status_code=422,
|
||||||
|
)
|
||||||
|
|
||||||
|
try:
|
||||||
|
text = serialize_837(claim_obj)
|
||||||
|
except SerializeError837 as exc:
|
||||||
|
return JSONResponse(
|
||||||
|
{"error": "Unprocessable", "detail": str(exc)},
|
||||||
|
status_code=422,
|
||||||
|
)
|
||||||
|
|
||||||
|
return Response(
|
||||||
|
content=text,
|
||||||
|
media_type="text/x12",
|
||||||
|
headers={
|
||||||
|
"Content-Disposition": f'attachment; filename="claim-{claim_id}.x12"'
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/claims/{claim_id}/line-reconciliation")
|
||||||
|
def get_claim_line_reconciliation(claim_id: str) -> dict:
|
||||||
|
"""Per-line reconciliation view for the ClaimDrawer tab.
|
||||||
|
|
||||||
|
Spec §5.1. Returns the 837 service lines and 835 SVC composites
|
||||||
|
side-by-side, with per-line CAS adjustments and a summary block.
|
||||||
|
|
||||||
|
Architecture note: 837 service lines live in ``Claim.raw_json``
|
||||||
|
(not a separate ORM table), so the 837-side rows are read from the
|
||||||
|
JSON blob; the 835-side rows come from ``ServiceLinePayment`` ORM.
|
||||||
|
``LineReconciliation.claim_service_line_number`` stores the 1-based
|
||||||
|
line number to join them.
|
||||||
|
"""
|
||||||
|
from sqlalchemy import select
|
||||||
|
|
||||||
|
from cyclone.db import (
|
||||||
|
CasAdjustment,
|
||||||
|
LineReconciliation,
|
||||||
|
ServiceLinePayment,
|
||||||
|
)
|
||||||
|
|
||||||
|
with db.SessionLocal()() as s:
|
||||||
|
claim = s.get(db.Claim, claim_id)
|
||||||
|
if claim is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=404,
|
||||||
|
detail={"error": "Not found", "detail": f"Claim {claim_id} not found"},
|
||||||
|
)
|
||||||
|
|
||||||
|
# 837 service lines: from raw_json.
|
||||||
|
raw = claim.raw_json or {}
|
||||||
|
claim_lines_raw = raw.get("service_lines") or []
|
||||||
|
# Normalize to dicts for the response.
|
||||||
|
claim_lines = [_claim_line_dict(d) for d in claim_lines_raw]
|
||||||
|
|
||||||
|
# 835 service payments: ORM rows from the matched remit.
|
||||||
|
remits = list(
|
||||||
|
s.execute(
|
||||||
|
select(db.Remittance).where(db.Remittance.claim_id == claim_id)
|
||||||
|
).scalars().all()
|
||||||
|
)
|
||||||
|
svc_payments: list[dict] = []
|
||||||
|
svc_ids: list[int] = []
|
||||||
|
if remits:
|
||||||
|
svc_rows = list(
|
||||||
|
s.execute(
|
||||||
|
select(ServiceLinePayment).where(
|
||||||
|
ServiceLinePayment.remittance_id.in_([r.id for r in remits])
|
||||||
|
).order_by(ServiceLinePayment.line_number)
|
||||||
|
).scalars().all()
|
||||||
|
)
|
||||||
|
for svc in svc_rows:
|
||||||
|
d = _svc_to_dict(svc)
|
||||||
|
svc_payments.append(d)
|
||||||
|
svc_ids.append(svc.id)
|
||||||
|
|
||||||
|
# LineReconciliation rows.
|
||||||
|
lrs = list(
|
||||||
|
s.execute(
|
||||||
|
select(LineReconciliation).where(LineReconciliation.claim_id == claim_id)
|
||||||
|
).scalars().all()
|
||||||
|
)
|
||||||
|
# Index by claim_service_line_number and service_line_payment_id.
|
||||||
|
lr_by_claim_num: dict[int, LineReconciliation] = {
|
||||||
|
lr.claim_service_line_number: lr for lr in lrs if lr.claim_service_line_number is not None
|
||||||
|
}
|
||||||
|
lr_by_svc: dict[int, LineReconciliation] = {
|
||||||
|
lr.service_line_payment_id: lr for lr in lrs if lr.service_line_payment_id is not None
|
||||||
|
}
|
||||||
|
|
||||||
|
# CAS rows grouped by svc id.
|
||||||
|
cas_by_svc: dict[int, list[CasAdjustment]] = {}
|
||||||
|
if svc_ids:
|
||||||
|
cas_rows = list(
|
||||||
|
s.execute(
|
||||||
|
select(CasAdjustment).where(CasAdjustment.service_line_payment_id.in_(svc_ids))
|
||||||
|
).scalars().all()
|
||||||
|
)
|
||||||
|
for c in cas_rows:
|
||||||
|
cas_by_svc.setdefault(c.service_line_payment_id, []).append(c)
|
||||||
|
|
||||||
|
# Build output lines array, preserving 837 order then 835-only.
|
||||||
|
svc_by_id: dict[int, dict] = {d["id"]: d for d in svc_payments}
|
||||||
|
lines_out: list[dict] = []
|
||||||
|
billed_total = Decimal("0")
|
||||||
|
paid_total = Decimal("0")
|
||||||
|
adjustment_total = Decimal("0")
|
||||||
|
matched_count = 0
|
||||||
|
used_svc_ids: set[int] = set()
|
||||||
|
|
||||||
|
for cl in claim_lines:
|
||||||
|
billed_total += Decimal(str(cl["charge"]))
|
||||||
|
lr = lr_by_claim_num.get(cl["line_number"])
|
||||||
|
if lr is None:
|
||||||
|
lines_out.append({
|
||||||
|
"claim_service_line": cl,
|
||||||
|
"service_line_payment": None,
|
||||||
|
"status": "unmatched_837_only",
|
||||||
|
"adjustments": [],
|
||||||
|
})
|
||||||
|
continue
|
||||||
|
svc_id = lr.service_line_payment_id
|
||||||
|
svc = svc_by_id.get(svc_id) if svc_id else None
|
||||||
|
if svc_id is not None:
|
||||||
|
used_svc_ids.add(svc_id)
|
||||||
|
cas_list = cas_by_svc.get(svc_id, []) if svc_id is not None else []
|
||||||
|
cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0"))
|
||||||
|
if svc:
|
||||||
|
paid_total += Decimal(str(svc["payment"]))
|
||||||
|
adjustment_total += cas_total
|
||||||
|
if lr.status == "matched":
|
||||||
|
matched_count += 1
|
||||||
|
lines_out.append({
|
||||||
|
"claim_service_line": cl,
|
||||||
|
"service_line_payment": svc,
|
||||||
|
"status": lr.status,
|
||||||
|
"adjustments": [
|
||||||
|
{"group_code": c.group_code, "reason_code": c.reason_code,
|
||||||
|
"amount": str(Decimal(str(c.amount)))}
|
||||||
|
for c in cas_list
|
||||||
|
],
|
||||||
|
})
|
||||||
|
|
||||||
|
# 835-only lines (no claim match).
|
||||||
|
for lr in lrs:
|
||||||
|
if lr.claim_service_line_number is not None:
|
||||||
|
continue
|
||||||
|
svc_id = lr.service_line_payment_id
|
||||||
|
if svc_id is None:
|
||||||
|
continue
|
||||||
|
if svc_id in used_svc_ids:
|
||||||
|
continue
|
||||||
|
svc = svc_by_id.get(svc_id)
|
||||||
|
cas_list = cas_by_svc.get(svc_id, [])
|
||||||
|
cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0"))
|
||||||
|
if svc:
|
||||||
|
paid_total += Decimal(str(svc["payment"]))
|
||||||
|
adjustment_total += cas_total
|
||||||
|
lines_out.append({
|
||||||
|
"claim_service_line": None,
|
||||||
|
"service_line_payment": svc,
|
||||||
|
"status": lr.status,
|
||||||
|
"adjustments": [
|
||||||
|
{"group_code": c.group_code, "reason_code": c.reason_code,
|
||||||
|
"amount": str(Decimal(str(c.amount)))}
|
||||||
|
for c in cas_list
|
||||||
|
],
|
||||||
|
})
|
||||||
|
|
||||||
|
return {
|
||||||
|
"claim_id": claim_id,
|
||||||
|
"summary": {
|
||||||
|
"billed_total": str(billed_total),
|
||||||
|
"paid_total": str(paid_total),
|
||||||
|
"adjustment_total": str(adjustment_total),
|
||||||
|
"matched_lines": matched_count,
|
||||||
|
"total_lines": len(claim_lines),
|
||||||
|
},
|
||||||
|
"lines": lines_out,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _claim_line_dict(d: dict) -> dict:
|
||||||
|
"""Project an 837 service-line dict from ``Claim.raw_json`` to wire shape."""
|
||||||
|
proc = d.get("procedure") or {}
|
||||||
|
charge = d.get("charge")
|
||||||
|
units = d.get("units")
|
||||||
|
return {
|
||||||
|
"line_number": d.get("line_number"),
|
||||||
|
"procedure_qualifier": proc.get("qualifier", "HC"),
|
||||||
|
"procedure_code": proc.get("code", ""),
|
||||||
|
"modifiers": proc.get("modifiers") or [],
|
||||||
|
"charge": str(Decimal(str(charge))) if charge is not None else "0",
|
||||||
|
"units": str(Decimal(str(units))) if units is not None else None,
|
||||||
|
"unit_type": d.get("unit_type"),
|
||||||
|
"service_date": d.get("service_date"),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _svc_to_dict(svc) -> dict:
|
||||||
|
"""Project an ORM ``ServiceLinePayment`` to wire shape."""
|
||||||
|
import json as _json
|
||||||
|
return {
|
||||||
|
"id": svc.id,
|
||||||
|
"line_number": svc.line_number,
|
||||||
|
"procedure_qualifier": svc.procedure_qualifier,
|
||||||
|
"procedure_code": svc.procedure_code,
|
||||||
|
"modifiers": _json.loads(svc.modifiers_json or "[]"),
|
||||||
|
"charge": str(Decimal(str(svc.charge))),
|
||||||
|
"payment": str(Decimal(str(svc.payment))),
|
||||||
|
"units": str(Decimal(str(svc.units))) if svc.units is not None else None,
|
||||||
|
"unit_type": svc.unit_type,
|
||||||
|
"service_date": svc.service_date.isoformat() if svc.service_date else None,
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user