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

feat(backend): add NDJSON streaming on list endpoints

Browse Source
This commit is contained in:
Tyler
2026-06-19 19:28:17 -06:00
parent cbffdbd2af
commit 74a673a360
2 changed files with 207 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/api.py
+94 -18
View File
@@ -142,6 +142,38 @@ def _client_wants_json(request: Request) -> bool:
return False
def _wants_ndjson(request: Request) -> bool:
"""Content negotiation for list endpoints: NDJSON wins ties, NDJSON is the
default when no Accept header is set (consistent with parse endpoints).
Used by the GET list routes (/api/batches, /api/claims, /api/remittances,
/api/providers, /api/activity). When both ``application/x-ndjson`` and
``application/json`` appear in Accept, NDJSON wins. When only JSON is
asked for, JSON wins. When neither is asked for (e.g. browser default
``*/*`` or an empty Accept), NDJSON wins.
"""
accept = request.headers.get("accept", "")
if "application/x-ndjson" in accept:
return True
if "application/json" in accept:
return False
return True
def _ndjson_stream_list(
items: list[dict], total: int, returned: int, has_more: bool,
) -> Iterator[str]:
"""Yield NDJSON lines for a list endpoint: one ``item`` per dict, then a
final ``summary`` line. Mirrors spec section 6.2 streaming rule.
"""
for it in items:
yield json.dumps({"type": "item", "data": it}) + "\n"
yield json.dumps({
"type": "summary",
"data": {"total": total, "returned": returned, "has_more": has_more},
}) + "\n"
def _has_claim_validation_errors(result: ParseResult) -> bool:
return any(not c.validation.passed for c in result.claims)
@@ -384,7 +416,10 @@ def _batch_summary_claim_count(rec: BatchRecord) -> int:
@app.get("/api/batches")
def list_batches(limit: int = Query(100, ge=1, le=1000)) -> dict[str, Any]:
def list_batches(
request: Request,
limit: int = Query(100, ge=1, le=1000),
) -> Any:
"""Summary of all parsed batches, newest first."""
records = store.list(limit=limit)
items = [
@@ -398,11 +433,19 @@ def list_batches(limit: int = Query(100, ge=1, le=1000)) -> dict[str, Any]:
for r in records
]
all_records = store.all()
total = len(all_records)
returned = len(items)
has_more = total > returned
if _wants_ndjson(request):
return StreamingResponse(
_ndjson_stream_list(items, total, returned, has_more),
media_type="application/x-ndjson",
)
return {
"items": items,
"total": len(all_records),
"returned": len(items),
"has_more": len(all_records) > len(items),
"total": total,
"returned": returned,
"has_more": has_more,
}
@@ -419,6 +462,7 @@ def get_batch(batch_id: str) -> Any:
@app.get("/api/claims")
def list_claims(
request: Request,
batch_id: str | None = Query(None),
status: str | None = Query(None),
provider_npi: str | None = Query(None),
@@ -429,7 +473,7 @@ def list_claims(
order: str = Query("desc"),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
) -> dict[str, Any]:
) -> Any:
common = dict(
batch_id=batch_id,
status=status,
@@ -442,16 +486,24 @@ def list_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": len(items),
"has_more": total > offset + len(items),
"returned": returned,
"has_more": has_more,
}
@app.get("/api/remittances")
def list_remittances(
request: Request,
batch_id: str | None = Query(None),
payer: str | None = Query(None),
claim_id: str | None = Query(None),
@@ -461,7 +513,7 @@ def list_remittances(
order: str = Query("desc"),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
) -> dict[str, Any]:
) -> Any:
common = dict(
batch_id=batch_id,
payer=payer,
@@ -473,51 +525,75 @@ def list_remittances(
sort=sort, order=order, limit=limit, offset=offset, **common,
))
total = len(list(store.iter_remittances(**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": len(items),
"has_more": total > offset + len(items),
"returned": returned,
"has_more": has_more,
}
@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),
) -> dict[str, Any]:
) -> 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": len(items),
"returned": len(paged),
"has_more": len(items) > offset + len(paged),
"total": total,
"returned": returned,
"has_more": has_more,
}
@app.get("/api/activity")
def list_activity(
request: Request,
kind: str | None = Query(None),
since: str | None = Query(None),
limit: int = Query(200, ge=1, le=500),
) -> dict[str, Any]:
) -> Any:
events = store.recent_activity(limit=limit)
if kind is not None:
events = [e for e in events if e["kind"] == kind]
if since is not None:
events = [e for e in events if e["timestamp"] >= since]
total = len(events)
has_more = False
if _wants_ndjson(request):
return StreamingResponse(
_ndjson_stream_list(events, total, total, has_more),
media_type="application/x-ndjson",
)
return {
"items": events,
"total": len(events),
"returned": len(events),
"has_more": False,
"total": total,
"returned": total,
"has_more": has_more,
}