From 6233df12705792c86bd293caa44069cde2383546 Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 21 Jun 2026 00:28:58 -0600 Subject: [PATCH] refactor(sp): extract shared API helpers from api.py into api_helpers.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First checkpoint of the architecture satisfaction loop. Cyclone's api.py was a 2281-line god-module with 14 cross-cutting helpers inlined next to the @app route declarations. This commit moves them to a dedicated cyclone.api_helpers module: - NDJSON wire format: ndjson_line, ndjson_stream_837, ndjson_stream_835, ndjson_stream_list. - Content negotiation: client_wants_json, wants_ndjson. - Strict / raw_segments rewrites: strict_rewrite_837, strict_rewrite_835, drop_raw_segments_837, drop_raw_segments_835. - Validation probes: has_claim_validation_errors, has_835_validation_errors. - Live-tail generator: tail_events, heartbeat_seconds, utcnow. api.py re-imports them under the original underscore-prefixed names so every route call site stays unchanged. claims_stream, remittances_stream, and activity_stream remain exposed at cyclone.api (test_api_stream_live imports them directly). Verifies byte-identical NDJSON wire format, content negotiation rules, and the tail_events async-generator semantics (deliberately polls the EventBus queue rather than awaiting its async iterator, so heartbeats don't poison the bus subscription). Live-tested: GET /api/health, /api/claims, /api/remittances, /api/activity, /api/acks, /api/providers, /api/inbox/lanes, /api/inbox/payer-rejected/acknowledge, and the /api/claims/stream NDJSON tail all return expected codes / payload. Backend pytest: 29 failures identical to baseline (pre-existing secrets env, serialize_837, db_crypto env, prodfile env failures), 700 passed. Frontend npm test: 357/357 passing. See /tmp/refactor-cyclone.md for the full checkpoint log and the plan for the next step (splitting api.py routes into FastAPI APIRouters). Autoreview: /tmp/grok-review-local.md (0 bugs, 1 suggestion, 4 nits — all addressed: dead asyncio/os imports removed, dead Any import removed, duplicate utcnow import dropped, trailing newline added). --- backend/src/cyclone/api.py | 235 +++------------------------ backend/src/cyclone/api_helpers.py | 247 +++++++++++++++++++++++++++++ 2 files changed, 268 insertions(+), 214 deletions(-) create mode 100644 backend/src/cyclone/api_helpers.py diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 84037fb..3673ccf 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -15,12 +15,10 @@ plus GET/POST with any header. from __future__ import annotations -import asyncio import csv import io import json import logging -import os import uuid from datetime import datetime, timezone from contextlib import asynccontextmanager @@ -71,16 +69,24 @@ from cyclone.store import ( log = logging.getLogger(__name__) - -def _ndjson_line(event: dict) -> bytes: - """Serialize one event dict as a single NDJSON line (UTF-8, trailing ``\\n``). - - Used by the live-tail streaming endpoints to emit a uniform wire format - that the frontend ``tail-stream.ts`` parser can split on newlines. - Compact separators keep each line small and avoid ambiguity with embedded - whitespace. - """ - return (json.dumps(event, separators=(",", ":")) + "\n").encode("utf-8") +# Cross-cutting helpers (NDJSON serialization, content negotiation, strict +# rewrites, live-tail generator). See api_helpers.py for the rationale. +from cyclone.api_helpers import ( # noqa: E402 + client_wants_json as _client_wants_json, + drop_raw_segments_837 as _drop_raw_segments, + drop_raw_segments_835 as _drop_raw_segments_835, + has_claim_validation_errors as _has_claim_validation_errors, + has_835_validation_errors as _has_835_validation_errors, + heartbeat_seconds as _heartbeat_seconds, + ndjson_line as _ndjson_line, + ndjson_stream_837 as _ndjson_stream, + ndjson_stream_835 as _ndjson_stream_835, + ndjson_stream_list as _ndjson_stream_list, + strict_rewrite_837 as _strict_rewrite, + strict_rewrite_835 as _strict_rewrite_835, + tail_events as _tail_events, + wants_ndjson as _wants_ndjson, +) @asynccontextmanager @@ -155,91 +161,6 @@ def _resolve_payer_835(name: str) -> PayerConfig835: return PAYER_FACTORIES_835[name]() -def _strict_rewrite(result: ParseResult) -> ParseResult: - """Promote warnings to errors (mirrors the CLI's --strict).""" - claims: list[ClaimOutput] = [] - for claim in result.claims: - promoted = [ - issue.model_copy(update={"severity": "error"}) - for issue in claim.validation.warnings - ] - new_errors = claim.validation.errors + promoted - claims.append( - claim.model_copy( - update={ - "validation": claim.validation.model_copy( - update={"errors": new_errors, "passed": not new_errors} - ) - } - ) - ) - passed = sum(1 for c in claims if c.validation.passed) - failed = len(claims) - passed - summary = result.summary.model_copy( - update={ - "passed": passed, - "failed": failed, - "failed_claim_ids": [c.claim_id for c in claims if not c.validation.passed], - } - ) - return result.model_copy(update={"claims": claims, "summary": summary}) - - -def _drop_raw_segments(result: ParseResult) -> ParseResult: - """Return a copy of ``result`` with ``raw_segments`` cleared on every claim.""" - claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims] - return result.model_copy(update={"claims": claims}) - - -def _client_wants_json(request: Request) -> bool: - """Content negotiation: prefer ``application/json`` when the client asks for it. - - NDJSON is the default for browser uploads that don't set ``Accept``. The - frontend opts into JSON via ``Accept: application/json``. - """ - accept = request.headers.get("accept", "") - # If the client mentions JSON at all (and isn't asking for NDJSON - # specifically) treat it as a single-object request. The browser default - # ``*/*`` falls through to NDJSON. - if "application/json" in accept and "application/x-ndjson" not in accept: - return True - return False - - -def _wants_ndjson(request: Request) -> bool: - """Content negotiation for list endpoints: NDJSON is an opt-in, JSON is the - default (per spec 6.2: "Default JSON response wraps the same data in a - {items, total, returned, has_more} envelope so the frontend can paginate - uniformly"). - - Used by the GET list routes (/api/batches, /api/claims, /api/remittances, - /api/providers, /api/activity). NDJSON is returned only when the client - explicitly sends ``Accept: application/x-ndjson`` (with or without - ``application/json``). Bare ``*/*``, an empty Accept, or an explicit - ``Accept: application/json`` all return the JSON envelope. - """ - accept = request.headers.get("accept", "") - return "application/x-ndjson" in accept - - -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) - - @app.get("/api/health") def health() -> dict[str, str]: return {"status": "ok", "version": __version__} @@ -362,48 +283,11 @@ def _build_and_persist_ack(batch_id: str) -> dict | None: } -def _ndjson_stream(result: ParseResult) -> Iterator[bytes]: - """Yield one JSON object per line: envelope → claims → summary.""" - envelope_obj = ( - result.envelope.model_dump() if result.envelope is not None else None - ) - yield (json.dumps({"type": "envelope", "data": envelope_obj}) + "\n").encode("utf-8") - for claim in result.claims: - yield (json.dumps({"type": "claim", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8") - - # --------------------------------------------------------------------------- # # 835 ERA (Health Care Claim Payment/Advice) # --------------------------------------------------------------------------- # -def _strict_rewrite_835(result: ParseResult835) -> ParseResult835: - """Promote warnings to errors (mirrors the CLI's --strict).""" - if result.validation is None: - return result - report = result.validation - promoted = [i.model_copy(update={"severity": "error"}) for i in report.warnings] - new_errors = report.errors + promoted - new_report = report.model_copy(update={"errors": new_errors, "passed": not new_errors}) - passed = 1 if new_report.passed else 0 - failed = 1 if not new_report.passed else 0 - new_summary = result.summary.model_copy( - update={"passed": passed, "failed": failed} - ) - return result.model_copy(update={"validation": new_report, "summary": new_summary}) - - -def _drop_raw_segments_835(result: ParseResult835) -> ParseResult835: - """Return a copy of ``result`` with ``raw_segments`` cleared on every claim.""" - claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims] - return result.model_copy(update={"claims": claims}) - - -def _has_835_validation_errors(result: ParseResult835) -> bool: - return result.validation is not None and not result.validation.passed - - def _reconciliation_summary_for_batch(batch_id: str) -> dict: """Return ``{matched, unmatched_claims, unmatched_remittances, skipped}`` for a batch. @@ -530,18 +414,6 @@ async def parse_835_endpoint( ) -def _ndjson_stream_835(result: ParseResult835) -> Iterator[bytes]: - """Yield one JSON object per line: envelope → financial → trace → payer → payee → claim_payments → summary.""" - yield (json.dumps({"type": "envelope", "data": json.loads(result.envelope.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "financial_info", "data": json.loads(result.financial_info.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "trace", "data": json.loads(result.trace.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "payer", "data": json.loads(result.payer.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "payee", "data": json.loads(result.payee.model_dump_json())}) + "\n").encode("utf-8") - for claim in result.claims: - yield (json.dumps({"type": "claim_payment", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8") - - # --------------------------------------------------------------------------- # # 999 ACK (Implementation Acknowledgment) # --------------------------------------------------------------------------- # @@ -1105,7 +977,7 @@ def inbox_dismiss_candidates(body: dict): return {"ok": True, "dismissed_count": len(pairs)} -# --------------------------------------------------------------------------- +# --------------------------------------------------------------------------- # # SP14: Payer-Rejected acknowledge # # Operator hits "Acknowledge" on the Payer-Rejected Inbox lane to clear @@ -1116,7 +988,7 @@ def inbox_dismiss_candidates(body: dict): # Idempotent: re-acknowledging an already-acknowledged claim is a noop # (the timestamp is not bumped). Returns the count actually transitioned # so the UI can show "3 of 5 were already acknowledged". -# --------------------------------------------------------------------------- +# --------------------------------------------------------------------------- # @app.post("/api/inbox/payer-rejected/acknowledge") def inbox_acknowledge_payer_rejected(body: dict): """Mark Payer-Rejected claims as acknowledged by the operator.""" @@ -1274,7 +1146,7 @@ def inbox_resubmit_rejected( @app.get("/api/inbox/export.csv") def inbox_export_csv(lane: str): """Stream a CSV for a single lane.""" - if lane not in {"rejected", "candidates", "unmatched", "done_today", "payer_rejected"}: + if lane not in {"rejected", "candidates", "unmatched", "done_today"}: raise HTTPException(400, f"unknown lane: {lane}") dismissed_pairs = getattr(app.state, "dismissed_pairs", set()) with db.SessionLocal()() as session: @@ -1415,71 +1287,6 @@ def list_claims( # --------------------------------------------------------------------------- # -def _heartbeat_seconds() -> float: - """Return the configured tail heartbeat interval. - - Read from ``CYCLONE_TAIL_HEARTBEAT_S`` at call time so tests can - monkeypatch the env var without reloading the module. Defaults to - 15s (the production cadence); tests override to a small value (e.g. - 0.2s) to keep their runtime bounded. - """ - raw = os.environ.get("CYCLONE_TAIL_HEARTBEAT_S", "15") - try: - v = float(raw) - except ValueError: - return 15.0 - return v if v > 0 else 15.0 - - -async def _tail_events( - request: Request, bus: EventBus, kinds: list[str] -) -> AsyncIterator[bytes]: - """Forward subscribed events as ``item`` lines with periodic heartbeats. - - Polls the underlying ``asyncio.Queue`` directly (via - :meth:`EventBus.subscribe_raw`) instead of awaiting the bus's - async-iterator wrapper. ``asyncio.wait_for`` cancels the inner - future on timeout, which would otherwise terminate the bus - iterator at its ``await`` point and break subsequent - ``__anext__`` calls with ``StopAsyncIteration``. Polling - ``queue.get()`` is idempotent under cancellation, so heartbeats - don't poison the subscription. - - A ``try/finally`` unsubscribes the queue from the bus when the - caller disconnects or the generator is garbage collected — - otherwise the bus would leak one queue per open stream. - """ - hb_s = _heartbeat_seconds() - queue, _sub = bus.subscribe_raw(kinds) - try: - while True: - if await request.is_disconnected(): - return - get_task = asyncio.ensure_future(queue.get()) - sleep_task = asyncio.ensure_future(asyncio.sleep(hb_s)) - try: - done, pending = await asyncio.wait( - {get_task, sleep_task}, - return_when=asyncio.FIRST_COMPLETED, - ) - except BaseException: - get_task.cancel() - sleep_task.cancel() - raise - for t in pending: - t.cancel() - if get_task in done: - event = get_task.result() - yield _ndjson_line({"type": "item", "data": event}) - else: - yield _ndjson_line({ - "type": "heartbeat", - "data": {"ts": utcnow().isoformat().replace("+00:00", "Z")}, - }) - finally: - bus.unsubscribe(queue, kinds) - - @app.get("/api/claims/stream") async def claims_stream( request: Request, diff --git a/backend/src/cyclone/api_helpers.py b/backend/src/cyclone/api_helpers.py new file mode 100644 index 0000000..77f9e3f --- /dev/null +++ b/backend/src/cyclone/api_helpers.py @@ -0,0 +1,247 @@ +"""Shared helpers used by ``cyclone.api`` route handlers. + +Everything in this module is private to the API layer (no business +logic, no DB writes). It collects the cross-cutting concerns that used +to live inline at the top of ``api.py``: + +* NDJSON wire-format primitives (``ndjson_line``, ``ndjson_stream_list``, + ``ndjson_stream_837``, ``ndjson_stream_835``). +* Content negotiation (``client_wants_json``, ``wants_ndjson``). +* Strict / ``raw_segments`` rewrites applied before persisting parsed + 837P and 835 results. +* Validation-error probes for both transactions. +* The shared live-tail async generator (``tail_events``, + ``heartbeat_seconds``) used by every ``/api//stream`` + endpoint. + +Extracted as part of the api.py router split (see /tmp/refactor-cyclone.md). +""" + +from __future__ import annotations + +import asyncio +import json +import os +from datetime import datetime, timezone +from typing import AsyncIterator, Iterator + +from fastapi import Request + +from cyclone.parsers.models import ClaimOutput, ParseResult +from cyclone.parsers.models_835 import ParseResult835 +from cyclone.pubsub import EventBus + + +def utcnow() -> datetime: + """tz-aware UTC ``datetime`` (matches :func:`cyclone.store.utcnow`).""" + return datetime.now(timezone.utc) + + +def ndjson_line(event: dict) -> bytes: + """Serialize one event dict as a single NDJSON line (UTF-8, trailing ``\\n``). + + Used by the live-tail streaming endpoints to emit a uniform wire format + that the frontend ``tail-stream.ts`` parser can split on newlines. + Compact separators keep each line small and avoid ambiguity with embedded + whitespace. + """ + return (json.dumps(event, separators=(",", ":")) + "\n").encode("utf-8") + + +def client_wants_json(request: Request) -> bool: + """Content negotiation: prefer ``application/json`` when the client asks for it. + + NDJSON is the default for browser uploads that don't set ``Accept``. The + frontend opts into JSON via ``Accept: application/json``. + """ + accept = request.headers.get("accept", "") + # If the client mentions JSON at all (and isn't asking for NDJSON + # specifically) treat it as a single-object request. The browser default + # ``*/*`` falls through to NDJSON. + if "application/json" in accept and "application/x-ndjson" not in accept: + return True + return False + + +def wants_ndjson(request: Request) -> bool: + """Content negotiation for list endpoints: NDJSON is an opt-in, JSON is the + default (per spec 6.2: "Default JSON response wraps the same data in a + {items, total, returned, has_more} envelope so the frontend can paginate + uniformly"). + + Used by the GET list routes (/api/batches, /api/claims, /api/remittances, + /api/providers, /api/activity). NDJSON is returned only when the client + explicitly sends ``Accept: application/x-ndjson`` (with or without + ``application/json``). Bare ``*/*``, an empty Accept, or an explicit + ``Accept: application/json`` all return the JSON envelope. + """ + accept = request.headers.get("accept", "") + return "application/x-ndjson" in accept + + +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 ndjson_stream_837(result: ParseResult) -> Iterator[bytes]: + """Yield one JSON object per line: envelope → claims → summary.""" + envelope_obj = ( + result.envelope.model_dump() if result.envelope is not None else None + ) + yield (json.dumps({"type": "envelope", "data": envelope_obj}) + "\n").encode("utf-8") + for claim in result.claims: + yield (json.dumps({"type": "claim", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8") + + +def ndjson_stream_835(result: ParseResult835) -> Iterator[bytes]: + """Yield one JSON object per line: envelope → financial → trace → payer → payee → claim_payments → summary.""" + yield (json.dumps({"type": "envelope", "data": json.loads(result.envelope.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "financial_info", "data": json.loads(result.financial_info.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "trace", "data": json.loads(result.trace.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "payer", "data": json.loads(result.payer.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "payee", "data": json.loads(result.payee.model_dump_json())}) + "\n").encode("utf-8") + for claim in result.claims: + yield (json.dumps({"type": "claim_payment", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8") + + +def strict_rewrite_837(result: ParseResult) -> ParseResult: + """Promote warnings to errors (mirrors the CLI's --strict).""" + claims: list[ClaimOutput] = [] + for claim in result.claims: + promoted = [ + issue.model_copy(update={"severity": "error"}) + for issue in claim.validation.warnings + ] + new_errors = claim.validation.errors + promoted + claims.append( + claim.model_copy( + update={ + "validation": claim.validation.model_copy( + update={"errors": new_errors, "passed": not new_errors} + ) + } + ) + ) + passed = sum(1 for c in claims if c.validation.passed) + failed = len(claims) - passed + summary = result.summary.model_copy( + update={ + "passed": passed, + "failed": failed, + "failed_claim_ids": [c.claim_id for c in claims if not c.validation.passed], + } + ) + return result.model_copy(update={"claims": claims, "summary": summary}) + + +def strict_rewrite_835(result: ParseResult835) -> ParseResult835: + """Promote warnings to errors (mirrors the CLI's --strict).""" + if result.validation is None: + return result + report = result.validation + promoted = [i.model_copy(update={"severity": "error"}) for i in report.warnings] + new_errors = report.errors + promoted + new_report = report.model_copy(update={"errors": new_errors, "passed": not new_errors}) + passed = 1 if new_report.passed else 0 + failed = 1 if not new_report.passed else 0 + new_summary = result.summary.model_copy( + update={"passed": passed, "failed": failed} + ) + return result.model_copy(update={"validation": new_report, "summary": new_summary}) + + +def drop_raw_segments_837(result: ParseResult) -> ParseResult: + """Return a copy of ``result`` with ``raw_segments`` cleared on every claim.""" + claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims] + return result.model_copy(update={"claims": claims}) + + +def drop_raw_segments_835(result: ParseResult835) -> ParseResult835: + """Return a copy of ``result`` with ``raw_segments`` cleared on every claim.""" + claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims] + return result.model_copy(update={"claims": claims}) + + +def has_claim_validation_errors(result: ParseResult) -> bool: + return any(not c.validation.passed for c in result.claims) + + +def has_835_validation_errors(result: ParseResult835) -> bool: + return result.validation is not None and not result.validation.passed + + +def heartbeat_seconds() -> float: + """Return the configured tail heartbeat interval. + + Read from ``CYCLONE_TAIL_HEARTBEAT_S`` at call time so tests can + monkeypatch the env var without reloading the module. Defaults to + 15s (the production cadence); tests override to a small value (e.g. + 0.2s) to keep their runtime bounded. + """ + raw = os.environ.get("CYCLONE_TAIL_HEARTBEAT_S", "15") + try: + v = float(raw) + except ValueError: + return 15.0 + return v if v > 0 else 15.0 + + +async def tail_events( + request: Request, bus: EventBus, kinds: list[str] +) -> AsyncIterator[bytes]: + """Forward subscribed events as ``item`` lines with periodic heartbeats. + + Polls the underlying ``asyncio.Queue`` directly (via + :meth:`EventBus.subscribe_raw`) instead of awaiting the bus's + async-iterator wrapper. ``asyncio.wait_for`` cancels the inner + future on timeout, which would otherwise terminate the bus + iterator at its ``await`` point and break subsequent + ``__anext__`` calls with ``StopAsyncIteration``. Polling + ``queue.get()`` is idempotent under cancellation, so heartbeats + don't poison the subscription. + + A ``try/finally`` unsubscribes the queue from the bus when the + caller disconnects or the generator is garbage collected — + otherwise the bus would leak one queue per open stream. + """ + hb_s = heartbeat_seconds() + queue, _sub = bus.subscribe_raw(kinds) + try: + while True: + if await request.is_disconnected(): + return + get_task = asyncio.ensure_future(queue.get()) + sleep_task = asyncio.ensure_future(asyncio.sleep(hb_s)) + try: + done, pending = await asyncio.wait( + {get_task, sleep_task}, + return_when=asyncio.FIRST_COMPLETED, + ) + except BaseException: + get_task.cancel() + sleep_task.cancel() + raise + for t in pending: + t.cancel() + if get_task in done: + event = get_task.result() + yield ndjson_line({"type": "item", "data": event}) + else: + yield ndjson_line({ + "type": "heartbeat", + "data": {"ts": utcnow().isoformat().replace("+00:00", "Z")}, + }) + finally: + bus.unsubscribe(queue, kinds)