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

feat(store): get_claim_detail returns full claim context with history + matched remit

Browse Source
This commit is contained in:
Tyler
2026-06-20 09:21:07 -06:00
parent bdc851f0f0
commit 423415ab64
2 changed files with 713 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/store.py
+287
View File
@@ -417,6 +417,203 @@ def to_ui_claim_from_orm(
}
# Max number of ActivityEvent rows surfaced in the detail drawer's
# state history. The spec caps it at 50; a higher claim volume (manual
# match/unmatch thrash) just shows the 50 most recent. Exposed as a
# module constant so the endpoint layer can pass it through as a
# default if it ever supports a `?limit=N` query param.
CLAIM_DETAIL_HISTORY_LIMIT = 50
def _iso_z(value: datetime | None) -> str:
"""Format a tz-aware-or-naive UTC datetime as ISO-8601 with trailing Z.
The DB columns are declared ``DateTime(timezone=True)`` and rows are
stored UTC at write time, but SQLite drops the tzinfo on read
(returning a naive ``datetime``). Re-attach UTC for naive values
so the spec contract holds: every ISO datetime field ends in Z.
"""
if value is None:
return ""
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
return value.isoformat().replace("+00:00", "Z")
def _address_to_ui(addr: dict | None) -> dict:
"""Render a raw ``Address`` dict in the spec's parties address shape.
Returns an empty dict when the source is missing so the UI can
branch on the field's presence rather than the value. The spec
shape is ``{line1, line2|null, city, state, zip}``.
"""
if not addr:
return {}
return {
"line1": addr.get("line1") or "",
"line2": addr.get("line2"),
"city": addr.get("city") or "",
"state": addr.get("state") or "",
"zip": addr.get("zip") or "",
}
def _validation_issues_to_ui(issues: list[dict] | None) -> list[dict]:
"""Project ValidationIssue dicts onto the spec's per-issue shape.
Source includes ``segment_index`` (a parser debug aid) which the
spec doesn't surface; we drop it. The endpoint contract is
``{rule, severity, message}`` per issue.
"""
if not issues:
return []
return [
{
"rule": issue.get("rule", ""),
"severity": issue.get("severity", "error"),
"message": issue.get("message", ""),
}
for issue in issues
]
def to_ui_claim_detail(
row: Claim,
*,
batch_id: str,
parsed_at: datetime,
) -> dict:
"""Map an ORM ``Claim`` row to the SP4 detail-drawer UI shape.
A superset of :func:`to_ui_claim_from_orm`: same top-level identity
fields, plus the full parties / validation / service-lines /
diagnoses / raw-segments / service-date / state-label payload that
the drawer needs. ``matchedRemittance`` and ``stateHistory`` are
*not* filled in here — they require extra queries and are stitched
in by :meth:`CycloneStore.get_claim_detail`.
The mapper is deliberately a pure function (no DB I/O) so the
endpoint layer can call it from a worker thread or swap the
history/remittance sources for tests without re-implementing the
body.
"""
raw = row.raw_json or {}
bp = raw.get("billing_provider", {}) or {}
payer_obj = raw.get("payer", {}) or {}
sub = raw.get("subscriber", {}) or {}
service_lines = raw.get("service_lines", []) or []
diagnoses = raw.get("diagnoses", []) or []
validation = raw.get("validation", {}) or {}
raw_segments = raw.get("raw_segments", []) or []
parsed_iso = parsed_at.isoformat().replace("+00:00", "Z")
state_value = (
row.state.value if hasattr(row.state, "value") else str(row.state)
)
# Service dates come from the dedicated ORM columns (denormalized at
# ingest in _claim_837_row so the list views can sort/filter on
# them without a JSON parse). ``isoformat()`` on a ``date`` gives
# ``YYYY-MM-DD`` — the spec shape.
service_date_from_iso = (
row.service_date_from.isoformat() if row.service_date_from else None
)
service_date_to_iso = (
row.service_date_to.isoformat() if row.service_date_to else None
)
return {
# -- identity + state -----------------------------------------
"id": row.id,
"batchId": batch_id,
"state": state_value,
"stateLabel": state_value.capitalize(),
# -- money + dates --------------------------------------------
"billedAmount": float(row.charge_amount or 0),
"serviceDateFrom": service_date_from_iso,
"serviceDateTo": service_date_to_iso,
"submissionDate": parsed_iso,
"parsedAt": parsed_iso,
# -- patient / provider / payer -------------------------------
"patientName": (
f"{sub.get('first_name', '')} {sub.get('last_name', '')}".strip()
),
"providerNpi": bp.get("npi") or row.provider_npi or "",
"providerName": bp.get("name") or "",
"payerName": payer_obj.get("name") or "",
"payerId": payer_obj.get("id") or row.payer_id or "",
# -- diagnoses ------------------------------------------------
"diagnoses": [
{
"code": d.get("code", ""),
"qualifier": d.get("qualifier"),
}
for d in diagnoses
],
# -- service lines --------------------------------------------
# ``service_lines[i].procedure`` is a nested dict in the
# serialized raw_json; the spec flattens it into the line.
# ``charge`` and ``units`` are stored as Decimal via Pydantic
# and serialized to string — coerce defensively. ``modifiers``
# defaults to [] so the UI doesn't have to handle null.
"serviceLines": [
{
"lineNumber": sl.get("line_number"),
"procedureQualifier": (
sl.get("procedure", {}).get("qualifier", "") or ""
),
"procedureCode": (
sl.get("procedure", {}).get("code", "") or ""
),
"modifiers": list(
sl.get("procedure", {}).get("modifiers") or []
),
"charge": float(sl.get("charge") or 0),
"units": (
float(sl["units"])
if sl.get("units") is not None
else None
),
"unitType": sl.get("unit_type"),
"serviceDate": sl.get("service_date"),
}
for sl in service_lines
],
# -- parties --------------------------------------------------
"parties": {
"billingProvider": {
"name": bp.get("name") or "",
"npi": bp.get("npi") or "",
"taxId": bp.get("tax_id") or "",
"address": _address_to_ui(bp.get("address")),
},
"subscriber": {
"firstName": sub.get("first_name") or "",
"lastName": sub.get("last_name") or "",
"memberId": sub.get("member_id") or "",
"dob": sub.get("dob"),
"gender": sub.get("gender"),
},
"payer": {
"name": payer_obj.get("name") or "",
"id": payer_obj.get("id") or "",
},
},
# -- validation ----------------------------------------------
"validation": {
"passed": bool(validation.get("passed", True)),
"errors": _validation_issues_to_ui(validation.get("errors")),
"warnings": _validation_issues_to_ui(validation.get("warnings")),
},
# -- raw segments (debug aid) --------------------------------
"rawSegments": raw_segments,
# -- matched remittance (filled by get_claim_detail) ---------
"matchedRemittance": None,
# -- state history (filled by get_claim_detail) --------------
"stateHistory": [],
}
def to_ui_remittance_from_orm(
row: Remittance,
*,
@@ -852,6 +1049,96 @@ class CycloneStore:
cas_rows=cas_rows,
)
def get_claim_detail(self, claim_id: str) -> dict | None:
"""Return the SP4 detail-drawer shape for one claim, or ``None``.
Drives ``GET /api/claims/{claim_id}``. Returns the spec-shaped
dict from :func:`to_ui_claim_detail` (header + state + parties +
validation + service lines + diagnoses + raw segments) stitched
with the claim's recent activity history and, if paired, a
matched-remittance summary.
Returns ``None`` when ``claim_id`` is not in the DB so the API
layer can map that to a 404 — the URL-driven drawer
distinguishes "claim doesn't exist" from "fetch failed" (the
spec §3.4 calls for a distinct 404 state in the drawer).
The history is capped at :data:`CLAIM_DETAIL_HISTORY_LIMIT`
(50, per the spec) and ordered ``ts DESC`` so the most recent
event is first. The status string in ``matchedRemittance``
follows the same ``reconciled``/``received`` mapping used by
:func:`to_ui_remittance_from_orm`.
"""
# Lazy import — same pattern used throughout this module to
# avoid a circular store ↔ db import on cold start.
from cyclone import db as _db
with _db.SessionLocal()() as s:
row = s.get(Claim, claim_id)
if row is None:
return None
history_rows = (
s.query(ActivityEvent)
.filter(ActivityEvent.claim_id == claim_id)
.order_by(ActivityEvent.ts.desc())
.limit(CLAIM_DETAIL_HISTORY_LIMIT)
.all()
)
parsed_at = (
row.batch.parsed_at
if row.batch is not None
else None
)
if parsed_at is not None and parsed_at.tzinfo is None:
parsed_at = parsed_at.replace(tzinfo=timezone.utc)
detail = to_ui_claim_detail(
row,
batch_id=row.batch_id,
parsed_at=parsed_at,
)
detail["stateHistory"] = [
{
"kind": ev.kind,
# SQLite drops tzinfo on read; rows are stored UTC
# at write time (see ``add`` / ``manual_match``),
# so re-attach UTC if needed to keep the spec
# contract that ``ts`` ends in Z.
"ts": _iso_z(ev.ts),
"batchId": ev.batch_id,
"remittanceId": ev.remittance_id,
}
for ev in history_rows
]
if row.matched_remittance_id is not None:
remit = s.get(Remittance, row.matched_remittance_id)
if remit is not None:
status = (
"reconciled"
if remit.status_code in ("21", "22")
else "received"
)
detail["matchedRemittance"] = {
"id": remit.id,
"totalPaid": float(remit.total_paid or 0),
"status": status,
"receivedAt": _iso_z(remit.received_at),
}
# If the remittance was deleted out from under the FK
# (the FK is ``ON DELETE SET NULL`` so the column is
# already cleared in normal flow), the matched_remittance_id
# would be None here and we wouldn't enter this branch.
# If the FK is non-null but the row is gone (e.g. tests
# that bypass the cascade), fall through with the
# default ``None`` — the UI shows "no match" rather
# than crashing.
return detail
def all(self) -> list[BatchRecord]:
"""Return every ``BatchRecord``, oldest first (no pagination)."""
with db.SessionLocal()() as s:
backend/tests/test_store_claim_detail.py
+426
View File
@@ -0,0 +1,426 @@
"""Tests for ``CycloneStore.get_claim_detail`` (SP4 Task 1).
The method is the read path behind the ``GET /api/claims/{id}`` endpoint
that drives the per-claim detail drawer. The response shape is locked in
``docs/superpowers/specs/2026-06-20-cyclone-claim-drawer-design.md`` §3.2.
What this exercises:
* ``test_get_claim_detail_returns_none_for_missing`` — a claim id that
isn't in the DB must return ``None`` (the API maps that to 404).
* ``test_get_claim_detail_happy_path_returns_full_shape`` — a fully
populated claim (diagnoses, service lines, parties, validation,
raw segments) round-trips into the spec dict.
* ``test_get_claim_detail_includes_state_history`` — the
``stateHistory`` array surfaces ``ActivityEvent`` rows for this
claim, newest first, with both batch and remittance associations
preserved.
* ``test_get_claim_detail_includes_matched_remittance_summary`` — a
claim that has been paired with a remittance via ``manual_match``
surfaces a ``matchedRemittance`` summary with id / totalPaid /
status / receivedAt.
* ``test_get_claim_detail_no_matched_remittance_returns_null_field``
— an unmatched claim returns ``matchedRemittance: null`` (not
missing, not an empty dict).
"""
from __future__ import annotations
from datetime import date, datetime, timezone
from decimal import Decimal
import pytest
from cyclone import db
from cyclone.parsers.models import (
Address,
BatchSummary,
BillingProvider,
ClaimHeader,
ClaimOutput,
Diagnosis,
Envelope,
Payer,
ParseResult,
Procedure,
ServiceLine,
Subscriber,
ValidationIssue,
ValidationReport,
)
from cyclone.parsers.models_835 import (
ClaimAdjustment,
ClaimPayment,
Envelope as Envelope835,
FinancialInfo,
ParseResult835,
Payer835,
Payee835,
ReassociationTrace,
ServicePayment,
BatchSummary as BatchSummary835,
)
from cyclone.store import BatchRecord837, BatchRecord835, CycloneStore
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _setup(tmp_path, monkeypatch):
"""Per-test DB init — matches the pattern in test_store_reconcile.py.
Defensive: conftest.py already provides an autouse fixture, but we
declare it here too so this file is self-contained and matches the
convention used by the other store-level test modules.
"""
monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
db._reset_for_tests()
db.init_db()
yield
db._reset_for_tests()
# ---------------------------------------------------------------------------
# Builders
# ---------------------------------------------------------------------------
def _make_rich_claim_837(claim_id: str = "CLM-1") -> ClaimOutput:
"""Build a ClaimOutput with every section populated.
Drives the happy-path test: the detail mapper should round-trip
diagnoses, service lines (with procedure+modifiers+charge+units),
parties (billing_provider with address, subscriber with dob+gender,
payer with id), validation issues, and raw_segments.
"""
bp_addr = Address(
line1="123 Main St", line2="Suite 400",
city="Denver", state="CO", zip="80202",
)
sub_addr = Address(
line1="456 Oak Ave", line2=None,
city="Boulder", state="CO", zip="80301",
)
return ClaimOutput(
claim_id=claim_id,
control_number="0001",
transaction_date=date(2026, 6, 11),
billing_provider=BillingProvider(
name="Test Provider", npi="1234567890",
tax_id="99-1234567", address=bp_addr,
),
subscriber=Subscriber(
first_name="John", last_name="Doe", member_id="M-001",
dob=date(1980, 1, 1), gender="M", address=sub_addr,
),
payer=Payer(name="COHCPF", id="SKCO0"),
claim=ClaimHeader(
claim_id=claim_id, total_charge=Decimal("100.00"),
frequency_code="1", place_of_service="11",
),
diagnoses=[Diagnosis(code="Z00", qualifier="ABK")],
service_lines=[
ServiceLine(
line_number=1,
procedure=Procedure(
qualifier="HC", code="99213",
modifiers=["25"],
),
charge=Decimal("80.00"),
unit_type="UN",
units=Decimal("1.0"),
service_date=date(2026, 6, 11),
),
ServiceLine(
line_number=2,
procedure=Procedure(qualifier="HC", code="85025"),
charge=Decimal("20.00"),
unit_type="UN",
units=Decimal("1.0"),
service_date=date(2026, 6, 11),
),
],
validation=ValidationReport(
passed=False,
errors=[
ValidationIssue(
rule="R050_diagnosis_present",
severity="error",
message="At least one diagnosis code is required",
),
],
warnings=[],
),
raw_segments=[["ISA", "*00*"], ["CLM", claim_id, "*100***"]],
)
def _add_837_with_claim(s: CycloneStore, claim_id: str = "CLM-1") -> str:
"""Add a single 837 claim via the store. Returns the batch id."""
co = _make_rich_claim_837(claim_id)
parsed_at = datetime(2026, 6, 11, 12, 0, tzinfo=timezone.utc)
pr = ParseResult(
envelope=Envelope(
sender_id="S", receiver_id="R", control_number="0001",
transaction_date=date(2026, 6, 11),
),
claims=[co],
summary=BatchSummary(
input_file="c.txt", control_number="0001",
transaction_date=date(2026, 6, 11),
total_claims=1, passed=0, failed=1,
),
)
batch_id = f"b-837-{claim_id}"
s.add(BatchRecord837(
id=batch_id, kind="837p", input_filename="c.txt",
parsed_at=parsed_at, result=pr,
))
return batch_id
def _add_835_with_remit(
s: CycloneStore,
pcn: str,
charge: str = "100",
paid: str = "100",
status: str = "1",
) -> str:
"""Add a single 835 remittance. Returns the batch id."""
cp = ClaimPayment(
payer_claim_control_number=pcn,
status_code=status,
status_label="Primary",
total_charge=charge, total_paid=paid,
service_payments=[
ServicePayment(
line_number=1, procedure_qualifier="HC",
procedure_code="99213",
charge=charge, payment=paid,
adjustments=[],
),
],
)
parsed_at = datetime(2026, 6, 12, 10, 0, tzinfo=timezone.utc)
batch_id = f"b-835-{pcn}"
s.add(BatchRecord835(
id=batch_id, kind="835", input_filename="era.txt",
parsed_at=parsed_at,
result=ParseResult835(
envelope=Envelope835(
sender_id="P", receiver_id="R", control_number="0002",
transaction_date=date(2026, 6, 12),
),
financial_info=FinancialInfo(
handling_code="C", paid_amount=Decimal("0"),
credit_debit_flag="C", payment_method=None,
),
trace=ReassociationTrace(
trace_type_code="1", trace_number="0001",
originating_company_id="P",
),
payer=Payer835(name="COHCPF", id="SKCO0"),
payee=Payee835(name="Test Provider", npi="1234567890"),
claims=[cp],
summary=BatchSummary835(
input_file="era.txt", control_number="0002",
transaction_date=date(2026, 6, 12),
total_claims=1, passed=1, failed=0,
),
),
))
return batch_id
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_get_claim_detail_returns_none_for_missing():
"""An unknown claim id must return ``None`` (not raise, not empty dict)."""
s = CycloneStore()
assert s.get_claim_detail("CLM-DOES-NOT-EXIST") is None
def test_get_claim_detail_happy_path_returns_full_shape():
"""A fully populated claim round-trips into the spec-shaped dict."""
s = CycloneStore()
batch_id = _add_837_with_claim(s, "CLM-1")
parsed_at = datetime(2026, 6, 11, 12, 0, tzinfo=timezone.utc)
out = s.get_claim_detail("CLM-1")
assert out is not None
# -- top-level identity -----------------------------------------
assert out["id"] == "CLM-1"
assert out["batchId"] == batch_id
# SP1 ingest stamps every claim as submitted; detail shape preserves it.
assert out["state"] == "submitted"
# stateLabel is a human-readable capitalization of state.
assert out["stateLabel"] == "Submitted"
# -- money + dates ----------------------------------------------
assert out["billedAmount"] == 100.0
assert isinstance(out["billedAmount"], float)
assert out["submissionDate"] == "2026-06-11T12:00:00Z"
assert out["parsedAt"] == "2026-06-11T12:00:00Z"
assert out["serviceDateFrom"] == "2026-06-11"
assert out["serviceDateTo"] == "2026-06-11"
# -- patient + provider + payer ---------------------------------
assert out["patientName"] == "John Doe"
assert out["providerNpi"] == "1234567890"
assert out["providerName"] == "Test Provider"
assert out["payerName"] == "COHCPF"
assert out["payerId"] == "SKCO0"
# -- diagnoses --------------------------------------------------
assert out["diagnoses"] == [{"code": "Z00", "qualifier": "ABK"}]
# -- service lines (procedure.flattened shape) ------------------
assert len(out["serviceLines"]) == 2
sl1 = out["serviceLines"][0]
assert sl1["lineNumber"] == 1
assert sl1["procedureQualifier"] == "HC"
assert sl1["procedureCode"] == "99213"
assert sl1["modifiers"] == ["25"]
assert sl1["charge"] == 80.0
assert sl1["units"] == 1.0
assert sl1["unitType"] == "UN"
assert sl1["serviceDate"] == "2026-06-11"
# Second line has no modifiers — must be an empty list, not None.
sl2 = out["serviceLines"][1]
assert sl2["modifiers"] == []
assert sl2["procedureCode"] == "85025"
assert sl2["charge"] == 20.0
# -- parties ----------------------------------------------------
bp = out["parties"]["billingProvider"]
assert bp["name"] == "Test Provider"
assert bp["npi"] == "1234567890"
assert bp["taxId"] == "99-1234567"
assert bp["address"] == {
"line1": "123 Main St",
"line2": "Suite 400",
"city": "Denver",
"state": "CO",
"zip": "80202",
}
sub = out["parties"]["subscriber"]
assert sub["firstName"] == "John"
assert sub["lastName"] == "Doe"
assert sub["memberId"] == "M-001"
assert sub["dob"] == "1980-01-01"
assert sub["gender"] == "M"
payer = out["parties"]["payer"]
assert payer == {"name": "COHCPF", "id": "SKCO0"}
# -- validation -------------------------------------------------
v = out["validation"]
assert v["passed"] is False
assert len(v["errors"]) == 1
err = v["errors"][0]
assert err["rule"] == "R050_diagnosis_present"
assert err["severity"] == "error"
assert "diagnosis" in err["message"].lower()
assert v["warnings"] == []
# -- raw segments ----------------------------------------------
assert out["rawSegments"] == [
["ISA", "*00*"],
["CLM", "CLM-1", "*100***"],
]
# -- unmatched (this test doesn't pair anything) ---------------
assert out["matchedRemittance"] is None
# -- state history: at minimum the claim_submitted event -------
history = out["stateHistory"]
assert isinstance(history, list)
assert any(ev["kind"] == "claim_submitted" for ev in history), history
# The submitted event must carry the batch id, not a remittance id.
submitted = next(ev for ev in history if ev["kind"] == "claim_submitted")
assert submitted["batchId"] == batch_id
assert submitted["remittanceId"] is None
# ts is ISO Z.
assert submitted["ts"].endswith("Z")
def test_get_claim_detail_includes_state_history():
"""``stateHistory`` must include the manual_match event after pairing."""
s = CycloneStore()
_add_837_with_claim(s, "CLM-1")
# 835 with PCN that intentionally differs from the 837's member id so
# reconcile doesn't auto-pair on ingest — we want manual_match to do it.
_add_835_with_remit(s, pcn="CLM-1", paid="100", status="1")
s.manual_match("CLM-1", "CLM-1")
out = s.get_claim_detail("CLM-1")
assert out is not None
history = out["stateHistory"]
kinds = [ev["kind"] for ev in history]
# manual_match must appear; claim_submitted must also appear.
assert "manual_match" in kinds, history
assert "claim_submitted" in kinds, history
# Ordering: ts DESC — manual_match was inserted after the submitted
# event, so it must come first.
assert kinds.index("manual_match") < kinds.index("claim_submitted"), (
f"stateHistory not ordered DESC by ts: {history}"
)
# The manual_match event must carry the remittanceId. It also
# carries the remittance's batch_id (per the manual_match
# implementation) — the mapper passes through whatever is in the
# ActivityEvent row, which is correct per the spec's
# ``{kind, ts, batchId|null, remittanceId|null}`` contract.
mm = next(ev for ev in history if ev["kind"] == "manual_match")
assert mm["remittanceId"] == "CLM-1"
assert mm["batchId"] is not None
assert mm["ts"].endswith("Z")
# The claim_submitted event has no remittance_id (only a batch id),
# which is the opposite of manual_match.
submitted = next(ev for ev in history if ev["kind"] == "claim_submitted")
assert submitted["remittanceId"] is None
assert submitted["batchId"] is not None
def test_get_claim_detail_includes_matched_remittance_summary():
"""A paired claim surfaces a ``matchedRemittance`` summary block."""
s = CycloneStore()
_add_837_with_claim(s, "CLM-1")
_add_835_with_remit(s, pcn="CLM-1", paid="100", status="1")
s.manual_match("CLM-1", "CLM-1")
out = s.get_claim_detail("CLM-1")
assert out is not None
mr = out["matchedRemittance"]
assert mr is not None
assert mr["id"] == "CLM-1"
assert mr["totalPaid"] == 100.0
assert isinstance(mr["totalPaid"], float)
# status_code "1" → "received" (per to_ui_remittance_from_orm mapping).
assert mr["status"] == "received"
# receivedAt is ISO Z (the remit's received_at is tz-aware UTC).
assert mr["receivedAt"].endswith("Z")
# Sanity: the receivedAt parses back to a real timestamp.
parsed = datetime.fromisoformat(mr["receivedAt"].replace("Z", "+00:00"))
assert parsed.tzinfo is not None
def test_get_claim_detail_no_matched_remittance_returns_null_field():
"""An unmatched claim has ``matchedRemittance: null`` (not missing, not {})."""
s = CycloneStore()
_add_837_with_claim(s, "CLM-1")
# No 835 added — claim is unmatched.
out = s.get_claim_detail("CLM-1")
assert out is not None
# Key must exist; value must be None. A 404 endpoint would have
# removed the field, but the spec keeps the key with a null value so
# the UI can branch on `matchedRemittance == null` cleanly.
assert "matchedRemittance" in out
assert out["matchedRemittance"] is None