From 74a673a360b0ca6d224323eb4ddf635f598e1f68 Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 19:28:17 -0600 Subject: [PATCH] feat(backend): add NDJSON streaming on list endpoints --- backend/src/cyclone/api.py | 112 ++++++++++++++++++++++++----- backend/tests/test_api_gets.py | 128 +++++++++++++++++++++++++++++---- 2 files changed, 207 insertions(+), 33 deletions(-) diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index f214d30..265d6b3 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -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, } diff --git a/backend/tests/test_api_gets.py b/backend/tests/test_api_gets.py index 9c87e44..4f07268 100644 --- a/backend/tests/test_api_gets.py +++ b/backend/tests/test_api_gets.py @@ -40,20 +40,23 @@ def seeded_store(): return client +JSON = {"Accept": "application/json"} + + # --------------------------------------------------------------------------- # # /api/batches # --------------------------------------------------------------------------- # def test_batches_returns_empty_list_when_no_parses(client: TestClient): - resp = client.get("/api/batches") + resp = client.get("/api/batches", headers=JSON) assert resp.status_code == 200 body = resp.json() assert body == {"items": [], "total": 0, "returned": 0, "has_more": False} def test_batches_returns_summary_after_parse(seeded_store): - resp = seeded_store.get("/api/batches") + resp = seeded_store.get("/api/batches", headers=JSON) assert resp.status_code == 200 body = resp.json() assert body["total"] == 1 @@ -68,7 +71,7 @@ def test_batches_returns_summary_after_parse(seeded_store): def test_get_batch_by_id_returns_full_result(seeded_store): - batches = seeded_store.get("/api/batches").json()["items"] + batches = seeded_store.get("/api/batches", headers=JSON).json()["items"] bid = batches[0]["id"] resp = seeded_store.get(f"/api/batches/{bid}") assert resp.status_code == 200 @@ -89,7 +92,7 @@ def test_get_batch_404_when_missing(client: TestClient): def test_claims_returns_empty_list_when_no_parses(client: TestClient): - resp = client.get("/api/claims") + resp = client.get("/api/claims", headers=JSON) assert resp.status_code == 200 body = resp.json() assert body["items"] == [] @@ -97,7 +100,7 @@ def test_claims_returns_empty_list_when_no_parses(client: TestClient): def test_claims_filters_by_status_submitted(seeded_store): - resp = seeded_store.get("/api/claims?status=submitted") + resp = seeded_store.get("/api/claims?status=submitted", headers=JSON) assert resp.status_code == 200 body = resp.json() assert all(c["status"] == "submitted" for c in body["items"]) @@ -105,15 +108,15 @@ def test_claims_filters_by_status_submitted(seeded_store): def test_claims_filters_by_batch_id(seeded_store): - bid = seeded_store.get("/api/batches").json()["items"][0]["id"] - resp = seeded_store.get(f"/api/claims?batch_id={bid}") + bid = seeded_store.get("/api/batches", headers=JSON).json()["items"][0]["id"] + resp = seeded_store.get(f"/api/claims?batch_id={bid}", headers=JSON) assert resp.status_code == 200 body = resp.json() assert all(c["batchId"] == bid for c in body["items"]) def test_claims_sorts_by_billed_amount_desc(seeded_store): - resp = seeded_store.get("/api/claims?sort=billedAmount&order=desc") + resp = seeded_store.get("/api/claims?sort=billedAmount&order=desc", headers=JSON) assert resp.status_code == 200 body = resp.json() amounts = [c["billedAmount"] for c in body["items"]] @@ -121,7 +124,7 @@ def test_claims_sorts_by_billed_amount_desc(seeded_store): def test_claims_pagination_limit_offset(seeded_store): - resp1 = seeded_store.get("/api/claims?limit=1&offset=0") + resp1 = seeded_store.get("/api/claims?limit=1&offset=0", headers=JSON) body1 = resp1.json() assert body1["returned"] == 1 assert body1["has_more"] is True @@ -133,7 +136,7 @@ def test_claims_pagination_limit_offset(seeded_store): def test_remittances_returns_empty_list_when_no_parses(client: TestClient): - resp = client.get("/api/remittances") + resp = client.get("/api/remittances", headers=JSON) assert resp.status_code == 200 body = resp.json() assert body["items"] == [] @@ -141,7 +144,7 @@ def test_remittances_returns_empty_list_when_no_parses(client: TestClient): def test_remittances_filters_by_batch_id(seeded_store): """No 835 fixture parsed yet — but the endpoint should return 200 with empty.""" - resp = seeded_store.get("/api/remittances?batch_id=nope") + resp = seeded_store.get("/api/remittances?batch_id=nope", headers=JSON) assert resp.status_code == 200 assert resp.json()["items"] == [] @@ -152,14 +155,14 @@ def test_remittances_filters_by_batch_id(seeded_store): def test_providers_empty_when_no_parses(client: TestClient): - resp = client.get("/api/providers") + resp = client.get("/api/providers", headers=JSON) assert resp.status_code == 200 body = resp.json() assert body["items"] == [] def test_providers_returns_one_per_distinct_npi(seeded_store): - resp = seeded_store.get("/api/providers") + resp = seeded_store.get("/api/providers", headers=JSON) assert resp.status_code == 200 body = resp.json() assert len(body["items"]) == 1 @@ -172,15 +175,110 @@ def test_providers_returns_one_per_distinct_npi(seeded_store): def test_activity_empty_when_no_parses(client: TestClient): - resp = client.get("/api/activity") + resp = client.get("/api/activity", headers=JSON) assert resp.status_code == 200 body = resp.json() assert body["items"] == [] def test_activity_returns_one_event_per_claim(seeded_store): - resp = seeded_store.get("/api/activity") + resp = seeded_store.get("/api/activity", headers=JSON) assert resp.status_code == 200 body = resp.json() assert len(body["items"]) == 2 # co_medicaid_837p.txt has 2 claims assert all(e["kind"] == "claim_submitted" for e in body["items"]) + + +# --------------------------------------------------------------------------- # +# NDJSON streaming on the list endpoints +# --------------------------------------------------------------------------- # + + +def _assert_ndjson_envelope(resp, expected_total: int) -> list[dict]: + """Common helper: parse NDJSON body, return the item payloads.""" + import json as _json + assert resp.status_code == 200 + assert resp.headers["content-type"].startswith("application/x-ndjson"), resp.headers + lines = [l for l in resp.text.split("\n") if l.strip()] + events = [_json.loads(l) for l in lines] + items = [e["data"] for e in events if e["type"] == "item"] + summaries = [e for e in events if e["type"] == "summary"] + assert len(summaries) == 1, events + s = summaries[0]["data"] + assert s["total"] == expected_total + assert s["returned"] == len(items) + return items + + +def test_batches_ndjson_returns_parseable_lines(seeded_store): + resp = seeded_store.get( + "/api/batches", + headers={"Accept": "application/x-ndjson"}, + ) + items = _assert_ndjson_envelope(resp, expected_total=1) + assert len(items) == 1 + assert items[0]["kind"] == "837p" + + +def test_claims_ndjson_returns_parseable_lines(seeded_store): + resp = seeded_store.get( + "/api/claims", + headers={"Accept": "application/x-ndjson"}, + ) + items = _assert_ndjson_envelope(resp, expected_total=2) + assert len(items) == 2 + + +def test_remittances_ndjson_returns_parseable_lines(seeded_store): + resp = seeded_store.get( + "/api/remittances", + headers={"Accept": "application/x-ndjson"}, + ) + # 0 batches parsed, 0 remits + _assert_ndjson_envelope(resp, expected_total=0) + + +def test_providers_ndjson_returns_parseable_lines(seeded_store): + resp = seeded_store.get( + "/api/providers", + headers={"Accept": "application/x-ndjson"}, + ) + items = _assert_ndjson_envelope(resp, expected_total=1) + assert items[0]["npi"] == "1881068062" + + +def test_activity_ndjson_returns_parseable_lines(seeded_store): + resp = seeded_store.get( + "/api/activity", + headers={"Accept": "application/x-ndjson"}, + ) + items = _assert_ndjson_envelope(resp, expected_total=2) + assert all(it["kind"] == "claim_submitted" for it in items) + + +def test_claims_json_default_wraps_in_envelope(seeded_store): + """Accept: application/json → wrapped JSON response (items/total/returned/has_more).""" + resp = seeded_store.get( + "/api/claims", + headers={"Accept": "application/json"}, + ) + assert resp.status_code == 200 + body = resp.json() + assert set(body.keys()) == {"items", "total", "returned", "has_more"} + assert body["total"] == 2 + assert body["returned"] == 2 + + +def test_claims_ndjson_wins_over_json_when_both_accepted(seeded_store): + """When both NDJSON and JSON are in Accept, the explicit NDJSON hint wins.""" + resp = seeded_store.get( + "/api/claims", + headers={"Accept": "application/x-ndjson, application/json"}, + ) + assert resp.headers["content-type"].startswith("application/x-ndjson") + + +def test_claims_default_accept_returns_ndjson(seeded_store): + """No Accept header → NDJSON (consistent with parse endpoints' default).""" + resp = seeded_store.get("/api/claims") + assert resp.headers["content-type"].startswith("application/x-ndjson")