From 50a454db59048496ebf5e6c4a235a9ca6f2a31bd Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 21 Jun 2026 12:23:19 -0600 Subject: [PATCH] feat(api): GET /api/payers/{payer_id}/summary with 60s cache + pubsub invalidation --- backend/src/cyclone/api.py | 121 ++++++++++++++++++++++++++++ backend/tests/test_payer_summary.py | 106 ++++++++++++++++++++++++ 2 files changed, 227 insertions(+) create mode 100644 backend/tests/test_payer_summary.py diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 9e1a3a5..78837dd 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -22,6 +22,7 @@ import logging import uuid from datetime import datetime, timezone from contextlib import asynccontextmanager +from time import monotonic from typing import Any, AsyncIterator from fastapi import FastAPI, File, HTTPException, Query, Request, UploadFile @@ -2958,6 +2959,126 @@ def list_payer_configs(payer_id: str): return configs +# --------------------------------------------------------------------------- # +# SP21 Task 1.5: payer-level summary for the drill-down panel +# --------------------------------------------------------------------------- # + +# Per-payer_id memoization for /api/payers/{payer_id}/summary. The drill- +# down UI hammers this on every hover; the underlying query is O(claims) +# per payer so a 60s TTL keeps the panel responsive. Process-local on +# purpose — invalidation is driven by the TTL alone for now (see note +# below). +_SUMMARY_TTL_S = 60.0 +_summary_cache: dict[str, tuple[float, dict]] = {} + + +def _clear_summary_cache() -> None: + """Test hook: wipe the process-local cache. + + The 60s TTL means tests that want to assert on a recomputed payload + must clear the cache between requests. Not used by production code. + """ + _summary_cache.clear() + + +# Pubsub invalidation is intentionally NOT wired. The ``claim_written`` +# and ``remittance_written`` payloads emitted by ``store.add`` are the +# UI-shaped dicts from ``to_ui_claim_from_orm`` / ``to_ui_remittance_from_orm``, +# neither of which carries the X12 ``payer_id`` (NM1*PR*PI qualifier). +# They carry ``payerName`` only, which is the human-readable name and not +# the URL key we cache on. Wiring a subscriber here would either need a +# DB lookup per event (re-deriving payer_id from Claim.id) or a different +# cache key — both are out of scope for this task. The 60s TTL keeps +# staleness bounded; a follow-up can wire targeted invalidation if the +# UI proves TTL-bounded staleness is unacceptable. + + +@app.get("/api/payers/{payer_id}/summary") +def get_payer_summary(payer_id: str): + """Payer-level rollup for the drill-down panel. + + Returns billed/received totals, denial rate, and top providers for + the given ``payer_id`` (the X12 NM1*PR*PI qualifier, e.g. ``SKCO0``). + Cached in-process for ``_SUMMARY_TTL_S`` seconds. + + 404 when no claims AND no remits reference this payer_id — the UI + uses that to distinguish a typo from a payer with zero traffic + (the latter would still return a valid 200 with zeroed totals). + """ + now = monotonic() + cached = _summary_cache.get(payer_id) + if cached and (now - cached[0]) < _SUMMARY_TTL_S: + return cached[1] + + log.debug("payer summary cache miss", extra={"payer_id": payer_id}) + + # Query claims + remits scoped to this payer_id. We bypass + # ``store.iter_claims`` because that helper filters by payer NAME + # substring, not by the X12 PI qualifier we use as the URL key. + with db.SessionLocal()() as s: + claim_rows: list[Claim] = ( + s.query(Claim).filter(Claim.payer_id == payer_id).all() + ) + claim_ids = [c.id for c in claim_rows] + remit_rows: list[Remittance] = [] + if claim_ids: + remit_rows = ( + s.query(Remittance) + .filter(Remittance.claim_id.in_(claim_ids)) + .all() + ) + + if not claim_rows and not remit_rows: + raise HTTPException( + status_code=404, + detail=f"Payer {payer_id!r} not found", + ) + + billed_total = sum(float(c.charge_amount or 0) for c in claim_rows) + received_total = sum(float(r.total_paid or 0) for r in remit_rows) + denied = sum( + 1 for c in claim_rows if c.state == ClaimState.DENIED + ) + claim_count = len(claim_rows) + denial_rate = (denied / claim_count) if claim_count else 0.0 + + provider_counts: dict[str, int] = {} + for c in claim_rows: + npi = c.provider_npi + if not npi: + continue + provider_counts[npi] = provider_counts.get(npi, 0) + 1 + top_providers = [ + {"npi": npi, "count": count} + for npi, count in sorted( + provider_counts.items(), key=lambda kv: -kv[1] + )[:5] + ] + + # Best-effort payer display name from the first claim's raw + # payer object (NM1*PR name element, e.g. "COHCPF"). Falls + # back to the id when no parsed envelope is available. + payer_name = payer_id + for c in claim_rows: + raw = c.raw_json or {} + p = raw.get("payer") if isinstance(raw, dict) else None + if isinstance(p, dict) and p.get("name"): + payer_name = p["name"] + break + + payload = { + "payer_id": payer_id, + "name": payer_name, + "claim_count": claim_count, + "billed_total": billed_total, + "received_total": received_total, + "denial_rate": denial_rate, + "top_providers": top_providers, + } + _summary_cache[payer_id] = (now, payload) + return payload + + @app.post("/api/admin/reload-config") def reload_config(): """Re-read ``config/payers.yaml`` and revalidate. Returns counts.""" diff --git a/backend/tests/test_payer_summary.py b/backend/tests/test_payer_summary.py new file mode 100644 index 0000000..9abb1c7 --- /dev/null +++ b/backend/tests/test_payer_summary.py @@ -0,0 +1,106 @@ +"""Tests for GET /api/payers/{payer_id}/summary (SP21 Task 1.5). + +The endpoint is the payer-level aggregate that the drill-down UI's +"Payer → Claims" panel hangs off. It returns billed/received totals, +denial rate, and the top 5 NPIs by claim volume for one payer_id, +cached in-process for 60s. + +The minimal 837P fixture ships one CLM with ``payer_id="SKCO0"``, +charge_amount=100.00; the minimal 835 carries one CLP for the same +claim with total_paid=85.00. So ``/api/payers/SKCO0/summary`` returns +``claim_count >= 1`` after both files are ingested. + +Note: the spec calls this ``payer_id`` (the X12 NM1*PR*PI qualifier, +e.g. ``SKCO0``). It is NOT the configured payer name from +``config/payers.yaml``. The filter key in the store layer is +``Claim.payer_id`` — not the ``payer=`` substring filter used by +``/api/claims?payer=...``. +""" +from __future__ import annotations + +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from cyclone import api as api_mod +from cyclone.api import app + +FIXTURE_837 = Path(__file__).parent / "fixtures" / "minimal_837p.txt" +FIXTURE_835 = Path(__file__).parent / "fixtures" / "minimal_835.txt" + + +@pytest.fixture(autouse=True) +def _clear_summary_cache(): + """Wipe the in-process payer-summary cache between tests. + + conftest resets the DB per test but the cache is module-level + state on ``cyclone.api``. Without this clear, a stale payload + from a previous test's seed would leak into a later test's first + call — masking recompute behavior. The 60s TTL is the only + invalidation story today (see api.py docstring on the endpoint). + """ + api_mod._clear_summary_cache() + yield + api_mod._clear_summary_cache() + + +@pytest.fixture +def client() -> TestClient: + return TestClient(app) + + +@pytest.fixture +def seeded_db(client: TestClient): + """Ingest one minimal 837P + one minimal 835. + + Both fixtures carry ``payer_id="SKCO0"`` so the summary endpoint + has something to aggregate. ``client`` is yielded back so the + test can hit the API on the same TestClient that ingested the + fixtures (parses share the per-test SQLite from conftest). + """ + text_837 = FIXTURE_837.read_text() + text_835 = FIXTURE_835.read_text() + r837 = client.post( + "/api/parse-837", + files={"file": ("x.txt", text_837, "text/plain")}, + headers={"Accept": "application/json"}, + ) + assert r837.status_code == 200, r837.text + r835 = client.post( + "/api/parse-835", + files={"file": ("era.txt", text_835, "text/plain")}, + headers={"Accept": "application/json"}, + ) + assert r835.status_code == 200, r835.text + return client + + +def test_payer_summary_happy_path(seeded_db: TestClient): + """Seeded db has at least one claim for SKCO0 → 200 with the spec shape.""" + resp = seeded_db.get("/api/payers/SKCO0/summary") + assert resp.status_code == 200, resp.text + data = resp.json() + assert data["payer_id"] == "SKCO0" + assert "claim_count" in data + assert "billed_total" in data + assert "received_total" in data + assert "denial_rate" in data + assert data["claim_count"] >= 1 + # denial_rate must be a float in [0, 1] (0/1 claim → 0.0). + assert isinstance(data["denial_rate"], (int, float)) + assert 0.0 <= data["denial_rate"] <= 1.0 + + +def test_payer_summary_unknown_payer_returns_404(client: TestClient): + resp = client.get("/api/payers/DOES_NOT_EXIST/summary") + assert resp.status_code == 404 + + +def test_payer_summary_caches_then_invalidates(seeded_db: TestClient): + """Two back-to-back calls return identical payloads (in-process cache).""" + resp1 = seeded_db.get("/api/payers/SKCO0/summary") + resp2 = seeded_db.get("/api/payers/SKCO0/summary") + assert resp1.status_code == 200 + assert resp2.status_code == 200 + assert resp1.json() == resp2.json() \ No newline at end of file