"""SQLAlchemy-backed batch store for parsed X12 files. The module exposes a single ``CycloneStore`` class and a module-level singleton (``store``). All persistence flows through SQLAlchemy sessions via ``db.SessionLocal()()`` (the double-paren "function-style accessor" established in T2/T4). Public API (preserved from the in-memory version): - add(record) - get(batch_id) / get_batch(batch_id) - list(limit) / all() - iter_claims(...) / iter_remittances(...) - distinct_providers() / recent_activity(limit) New API (T12): - list_unmatched(kind="both") - manual_match(claim_id, remit_id) - manual_unmatch(claim_id) - AlreadyMatchedError, NotMatchedError, InvalidStateError exception classes Backward-compat shims for tests that relied on the in-memory internals: - ``_lock`` — a no-op ``threading.RLock``. SQLAlchemy handles concurrency via the engine's connection pool, but some existing tests use it as a context manager around cleanup. - ``_batches.clear()`` — wipes all rows from the DB tables so tests that depended on a fresh in-memory list per-test get a fresh DB state per-test. """ from __future__ import annotations import json import threading from datetime import datetime, timezone from decimal import Decimal from typing import Any, Literal from pydantic import BaseModel, ConfigDict, model_validator from cyclone import db from cyclone.db import ( Ack, ActivityEvent, Batch, CasAdjustment, Claim, ClaimState, JSONText, Match, Remittance, ) from cyclone.parsers.models import ClaimOutput, ParseResult from cyclone.parsers.models_835 import ClaimPayment, ParseResult835 from cyclone.parsers.payer import PayerConfig835 from cyclone.providers import Clearhouse, Payer, Provider # SP9: ORM-row DTOs class AlreadyMatchedError(Exception): """Raised by ``CycloneStore.manual_match`` when the claim is already paired. The claim's ``matched_remittance_id`` is set, so any new pairing would clobber an existing match. Callers (the T15 API endpoint) should surface this as a 409 Conflict. """ class NotMatchedError(Exception): """Raised by ``CycloneStore.manual_unmatch`` when the claim has no match. Mirrors ``AlreadyMatchedError`` for the unpair operation. Same HTTP treatment: 409 Conflict at the API layer. """ class InvalidStateError(Exception): """Raised when an apply_* pure fn returns a skipped ApplyIntent. ``reconcile.apply_payment`` / ``apply_reversal`` may return ``skipped=True`` (e.g. claim already in a terminal state, or reversal on a non-paid claim). The store surfaces that as ``InvalidStateError`` rather than silently pairing. The T15 API endpoint maps this to a 409 Conflict and echoes ``current_state`` and ``activity_kind`` so the UI can render a precise message. """ def __init__(self, current_state: str, activity_kind: str = "invalid_state"): self.current_state = current_state self.activity_kind = activity_kind super().__init__( f"invalid state {current_state} for apply (kind={activity_kind})" ) BatchKind = Literal["837p", "835"] # --------------------------------------------------------------------------- # BatchRecord: value object preserved from sub-project 1. # --------------------------------------------------------------------------- class BatchRecord(BaseModel): """One parsed file, with a stable uuid4 id and the full ParseResult. ``result`` is a union: ``ParseResult`` for ``kind="837p"`` and ``ParseResult835`` for ``kind="835"``. The concrete subclasses ``BatchRecord837`` and ``BatchRecord835`` narrow those fields, so callers that want type-checked access should use them and check ``isinstance`` rather than pattern-matching on ``kind``. Constructing ``BatchRecord(kind="837p", ...)`` dispatches to ``BatchRecord837``; ``kind="835"`` dispatches to ``BatchRecord835``. This lets the union-member narrowing work transparently. """ model_config = ConfigDict(extra="ignore") id: str kind: BatchKind input_filename: str parsed_at: datetime # tz-aware UTC result: ParseResult | ParseResult835 def __new__( cls, *args: Any, **kwargs: Any, ) -> BatchRecord837 | BatchRecord835 | BatchRecord: # Dispatch base-class construction to the right concrete subclass # so isinstance checks downstream narrow `result` correctly. if cls is BatchRecord: kind = kwargs.get("kind") if kind is None and args and isinstance(args[0], dict): kind = args[0].get("kind") if kind == "837p": return BatchRecord837(*args, **kwargs) if kind == "835": return BatchRecord835(*args, **kwargs) return super().__new__(cls) @model_validator(mode="after") def _check_parsed_at_tz(self) -> BatchRecord: if self.parsed_at.tzinfo is None: raise ValueError( "parsed_at must be tz-aware (use datetime.now(timezone.utc))" ) return self class BatchRecord837(BatchRecord): """A parsed 837P (professional claim) batch.""" kind: Literal["837p"] = "837p" result: ParseResult class BatchRecord835(BatchRecord): """A parsed 835 (remittance advice) batch.""" kind: Literal["835"] = "835" result: ParseResult835 def utcnow() -> datetime: """tz-aware UTC `datetime` (replaces the old `utcnow_iso` string helper).""" return datetime.now(timezone.utc) # --------------------------------------------------------------------------- # ORM row builders. # --------------------------------------------------------------------------- def _service_dates_from_claim(claim: ClaimOutput) -> tuple[date | None, date | None]: """Extract (service_date_from, service_date_to) from a ClaimOutput. The 837P model has ``service_lines[*].service_date`` (one per SV1). We use the earliest as ``from`` and the latest as ``to``; if there are no service lines, both are ``None``. """ dates: list[date] = [] for sl in claim.service_lines: if sl.service_date is not None: dates.append(sl.service_date) if not dates: return None, None return min(dates), max(dates) def _claim_837_row(claim: ClaimOutput, batch_id: str) -> Claim: """Build a Claim ORM row from a ClaimOutput. NOT yet persisted.""" d_from, d_to = _service_dates_from_claim(claim) return Claim( id=claim.claim_id, batch_id=batch_id, patient_control_number=claim.subscriber.member_id or "", service_date_from=d_from, service_date_to=d_to, charge_amount=Decimal(claim.claim.total_charge or 0), provider_npi=claim.billing_provider.npi, payer_id=claim.payer.id, state=ClaimState.SUBMITTED, raw_json=json.loads(claim.model_dump_json()), ) def _remittance_835_row(cp: ClaimPayment, batch_id: str) -> Remittance: """Build a Remittance ORM row from a ClaimPayment. NOT yet persisted.""" received_at = utcnow() # Adjustment amount: sum the CAS rows for the first service line. # NOTE: This is a best-effort placeholder used until the reconciliation # pass (T10) overwrites it from the persisted CasAdjustment rows. The # authoritative value comes from `reconcile.run()`, which sums # ``CasAdjustment.amount`` per ``remittance_id`` and writes the result # back to ``Remittance.adjustment_amount``. We keep this stub so the # row has a sane value if reconciliation is disabled or fails. adjustment = Decimal("0") if cp.service_payments: sp = cp.service_payments[0] for adj in sp.adjustments: adjustment += adj.amount # Use the first service line's service_date as the remit service_date. service_date: date | None = None if cp.service_payments and cp.service_payments[0].service_date is not None: service_date = cp.service_payments[0].service_date return Remittance( id=cp.payer_claim_control_number, batch_id=batch_id, payer_claim_control_number=cp.payer_claim_control_number, claim_id=None, status_code=cp.status_code, status_label=cp.status_label, total_charge=Decimal(cp.total_charge or 0), total_paid=Decimal(cp.total_paid or 0), patient_responsibility=cp.patient_responsibility, adjustment_amount=adjustment, received_at=received_at, service_date=service_date, is_reversal=cp.status_code in ("21", "22"), raw_json=json.loads(cp.model_dump_json()), ) def _cas_adjustment_row(adj, remittance_id: str) -> "db.CasAdjustment": """Build a CasAdjustment ORM row from a ClaimAdjustment. NOT yet persisted. One row per SVC-level CAS adjustment is persisted so the T10 reconcile aggregator can compute ``Remittance.adjustment_amount`` as ``SUM(CasAdjustment.amount) WHERE remittance_id = ...``. ``quantity`` is optional in the X12 CAS spec; we coerce to Decimal only when present to keep the column NULL for the common no-QTY case. """ from cyclone.db import CasAdjustment quantity = getattr(adj, "quantity", None) return CasAdjustment( remittance_id=remittance_id, group_code=adj.group_code, reason_code=adj.reason_code, amount=Decimal(str(adj.amount)), quantity=Decimal(str(quantity)) if quantity is not None else None, ) def _persist_835_remit(session, cp: "ClaimPayment", remittance_id: str) -> None: """SP7: persist ServiceLinePayment + CAS rows for one CLP composite. For each 835 SVC composite in ``cp.service_payments``: - insert a ServiceLinePayment row (line_number, procedure, modifiers, charge, payment, units, service_date). - flush to populate slp.id. - insert each per-SVC CAS adjustment with ``service_line_payment_id`` set to slp.id. For CLP-level CAS adjustments (``cp.claim_adjustments``, a future extension; not produced by today's 835 parser but allowed by the spec): - insert CAS rows with ``service_line_payment_id IS NULL``. The caller controls the transaction; this function does not commit. The 835 ingest site calls this after ``_remittance_835_row`` is flushed so the FK target is populated. """ import json as _json from cyclone.db import ServiceLinePayment, CasAdjustment for svc in cp.service_payments: slp = ServiceLinePayment( remittance_id=remittance_id, line_number=svc.line_number, procedure_qualifier=svc.procedure_qualifier, procedure_code=svc.procedure_code, modifiers_json=_json.dumps(svc.modifiers or []), charge=Decimal(str(svc.charge)), payment=Decimal(str(svc.payment)), units=Decimal(str(svc.units)) if svc.units is not None else None, unit_type=svc.unit_type, service_date=svc.service_date, ref_benefit_plan=svc.ref_benefit_plan, ) session.add(slp) session.flush() # populate slp.id for the FK below for adj in svc.adjustments: quantity = getattr(adj, "quantity", None) session.add(CasAdjustment( remittance_id=remittance_id, group_code=adj.group_code, reason_code=adj.reason_code, amount=Decimal(str(adj.amount)), quantity=Decimal(str(quantity)) if quantity is not None else None, service_line_payment_id=slp.id, )) # CLP-level CAS (no SVC composite to attach to). Today's parser does # not produce these; the branch is forward-compatible. for adj in getattr(cp, "claim_adjustments", []) or []: quantity = getattr(adj, "quantity", None) session.add(CasAdjustment( remittance_id=remittance_id, group_code=adj.group_code, reason_code=adj.reason_code, amount=Decimal(str(adj.amount)), quantity=Decimal(str(quantity)) if quantity is not None else None, service_line_payment_id=None, )) # --------------------------------------------------------------------------- # UI mappers: ORM rows → simpler UI types. # --------------------------------------------------------------------------- def _claim_status_from_validation(claim: ClaimOutput) -> str: """Re-implement the in-memory status rules (sub-project 1 §6.2).""" v = claim.validation if not v.passed: has_r050 = any(e.rule == "R050_diagnosis_present" for e in v.errors) return "draft" if has_r050 else "denied" if claim.claim.frequency_code == "1": return "submitted" if v.warnings: return "pending" return "draft" def to_ui_claim( claim: ClaimOutput, *, batch_id: str, parsed_at: datetime, ) -> dict: """Map a 837P ClaimOutput to the UI's `Claim` shape (preserved).""" parsed_iso = parsed_at.isoformat().replace("+00:00", "Z") return { "id": claim.claim_id, "patientName": f"{claim.subscriber.first_name} {claim.subscriber.last_name}".strip(), "providerNpi": claim.billing_provider.npi, "payerName": claim.payer.name, "cptCode": ( claim.service_lines[0].procedure.code if claim.service_lines else "" ), "billedAmount": float(claim.claim.total_charge or 0.0), "receivedAmount": 0.0, "status": _claim_status_from_validation(claim), "denialReason": None, "submissionDate": parsed_iso, "batchId": batch_id, "parsedAt": parsed_iso, } def to_ui_remittance( cp: ClaimPayment, *, batch_id: str, parsed_at: datetime, payer_config: PayerConfig835 | None = None, payer_name: str = "", ) -> dict: """Map an 835 ClaimPayment to the UI's `Remittance` shape (preserved).""" code = cp.status_code if code in {"21", "22"}: status = "reconciled" else: status = "received" denial_reason: str | None = None if code == "4" and cp.service_payments: sp = cp.service_payments[0] if sp.adjustments: adj = sp.adjustments[0] denial_reason = ( f"{adj.group_code}-{adj.reason_code}: ${float(adj.amount):.2f}" ) cfg = payer_config if payer_config is not None else PayerConfig835.generic_835() validation_warnings: list[str] = [] if code not in cfg.allowed_status_codes: validation_warnings.append( f"CLP02 code {code} not in payer allowlist" ) # Aggregate adjustmentAmount across ALL service-line CAS rows, not just # the first line. Mirrors the SUM the reconcile aggregator (T10) # computes against persisted CasAdjustment rows; this inline version # is the write-path equivalent (used when streaming 835 NDJSON # responses before persistence finishes). adjustment_total = Decimal("0") for sp in cp.service_payments: for adj in sp.adjustments: adjustment_total += adj.amount parsed_iso = parsed_at.isoformat().replace("+00:00", "Z") return { "id": cp.payer_claim_control_number, "claimId": cp.original_claim_id or "", "payerName": payer_name, "paidAmount": float(cp.total_paid or 0.0), "adjustmentAmount": float(adjustment_total), "status": status, "denialReason": denial_reason, "validationWarnings": validation_warnings, "receivedDate": parsed_iso, "batchId": batch_id, "parsedAt": parsed_iso, } def to_ui_claim_from_orm( row: Claim, *, batch_id: str, parsed_at: datetime, ) -> dict: """Map an ORM ``Claim`` row to the UI's claim shape. ``to_ui_claim`` takes a Pydantic ``ClaimOutput`` (used on the write path during 837 ingest). For read paths — list_unmatched, manual_match return values — we already have the ORM row and the serialized fields it carries in ``raw_json``. Reading from ``raw_json`` keeps the UI shape in sync with the original 837 parse without re-deserializing to a Pydantic model. Adds two fields ``to_ui_claim`` doesn't emit: ``state`` (the reconciliation state machine value) and ``matchedRemittanceId`` (the FK to the paired remittance, or None). Both are required by the UI. """ raw = row.raw_json or {} bp = raw.get("billing_provider", {}) payer_obj = raw.get("payer", {}) sub = raw.get("subscriber", {}) service_lines = raw.get("service_lines", []) parsed_iso = parsed_at.isoformat().replace("+00:00", "Z") cpt = ( service_lines[0].get("procedure", {}).get("code", "") if service_lines else "" ) state_value = ( row.state.value if hasattr(row.state, "value") else str(row.state) ) return { "id": row.id, "state": state_value, "billedAmount": float(row.charge_amount or 0), "patientName": ( f"{sub.get('first_name', '')} {sub.get('last_name', '')}".strip() ), "providerNpi": bp.get("npi") or row.provider_npi or "", "payerName": payer_obj.get("name") or "", "cptCode": cpt, "submissionDate": parsed_iso, "parsedAt": parsed_iso, "status": state_value, "matchedRemittanceId": row.matched_remittance_id, "batchId": batch_id, # Parity with ``to_ui_claim``'s shape — the UI tolerates extra keys # but expects these on freshly-loaded rows from /api/claims too. "receivedAmount": 0.0, "denialReason": None, } # 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, *, batch_id: str, parsed_at: datetime, ) -> dict: """Map an ORM ``Remittance`` row to the UI's remittance shape. Same idea as ``to_ui_claim_from_orm``: read the PayerName from the parent batch's ``raw_result_json`` (the ParseResult835 stashed at insert time) since ``Remittance`` itself doesn't carry it. """ parsed_iso = parsed_at.isoformat().replace("+00:00", "Z") payer_name = "" if row.batch is not None and row.batch.raw_result_json: payer_obj = row.batch.raw_result_json.get("payer", {}) or {} payer_name = payer_obj.get("name") or "" status = ( "reconciled" if row.status_code in ("21", "22") else "received" ) return { "id": row.id, "payerClaimControlNumber": row.payer_claim_control_number, "claimId": row.claim_id or "", "payerName": payer_name, "paidAmount": float(row.total_paid or 0), "adjustmentAmount": float(row.adjustment_amount or 0), "status": status, "denialReason": None, "validationWarnings": [], "receivedDate": row.received_at.isoformat().replace("+00:00", "Z"), "batchId": batch_id, "parsedAt": parsed_iso, } def to_ui_remittance_with_adjustments( row: Remittance, *, batch_id: str, parsed_at: datetime, cas_rows: list["db.CasAdjustment"] | None = None, ) -> dict: """Same shape as :func:`to_ui_remittance_from_orm` plus an ``adjustments`` array. Each persisted ``CasAdjustment`` row is rendered as ``{"group", "reason", "label", "amount", "quantity"}``. The ``label`` is resolved through :func:`cyclone.parsers.cas_codes.reason_label` so the UI does not have to ship its own CARC dictionary. ``cas_rows`` is optional so callers that don't have the rows handy can still get the base dict; in that case ``adjustments`` is ``[]``. Pass ``cas_rows`` to avoid an extra round-trip; the endpoint at ``GET /api/remittances/{id}`` passes them in to keep this mapper a pure function. """ base = to_ui_remittance_from_orm( row, batch_id=batch_id, parsed_at=parsed_at, ) if not cas_rows: base["adjustments"] = [] return base # Lazy import to avoid the circular store ↔ parsers import that # happens on cold start; mirrors the same pattern used elsewhere # in this module. from cyclone.parsers.cas_codes import reason_label base["adjustments"] = [ { "group": c.group_code, "reason": c.reason_code, "label": reason_label(c.group_code, c.reason_code), "amount": float(c.amount or 0), "quantity": ( float(c.quantity) if c.quantity is not None else None ), } for c in cas_rows ] return base def _svc_to_wire_dict(svc) -> dict: """Project an ORM ``ServiceLinePayment`` to the wire format used by the remit drawer's ``serviceLinePayments`` array. Mirrors the shape produced by the line-reconciliation endpoint so the UI can render the same components from either source. """ import json as _json return { "id": svc.id, "line_number": svc.line_number, "procedure_qualifier": svc.procedure_qualifier, "procedure_code": svc.procedure_code, "modifiers": _json.loads(svc.modifiers_json or "[]"), "charge": str(Decimal(str(svc.charge))), "payment": str(Decimal(str(svc.payment))), "units": str(Decimal(str(svc.units))) if svc.units is not None else None, "unit_type": svc.unit_type, "service_date": svc.service_date.isoformat() if svc.service_date else None, } def to_ui_provider( *, npi: str, name: str, tax_id: str | None = None, address: str | None = None, city: str | None = None, state: str | None = None, zip: str | None = None, phone: str | None = None, claim_count: int = 0, outstanding_ar: float = 0.0, ) -> dict: return { "npi": npi, "name": name, "taxId": tax_id or "", "address": address or "", "city": city or "", "state": state or "", "zip": zip or "", "phone": phone or "", "claimCount": claim_count, "outstandingAr": float(outstanding_ar), } def to_activity_event( *, id: str, kind: str, message: str, timestamp: datetime, npi: str | None = None, amount: float | None = None, ) -> dict: return { "id": id, "kind": kind, "message": message, "timestamp": timestamp.isoformat().replace("+00:00", "Z"), "npi": npi, "amount": amount, } def _date_in_bounds( item: dict, field: str, date_from: str | None, date_to: str | None, ) -> bool: """True if ``item[field]`` falls within ``[date_from, date_to]``.""" val = item.get(field) if val is None: return date_from is None and date_to is None date_part = val[:10] if date_from is not None and date_part < date_from: return False if date_to is not None and date_part > date_to: return False return True # --------------------------------------------------------------------------- # Backward-compat shim: tests called ``_batches.clear()`` on the in-memory # store. The DB-backed store doesn't have an in-memory list, so we expose # a tiny shim object whose ``.clear()`` wipes the DB. # --------------------------------------------------------------------------- class _BatchesShim: """Drop-in replacement for the old in-memory ``_batches`` list. ``clear()`` removes every row from the DB tables in FK-safe order. Other list operations are not implemented because the only call site is the ``clear()`` inside the test fixtures (``test_api_gets.py`` and ``test_api_parse_persists.py``). """ def clear(self) -> None: # type: ignore[no-untyped-def] with db.SessionLocal()() as s: s.query(ActivityEvent).delete() s.query(Match).delete() s.query(CasAdjustment).delete() s.query(Remittance).delete() s.query(Claim).delete() s.query(Batch).delete() s.commit() # --------------------------------------------------------------------------- # CycloneStore: the SQLAlchemy-backed facade. # --------------------------------------------------------------------------- class CycloneStore: """SQLAlchemy-backed facade over the parsed X12 store. Each public method opens a short-lived session via ``db.SessionLocal()()`` so callers don't have to manage session lifecycles. Concurrency is handled by the SQLAlchemy engine; the ``_lock`` attribute is a no-op ``RLock`` retained for backward compatibility with code that wrapped cleanup in a lock context. """ def __init__(self) -> None: self._lock = threading.RLock() self._batches = _BatchesShim() # -- write path ----------------------------------------------------- def add( self, record: BatchRecord, *, event_bus: "EventBus | None" = None, ) -> None: """Persist a parsed batch (837P or 835) to the DB. For 837P batches: inserts the Batch row, one Claim row per claim, and a ``claim_submitted`` ActivityEvent per claim. For 835 batches: inserts the Batch row, one Remittance row per ClaimPayment, and a ``remit_received`` ActivityEvent per ClaimPayment. After commit, calls ``_run_reconcile`` (T10 stub) in a fail-soft manner — reconciliation errors are logged but do not roll back the persisted batch. Idempotency: ``Claim.id`` and ``Remittance.id`` are PRIMARY KEYS, so a re-ingest of the same fixture (e.g. ``/api/parse-837`` called twice with the same file) would otherwise raise ``IntegrityError``. We do a per-row ``session.get(...)`` check before each insert; if the row already exists, we log a warning and skip. The batch row itself is still inserted (each parse has a fresh ``uuid4`` id from the API). O(n) per row, but acceptable for the small fixture sizes — production load is one batch at a time via the API, not bulk inserts. When ``event_bus`` is provided, publishes one ``claim_written`` or ``remittance_written`` event per newly-inserted row plus an ``activity_recorded`` event per activity row, after commit. The publish calls are best-effort — failures are logged but do not roll back the persisted batch. """ from cyclone.pubsub import EventBus import logging log = logging.getLogger(__name__) # Track rows we actually inserted so we can publish events for them. inserted_claim_ids: list[str] = [] inserted_remit_ids: list[str] = [] with db.SessionLocal()() as s: batch_row = Batch( id=record.id, kind=record.kind, input_filename=record.input_filename, parsed_at=record.parsed_at, totals_json=None, validation_json=None, raw_result_json=json.loads(record.result.model_dump_json()), ) s.add(batch_row) if isinstance(record, BatchRecord837): result: ParseResult = record.result for claim in result.claims: if s.get(Claim, claim.claim_id) is not None: log.warning( "add: claim %s already exists; skipping (batch=%s)", claim.claim_id, record.id, ) continue s.add(_claim_837_row(claim, record.id)) s.add(ActivityEvent( ts=record.parsed_at, kind="claim_submitted", batch_id=record.id, claim_id=claim.claim_id, payload_json={ "message": ( f"Claim {claim.claim_id} submitted · " f"{claim.payer.name}" ), "npi": claim.billing_provider.npi, "amount": float(claim.claim.total_charge or 0.0), }, )) inserted_claim_ids.append(claim.claim_id) elif isinstance(record, BatchRecord835): result835: ParseResult835 = record.result payer_name = result835.payer.name for cp in result835.claims: if s.get(Remittance, cp.payer_claim_control_number) is not None: log.warning( "add: remittance %s already exists; skipping (batch=%s)", cp.payer_claim_control_number, record.id, ) continue remit_row = _remittance_835_row(cp, record.id) s.add(remit_row) # Flush so remit_row.id (FK target of cas_adjustments) is # populated. SQLAlchemy assigns the PK on flush; without # this the CasAdjustment inserts below would reference an # unset id and violate the FK. s.flush() # SP7: persist per-line ServiceLinePayment + linked # SVC-level CAS rows + claim-level CAS bucket. Replaces # the previous per-SVC CAS insert loop so the # service_line_payment_id FK is set correctly. _persist_835_remit(s, cp, remit_row.id) s.add(ActivityEvent( ts=record.parsed_at, kind="remit_received", batch_id=record.id, remittance_id=cp.payer_claim_control_number, payload_json={ "message": ( f"Remit {cp.payer_claim_control_number} received" ), "payerName": payer_name, "amount": float(cp.total_paid or 0.0), }, )) inserted_remit_ids.append(cp.payer_claim_control_number) else: raise TypeError( f"Unsupported BatchRecord subclass: {type(record).__name__}" ) s.commit() # Reconcile 835 batches after the batch is durably persisted. # Fail-soft: errors are logged, not raised, so an 835 parse that # crashes in reconciliation still shows up in /api/batches. if record.kind == "835": try: self._run_reconcile(record.id) except Exception: # pragma: no cover - logged via default handler import logging logging.getLogger(__name__).exception( "reconcile.run failed for batch %s", record.id, ) # Publish live-tail events synchronously. EventBus.publish is async # but its body is purely synchronous ``put_nowait`` enqueues; we # bypass the async wrapper and call the internal enqueue directly # so callers (sync FastAPI endpoints, sync test harnesses) don't # need to await. if event_bus is not None and (inserted_claim_ids or inserted_remit_ids): self._publish_events_sync( event_bus, record, inserted_claim_ids, inserted_remit_ids, ) def _publish_events_sync( self, event_bus: "EventBus", record: BatchRecord, claim_ids: list[str], remit_ids: list[str], ) -> None: """Build UI-shaped payloads for newly-inserted rows and publish. Runs after commit so subscribers can immediately re-fetch from the API and see consistent data. Each ``claim_written`` / ``remittance_written`` payload is identical to what the matching list endpoint would return for that row. This is sync because EventBus's enqueue path is sync; we don't need a coroutine for ``put_nowait``. """ import logging log = logging.getLogger(__name__) try: with db.SessionLocal()() as s: for cid in claim_ids: row = s.get(Claim, cid) if row is None: continue ui = to_ui_claim_from_orm( row, batch_id=row.batch_id or record.id, parsed_at=record.parsed_at, ) self._sync_publish(event_bus, "claim_written", ui) for rid in remit_ids: row = s.get(Remittance, rid) if row is None: continue ui = to_ui_remittance_from_orm( row, batch_id=row.batch_id or record.id, parsed_at=record.parsed_at, ) self._sync_publish(event_bus, "remittance_written", ui) # Activity events for this batch. from sqlalchemy import select activity_rows = s.execute( select(ActivityEvent).where(ActivityEvent.batch_id == record.id) ).scalars().all() for arow in activity_rows: ui = { "kind": arow.kind, "ts": arow.ts.isoformat().replace("+00:00", "Z"), "batchId": arow.batch_id, "claimId": arow.claim_id, "remittanceId": arow.remittance_id, "payload": arow.payload_json, } self._sync_publish(event_bus, "activity_recorded", ui) except Exception: log.exception("add: event publish failed for batch %s", record.id) @staticmethod def _sync_publish(event_bus: "EventBus", kind: str, payload: dict) -> None: """Synchronous fan-out helper. Mirrors ``EventBus.publish`` but bypasses the async wrapper so callers don't need an event loop. """ event = {**payload, "_kind": kind} for queue in list(event_bus._subscribers.get(kind, ())): event_bus._enqueue_or_drop_oldest(queue, event) def _run_reconcile(self, batch_id: str) -> None: """T10 stub: invoke the reconcile orchestrator for a batch. The actual reconciliation is implemented in T10 (this method will import ``cyclone.reconcile`` and call ``reconcile.run`` with the current session). For T9 we keep the import lazy and fail-soft so a missing or NotImplementedError reconcile module never breaks the 835 ingest path. """ # T10 will replace this with the real implementation. Until then # we accept the failure modes the spec lists: ImportError # (module not yet wired), NotImplementedError (stub in place), # or any other transient reconcile error — all swallowed. try: from cyclone import reconcile as _reconcile with db.SessionLocal()() as s: _reconcile.run(s, batch_id) except (ImportError, NotImplementedError): pass # -- read path ------------------------------------------------------ def _row_to_record(self, row: Batch) -> BatchRecord: """Rehydrate a ``BatchRecord`` (837 or 835) from a Batch ORM row. The full ``ParseResult`` / ``ParseResult835`` lives in ``raw_result_json`` (stashed at insert time). Re-parsing JSON here means callers get the same typed Pydantic object the old in-memory store handed out, so api.py and tests that do ``rec.result.claims`` keep working unchanged. SQLite drops tz info on round-trip even though the column type is ``DateTime(timezone=True)``. We re-attach UTC so the ``BatchRecord`` validator (``parsed_at must be tz-aware``) passes. """ if row.kind == "835": result_cls = ParseResult835 else: result_cls = ParseResult payload = row.raw_result_json or {} result = result_cls.model_validate(payload) parsed_at = row.parsed_at if parsed_at is not None and parsed_at.tzinfo is None: parsed_at = parsed_at.replace(tzinfo=timezone.utc) record_cls = BatchRecord835 if row.kind == "835" else BatchRecord837 return record_cls( id=row.id, kind=row.kind, input_filename=row.input_filename, parsed_at=parsed_at, result=result, ) def get_batch(self, batch_id: str) -> dict | None: """Return a summary dict for ``batch_id`` or ``None`` if missing. The dict shape matches what ``/api/batches/{id}`` callers need: ``id``, ``kind``, ``input_filename``, ``parsed_at``, and the full ``result`` (raw_result_json) as a dict. """ with db.SessionLocal()() as s: row = s.get(Batch, batch_id) if row is None: return None return { "id": row.id, "kind": row.kind, "input_filename": row.input_filename, "parsed_at": row.parsed_at, "result": row.raw_result_json, } def get(self, batch_id: str) -> BatchRecord | None: """Return the ``BatchRecord`` for ``batch_id`` or ``None``. Preserves the in-memory store contract: callers get a Pydantic ``BatchRecord`` (subclass ``BatchRecord837`` / ``BatchRecord835``) with ``.id``, ``.kind``, ``.input_filename``, ``.parsed_at``, and ``.result`` (typed ``ParseResult`` / ``ParseResult835``). """ with db.SessionLocal()() as s: row = s.get(Batch, batch_id) if row is None: return None return self._row_to_record(row) def list(self, *, limit: int = 100) -> list[BatchRecord]: """Return up to ``limit`` ``BatchRecord``s, newest first.""" with db.SessionLocal()() as s: rows = ( s.query(Batch) .order_by(Batch.parsed_at.desc()) .limit(limit) .all() ) return [self._row_to_record(r) for r in rows] def get_remittance(self, remittance_id: str) -> dict | None: """Return a UI-shaped remittance dict with ``adjustments`` array. Joins the persisted ``CasAdjustment`` rows for ``remittance_id`` and labels each via :mod:`cyclone.parsers.cas_codes`. Returns ``None`` when the remittance is not found so the API layer can map that to a 404. SP7: also returns the per-line SVC composites (``serviceLinePayments``) and the CLP-level (claim-level) CAS bucket (``claimLevelAdjustments``) so the remit drawer can show per-line payments + adjustments without a second fetch. """ with db.SessionLocal()() as s: row = s.get(Remittance, remittance_id) if row is None: return None cas_rows = ( s.query(CasAdjustment) .filter(CasAdjustment.remittance_id == remittance_id) .all() ) parsed_at = ( row.batch.parsed_at if row.batch is not None else row.received_at ) if parsed_at is not None and parsed_at.tzinfo is None: parsed_at = parsed_at.replace(tzinfo=timezone.utc) body = to_ui_remittance_with_adjustments( row, batch_id=row.batch_id, parsed_at=parsed_at, cas_rows=cas_rows, ) # SP7: per-line SVC composites + claim-level CAS bucket. from cyclone.db import ServiceLinePayment as SLP slps = ( s.query(SLP) .filter(SLP.remittance_id == remittance_id) .order_by(SLP.line_number) .all() ) body["serviceLinePayments"] = [ _svc_to_wire_dict(svc) for svc in slps ] body["claimLevelAdjustments"] = [ { "id": c.id, "group_code": c.group_code, "reason_code": c.reason_code, "amount": str(Decimal(str(c.amount))), "quantity": ( str(Decimal(str(c.quantity))) if c.quantity is not None else None ), } for c in cas_rows if c.service_line_payment_id is None ] return body 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() ) # Claim.batch_id is FK NOT NULL with ON DELETE CASCADE, so # ``row.batch`` is always populated in normal flow. Re-attach # UTC only when SQLite drops the tzinfo on read. parsed_at = row.batch.parsed_at if 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. # SP7 §5.2: slim per-line projection so the ServiceLinesTable # can show Paid + Adjustments columns without a second fetch. # The 837 side is keyed by ``claim_service_line_number`` (the # 1-based line number from raw_json) since 837 service lines # are not a separate ORM table. from cyclone.db import ( LineReconciliation, ServiceLinePayment, CasAdjustment, ) slim_lrs = list( s.query(LineReconciliation) .filter(LineReconciliation.claim_id == claim_id) .all() ) svc_ids_for_cas = [ lr.service_line_payment_id for lr in slim_lrs if lr.service_line_payment_id is not None ] cas_sums_by_svc: dict = {} svc_by_id_slim: dict = {} if svc_ids_for_cas: cas_rows = ( s.query(CasAdjustment.service_line_payment_id, CasAdjustment.amount) .filter(CasAdjustment.service_line_payment_id.in_(svc_ids_for_cas)) .all() ) from collections import defaultdict agg = defaultdict(lambda: Decimal("0")) for svc_id, amount in cas_rows: agg[svc_id] += Decimal(str(amount)) cas_sums_by_svc = {k: str(v) for k, v in agg.items()} for svc in ( s.query(ServiceLinePayment) .filter(ServiceLinePayment.id.in_(svc_ids_for_cas)) .all() ): svc_by_id_slim[svc.id] = svc slim_by_num: dict = { lr.claim_service_line_number: lr for lr in slim_lrs if lr.claim_service_line_number is not None } line_reconciliation_slim: list = [] for sl in detail["serviceLines"]: ln = sl.get("lineNumber") lr = slim_by_num.get(ln) if lr is None: line_reconciliation_slim.append({ "lineNumber": ln, "status": "unmatched_837_only", "paid": None, "adjustmentsSum": None, }) continue svc = ( svc_by_id_slim.get(lr.service_line_payment_id) if lr.service_line_payment_id else None ) line_reconciliation_slim.append({ "lineNumber": ln, "status": lr.status, "paid": str(Decimal(str(svc.payment))) if svc else None, "adjustmentsSum": ( cas_sums_by_svc.get(lr.service_line_payment_id) if lr.service_line_payment_id else None ), }) detail["lineReconciliation"] = line_reconciliation_slim return detail def all(self) -> list[BatchRecord]: """Return every ``BatchRecord``, oldest first (no pagination).""" with db.SessionLocal()() as s: rows = s.query(Batch).order_by(Batch.parsed_at.asc()).all() return [self._row_to_record(r) for r in rows] def load_two_for_diff( self, a_id: str, b_id: str, ) -> tuple[BatchRecord, BatchRecord]: """Load two batches by id for the side-by-side diff view. Returns ``(a, b)`` as ``BatchRecord`` objects. Raises :class:`LookupError` when either id is missing — the API layer catches it and maps it to ``404 Not Found`` (matching the ``GET /api/batches/{id}`` contract). The two loads happen in independent sessions so a transient failure on one side can't poison the other. Used exclusively by :mod:`cyclone.batch_diff` via the ``/api/batch-diff`` endpoint. """ a = self.get(a_id) if a is None: raise LookupError(f"batch {a_id} not found") b = self.get(b_id) if b is None: raise LookupError(f"batch {b_id} not found") return a, b def iter_claims( self, *, batch_id: str | None = None, status: str | None = None, provider_npi: str | None = None, payer: str | None = None, date_from: str | None = None, date_to: str | None = None, sort: str | None = None, order: str = "desc", limit: int = 100, offset: int = 0, ) -> list[dict]: """Return UI-shaped claim dicts from the DB. Filters mirror the in-memory version. The ``payer`` filter is a case-insensitive substring on the payer's ``name``, recovered from each claim's ``raw_json`` payload (the DB stores it there because ``Claim`` itself only carries ``payer_id``). """ with db.SessionLocal()() as s: q = s.query(Claim) if batch_id is not None: q = q.filter(Claim.batch_id == batch_id) if status is not None: q = q.filter(Claim.state == ClaimState(status)) if provider_npi is not None: q = q.filter(Claim.provider_npi == provider_npi) rows = q.all() out: list[dict] = [] for r in rows: raw = r.raw_json or {} bp = raw.get("billing_provider", {}) payer_obj = raw.get("payer", {}) sub = raw.get("subscriber", {}) claim_hdr = raw.get("claim", {}) service_lines = raw.get("service_lines", []) parsed_at_iso = ( r.batch.parsed_at.isoformat().replace("+00:00", "Z") if r.batch is not None else "" ) cpt = ( service_lines[0].get("procedure", {}).get("code", "") if service_lines else "" ) out.append({ "id": r.id, "patientName": ( f"{sub.get('first_name', '')} " f"{sub.get('last_name', '')}".strip() ), "providerNpi": bp.get("npi") or r.provider_npi or "", "payerName": payer_obj.get("name") or "", "cptCode": cpt, "billedAmount": float(r.charge_amount or 0), "receivedAmount": 0.0, "status": r.state.value if hasattr(r.state, "value") else str(r.state), "state": r.state.value if hasattr(r.state, "value") else str(r.state), "denialReason": None, "submissionDate": parsed_at_iso, "batchId": r.batch_id, "parsedAt": parsed_at_iso, # Keep these so we can sort on them in-memory below. "_sort_billedAmount": float(r.charge_amount or 0), "_sort_submissionDate": parsed_at_iso, }) if payer is not None: needle = payer.casefold() out = [ c for c in out if needle in (c.get("payerName") or "").casefold() ] out = [ c for c in out if _date_in_bounds(c, "submissionDate", date_from, date_to) ] if sort is not None: out.sort( key=lambda c: c.get(f"_sort_{sort}", 0) or 0, reverse=(order == "desc"), ) # Drop the private sort keys before returning. for c in out: c.pop("_sort_billedAmount", None) c.pop("_sort_submissionDate", None) return out[offset:offset + limit] def iter_remittances( self, *, batch_id: str | None = None, payer: str | None = None, claim_id: str | None = None, date_from: str | None = None, date_to: str | None = None, sort: str | None = None, order: str = "desc", limit: int = 100, offset: int = 0, ) -> list[dict]: """Return UI-shaped remittance dicts from the DB.""" with db.SessionLocal()() as s: q = s.query(Remittance) if batch_id is not None: q = q.filter(Remittance.batch_id == batch_id) if claim_id is not None: q = q.filter(Remittance.claim_id == claim_id) rows = q.all() # Bulk-fetch all CAS rows for these remittances in one query # (SP3 P2 follow-up — fixes the list-view's empty adjustments # expansion). N+1-free. cas_by_remit: dict[str, list] = {} if rows: from cyclone.parsers.cas_codes import reason_label cas_rows = ( s.query(CasAdjustment) .filter(CasAdjustment.remittance_id.in_([r.id for r in rows])) .all() ) for c in cas_rows: cas_by_remit.setdefault(c.remittance_id, []).append(c) out: list[dict] = [] for r in rows: raw = r.raw_json or {} parsed_at_iso = ( r.batch.parsed_at.isoformat().replace("+00:00", "Z") if r.batch is not None else r.received_at.isoformat().replace("+00:00", "Z") ) payer_name = "" if r.batch is not None and r.batch.raw_result_json: payer_name = ( r.batch.raw_result_json.get("payer", {}).get("name", "") ) adjustments = [ { "group": c.group_code, "reason": c.reason_code, "label": reason_label(c.group_code, c.reason_code), "amount": float(c.amount), "quantity": float(c.quantity) if c.quantity is not None else None, } for c in cas_by_remit.get(r.id, []) ] out.append({ "id": r.id, "claimId": r.claim_id or "", "payerName": payer_name, "paidAmount": float(r.total_paid or 0), "adjustmentAmount": float(r.adjustment_amount or 0), "status": ( "reconciled" if r.status_code in ("21", "22") else "received" ), "denialReason": None, "validationWarnings": [], "receivedDate": r.received_at.isoformat().replace("+00:00", "Z"), "batchId": r.batch_id, "parsedAt": parsed_at_iso, "adjustments": adjustments, "_sort_receivedDate": r.received_at.isoformat().replace("+00:00", "Z"), }) if payer is not None: out = [r for r in out if r.get("payerName") == payer] out = [ r for r in out if _date_in_bounds(r, "receivedDate", date_from, date_to) ] if sort is not None: out.sort( key=lambda r: r.get(f"_sort_{sort}", 0) or 0, reverse=(order == "desc"), ) for r in out: r.pop("_sort_receivedDate", None) return out[offset:offset + limit] def distinct_providers(self) -> list[dict]: """Group claims by NPI and return one row per provider.""" with db.SessionLocal()() as s: rows = s.query(Claim).all() by_npi: dict[str, dict] = {} for r in rows: npi = r.provider_npi or "" if npi not in by_npi: raw = r.raw_json or {} bp = raw.get("billing_provider", {}) by_npi[npi] = to_ui_provider( npi=npi, name=bp.get("name") or "", tax_id=bp.get("tax_id"), address=None, city=None, state=None, zip=None, phone=None, claim_count=0, outstanding_ar=0.0, ) by_npi[npi]["claimCount"] += 1 return list(by_npi.values()) def recent_activity(self, *, limit: int = 200) -> list[dict]: """Return recent activity events from the DB, newest first.""" with db.SessionLocal()() as s: rows = ( s.query(ActivityEvent) .order_by(ActivityEvent.ts.desc()) .limit(limit) .all() ) return [ { "id": f"ae-{r.id}", "kind": r.kind, "message": (r.payload_json or {}).get("message", ""), "timestamp": r.ts.isoformat().replace("+00:00", "Z"), "npi": (r.payload_json or {}).get("npi"), "amount": (r.payload_json or {}).get("amount"), } for r in rows ] # -- 999 ACKs (SP3 P3 T13) ------------------------------------------- def add_ack( self, *, source_batch_id: str, accepted_count: int, rejected_count: int, received_count: int, ack_code: str, raw_json: dict, ) -> db.Ack: """Persist a 999 ACK row and return it. ``source_batch_id`` must reference an existing ``batches.id``. For received 999s with no source batch the caller should pass a synthetic id (e.g. ``"999-"``) — see ``/api/parse-999`` for that policy. ``raw_json`` is the full ``ParseResult999`` model dump; the detail endpoint surfaces it without re-parsing the original X12 text. """ with db.SessionLocal()() as s: row = Ack( source_batch_id=source_batch_id, accepted_count=accepted_count, rejected_count=rejected_count, received_count=received_count, ack_code=ack_code, parsed_at=utcnow(), raw_json=raw_json, ) s.add(row) s.commit() s.refresh(row) return row def list_acks(self) -> list[db.Ack]: """Return every 999 ACK row, newest first (auto-increment id desc).""" with db.SessionLocal()() as s: return ( s.query(Ack) .order_by(Ack.id.desc()) .all() ) def get_ack(self, ack_id: int) -> db.Ack | None: """Return a single ACK row by id, or ``None`` if not found.""" with db.SessionLocal()() as s: return s.get(Ack, ack_id) # -- TA1 (Interchange Acknowledgment) ------------------------------- def add_ta1_ack( self, *, source_batch_id: str, control_number: str, interchange_date: date | None, interchange_time: str | None, ack_code: str, note_code: str | None, ack_generated_date: date | None, sender_id: str, receiver_id: str, raw_json: dict, ) -> db.Ta1Ack: """Persist a TA1 (Interchange Acknowledgment) row and return it. Mirrors :meth:`add_ack` for the lower-level envelope ack. The flat columns are promoted out of ``raw_json`` so the list endpoint can sort/filter without a JSON parse; the full ``ParseResultTa1`` stays in ``raw_json`` for the detail endpoint. """ with db.SessionLocal()() as s: row = db.Ta1Ack( source_batch_id=source_batch_id, control_number=control_number, interchange_date=interchange_date, interchange_time=interchange_time, ack_code=ack_code, note_code=note_code, ack_generated_date=ack_generated_date, sender_id=sender_id, receiver_id=receiver_id, parsed_at=utcnow(), raw_json=raw_json, ) s.add(row) s.commit() s.refresh(row) return row def list_ta1_acks(self) -> list[db.Ta1Ack]: """Return every TA1 ACK row, newest first (auto-increment id desc). Mirrors :meth:`list_acks` — the API endpoint slices to its own ``limit`` so the ``total`` field reflects the full row count. """ with db.SessionLocal()() as s: return ( s.query(db.Ta1Ack) .order_by(db.Ta1Ack.id.desc()) .all() ) def get_ta1_ack(self, ack_id: int) -> db.Ta1Ack | None: """Return a single TA1 ACK row by id, or ``None`` if not found.""" with db.SessionLocal()() as s: return s.get(db.Ta1Ack, ack_id) # -- 277CA (SP10) -------------------------------------------------- def add_277ca_ack( self, *, source_batch_id: str, control_number: str, accepted_count: int, rejected_count: int, paid_count: int, pended_count: int, raw_json: dict, ) -> db.Two77caAck: """Persist a 277CA (Claim Acknowledgment) row and return it. Mirrors :meth:`add_ack` but for the claim-level ack. The per-claim status detail stays in ``raw_json``; only the four counts are promoted so the list endpoint stays fast. """ with db.SessionLocal()() as s: row = db.Two77caAck( source_batch_id=source_batch_id, control_number=control_number, accepted_count=accepted_count, rejected_count=rejected_count, paid_count=paid_count, pended_count=pended_count, parsed_at=utcnow(), raw_json=raw_json, ) s.add(row) s.commit() s.refresh(row) return row def list_277ca_acks(self) -> list[db.Two77caAck]: """Return every 277CA ACK row, newest first (auto-increment id desc).""" with db.SessionLocal()() as s: return ( s.query(db.Two77caAck) .order_by(db.Two77caAck.id.desc()) .all() ) def get_277ca_ack(self, ack_id: int) -> db.Two77caAck | None: """Return a single 277CA ACK row by id, or ``None`` if not found.""" with db.SessionLocal()() as s: return s.get(db.Two77caAck, ack_id) # -- SP17: encrypted DB backups ------------------------------------- def add_backup_pending(self, *, filename: str, backup_dir: str) -> db.DbBackup: """Insert a ``pending`` row for a backup that is about to start. The BackupService fills in ``status`` / ``size_bytes`` / ``db_fingerprint`` / ``table_count`` / ``completed_at`` after the encrypted blob lands on disk. """ with db.SessionLocal()() as s: row = db.DbBackup( filename=filename, backup_dir=backup_dir, size_bytes=0, db_fingerprint=None, table_count=0, created_at=utcnow(), completed_at=None, status="pending", error_message=None, ) s.add(row) s.commit() s.refresh(row) return row # -- manual reconciliation (T12) ----------------------------------- def list_unmatched(self, *, kind: str = "both") -> dict: """Return unmatched claims and/or remittances. An unmatched claim is one with ``matched_remittance_id IS NULL`` — either auto-match never paired it, or it was unpaired by ``manual_unmatch``. An unmatched remittance is one with ``claim_id IS NULL`` — symmetric FK on the remittance side; we update this in ``manual_match`` so the filter reflects the pair. Note: T10's ``reconcile.run`` only writes the claim-side FK (``Claim.matched_remittance_id``) when auto-pairing. Auto-matched remittances therefore still show as "unmatched" here until the pair is touched (re-ingest, manual unmatch + rematch). That gap is intentional for T12 — fixing it requires modifying T10. ``kind`` selects which side(s) to return: - "claims": only claims - "remittances": only remittances - "both": both (default) Returns ``{"claims": [...], "remittances": [...]}`` with the unused side always an empty list (never absent) so callers can unconditionally index. """ if kind not in ("claims", "remittances", "both"): raise ValueError( f"list_unmatched: unknown kind={kind!r} " "(expected 'claims', 'remittances', or 'both')" ) result: dict = {"claims": [], "remittances": []} with db.SessionLocal()() as s: if kind in ("claims", "both"): rows = ( s.query(Claim) .filter(Claim.matched_remittance_id.is_(None)) .order_by(Claim.id.asc()) .all() ) for r in rows: parsed_at = ( r.batch.parsed_at if r.batch is not None else r.service_date_from or utcnow() ) result["claims"].append( to_ui_claim_from_orm( r, batch_id=r.batch_id, parsed_at=parsed_at, ) ) if kind in ("remittances", "both"): rows = ( s.query(Remittance) .filter(Remittance.claim_id.is_(None)) .order_by(Remittance.id.asc()) .all() ) for r in rows: parsed_at = ( r.batch.parsed_at if r.batch is not None else r.received_at ) result["remittances"].append( to_ui_remittance_from_orm( r, batch_id=r.batch_id, parsed_at=parsed_at, ) ) return result def manual_match(self, claim_id: str, remit_id: str) -> dict: """Pair a claim with a remittance manually (operator override). Steps: 1. Load the claim; raise ``AlreadyMatchedError`` if it already has a match (we never silently overwrite an existing pair). 2. Load the remittance; raise ``LookupError`` if missing. 3. Compute the new claim state via ``reconcile.apply_payment`` (or ``apply_reversal`` for status codes 21/22). 4. Insert a ``Match`` row with ``strategy="manual"``. 5. Update the claim (``state``, ``matched_remittance_id``) AND the remittance (``claim_id``) so the symmetric FK reflects the pair — required for ``list_unmatched`` to drop them. 6. Record an ``ActivityEvent(kind="manual_match", ...)``. 7. Commit; return ``{"claim": , "match": }``. ``reconcile.apply_payment`` may return a noop (claim in terminal state); we surface that as ``InvalidStateError`` rather than silently pairing, because the operator clearly intended a state change. The T15 API endpoint maps this to a 409 Conflict. """ from cyclone import reconcile as _reconcile with db.SessionLocal()() as s: claim = s.get(Claim, claim_id) if claim is None: raise LookupError(f"claim {claim_id} not found") if claim.matched_remittance_id is not None: raise AlreadyMatchedError( f"claim {claim_id} already matched to " f"{claim.matched_remittance_id}" ) remit = s.get(Remittance, remit_id) if remit is None: raise LookupError(f"remittance {remit_id} not found") prior_state = claim.state if remit.is_reversal: intent = _reconcile.apply_reversal(claim, remit) else: intent = _reconcile.apply_payment( claim, remit, charge=claim.charge_amount, paid=remit.total_paid, status_code=remit.status_code, ) if intent.skipped or intent.new_state is None: current = ( claim.state.value if hasattr(claim.state, "value") else str(claim.state) ) raise InvalidStateError( current_state=current, activity_kind=intent.activity_kind, ) new_state = intent.new_state now = utcnow() s.add(Match( claim_id=claim_id, remittance_id=remit_id, strategy="manual", matched_at=now, prior_claim_state=prior_state, is_reversal=remit.is_reversal, )) claim.state = new_state claim.matched_remittance_id = remit_id # Symmetric FK update — see list_unmatched docstring for why # we don't rely on T10 to do this. remit.claim_id = claim_id # SP7: line-level reconciliation + claim-level CAS aggregate. # Skipped for reversals — they don't have SV1↔SVC line pairs. if not remit.is_reversal: _reconcile._reconcile_pair(s, claim, remit) s.add(ActivityEvent( ts=now, kind="manual_match", batch_id=remit.batch_id, claim_id=claim_id, remittance_id=remit_id, payload_json={ "strategy": "manual", "new_state": new_state.value, "prior_state": prior_state.value, "is_reversal": remit.is_reversal, }, )) s.commit() parsed_at = ( claim.batch.parsed_at if claim.batch is not None else now ) claim_dict = to_ui_claim_from_orm( claim, batch_id=claim.batch_id, parsed_at=parsed_at, ) matched_at_iso = now.isoformat().replace("+00:00", "Z") return { "claim": claim_dict, "match": { "strategy": "manual", "claimId": claim_id, "remittanceId": remit_id, "matchedAt": matched_at_iso, "isReversal": remit.is_reversal, "priorState": prior_state.value, "newState": new_state.value, }, } def manual_unmatch(self, claim_id: str) -> dict: """Unpair a previously matched claim and restore its prior state. Reverses ``manual_match`` (and auto-match as a side-effect of clearing the FK). Strategy: 1. Load the claim; raise ``NotMatchedError`` if it isn't currently matched. 2. Delete every ``Match`` row for the claim (there may be more than one — reversals create a 2nd row, see T10 spec). 3. Restore ``claim.state`` from the latest Match's ``prior_claim_state``; fall back to ``SUBMITTED`` when ``prior_claim_state`` is NULL (auto-match doesn't set it for non-reversal payments — see reconcile.run line ~278). 4. Clear ``claim.matched_remittance_id`` and the symmetric ``remit.claim_id`` so ``list_unmatched`` surfaces the pair again. 5. Record ``ActivityEvent(kind="manual_unmatch", ...)`` and commit. 6. Return ``{"claim": , "deletedMatches": }``. """ with db.SessionLocal()() as s: claim = s.get(Claim, claim_id) if claim is None: raise LookupError(f"claim {claim_id} not found") if claim.matched_remittance_id is None: raise NotMatchedError( f"claim {claim_id} has no active match" ) matches = ( s.query(Match) .filter(Match.claim_id == claim_id) .order_by(Match.matched_at.desc()) .all() ) if not matches: # Defensive: matched_remittance_id was set but no Match # rows exist. Shouldn't happen, but if it does, fall back # to clearing the FK and starting fresh. latest = None restored_state = ClaimState.SUBMITTED else: latest = matches[0] restored_state = ( latest.prior_claim_state if latest.prior_claim_state is not None else ClaimState.SUBMITTED ) deleted_count = len(matches) for m in matches: s.delete(m) claim.state = restored_state claim.matched_remittance_id = None # Clear the symmetric FK on the remittance so list_unmatched # surfaces the pair again. The remittance may have been # deleted between the match and this call — guard with a # get() so we don't blow up on a stale FK. if latest is not None: paired_remit = s.get(Remittance, latest.remittance_id) if paired_remit is not None: paired_remit.claim_id = None now = utcnow() s.add(ActivityEvent( ts=now, kind="manual_unmatch", claim_id=claim_id, payload_json={ "restored_state": restored_state.value, "deleted_matches": deleted_count, }, )) s.commit() parsed_at = ( claim.batch.parsed_at if claim.batch is not None else now ) claim_dict = to_ui_claim_from_orm( claim, batch_id=claim.batch_id, parsed_at=parsed_at, ) return { "claim": claim_dict, "deletedMatches": deleted_count, } # ------------------------------------------------------------------ # SP9: providers / payers / payer_configs / clearhouse # ------------------------------------------------------------------ def list_providers(self, *, is_active: bool | None = True) -> list[Provider]: """List providers. ``is_active=None`` returns all.""" from cyclone.db import Provider as ProviderORM from cyclone.providers import Provider with db.SessionLocal()() as s: q = s.query(ProviderORM) if is_active is not None: q = q.filter(ProviderORM.is_active == (1 if is_active else 0)) rows = q.order_by(ProviderORM.label).all() return [Provider.model_validate(_provider_orm_to_dict(r)) for r in rows] def get_provider(self, npi: str) -> Provider | None: from cyclone.db import Provider as ProviderORM from cyclone.providers import Provider with db.SessionLocal()() as s: row = s.get(ProviderORM, npi) return Provider.model_validate(_provider_orm_to_dict(row)) if row else None def upsert_provider(self, provider: Provider) -> Provider: from cyclone.db import Provider as ProviderORM with db.SessionLocal()() as s: row = s.get(ProviderORM, provider.npi) now = utcnow().isoformat() if row is None: row = ProviderORM( npi=provider.npi, label=provider.label, legal_name=provider.legal_name, tax_id=provider.tax_id, taxonomy_code=provider.taxonomy_code, address_line1=provider.address_line1, address_line2=provider.address_line2, city=provider.city, state=provider.state, zip=provider.zip, is_active=1 if provider.is_active else 0, created_at=provider.created_at.isoformat(), updated_at=now, ) s.add(row) else: row.label = provider.label row.legal_name = provider.legal_name row.tax_id = provider.tax_id row.taxonomy_code = provider.taxonomy_code row.address_line1 = provider.address_line1 row.address_line2 = provider.address_line2 row.city = provider.city row.state = provider.state row.zip = provider.zip row.is_active = 1 if provider.is_active else 0 row.updated_at = now s.commit() return self.get_provider(provider.npi) # type: ignore[return-value] def list_payers(self, *, is_active: bool | None = True) -> list[Payer]: from cyclone.db import Payer as PayerORM from cyclone.providers import Payer with db.SessionLocal()() as s: q = s.query(PayerORM) if is_active is not None: q = q.filter(PayerORM.is_active == (1 if is_active else 0)) rows = q.order_by(PayerORM.payer_id).all() return [Payer.model_validate(_payer_orm_to_dict(r)) for r in rows] def get_payer_config(self, payer_id: str, transaction_type: str) -> dict | None: from cyclone.db import PayerConfigORM with db.SessionLocal()() as s: row = s.get(PayerConfigORM, (payer_id, transaction_type)) return dict(row.config_json) if row else None def get_clearhouse(self) -> Clearhouse | None: from cyclone.db import ClearhouseORM from cyclone.providers import Clearhouse with db.SessionLocal()() as s: row = s.get(ClearhouseORM, 1) if row is None: return None return Clearhouse.model_validate({ "id": 1, "name": row.name, "tpid": row.tpid, "submitter_name": row.submitter_name, "submitter_id_qual": row.submitter_id_qual, "submitter_contact_name": row.submitter_contact_name, "submitter_contact_email": row.submitter_contact_email, "filename_block": dict(row.filename_block_json), "sftp_block": dict(row.sftp_block_json), "updated_at": row.updated_at, }) def ensure_clearhouse_seeded(self) -> None: """Insert the default clearhouse singleton + 3 providers + CO_TXIX payer if they don't exist. Idempotent. Called from the API lifespan.""" from cyclone.db import ClearhouseORM, Payer as PayerORM, PayerConfigORM, Provider as ProviderORM from cyclone.providers import Clearhouse with db.SessionLocal()() as s: if s.get(ClearhouseORM, 1) is None: ch = Clearhouse( id=1, name="dzinesco", tpid="11525703", submitter_name="Dzinesco", submitter_id_qual="46", submitter_contact_name="Tyler Martinez", submitter_contact_email="tyler@dzinesco.com", filename_block={ "tz": "America/Denver", "outbound_template": "{tpid}-{tx}-{ts_mt}-1of1.{ext}", "inbound_template": "TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12", }, sftp_block={ "host": "mft.gainwelltechnologies.com", "port": 22, "username": "colorado-fts\\coxix_prod_11525703", "paths": { "outbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE", "inbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE", }, "stub": True, "staging_dir": "./var/sftp/staging", "poll_seconds": 300, "auth": {"method": "keychain", "secret_ref": "sftp.gainwell.password"}, }, updated_at=utcnow(), ) s.add(ClearhouseORM( id=1, name=ch.name, tpid=ch.tpid, submitter_name=ch.submitter_name, submitter_id_qual=ch.submitter_id_qual, submitter_contact_name=ch.submitter_contact_name, submitter_contact_email=ch.submitter_contact_email, filename_block_json=ch.filename_block.model_dump(), sftp_block_json=ch.sftp_block.model_dump(), updated_at=ch.updated_at.isoformat(), )) # Seed 3 providers (idempotent) from cyclone.providers import Provider now = utcnow().isoformat() for npi, label in [ ("1881068062", "Montrose"), ("1851446637", "Delta"), ("1467507269", "Salida"), ]: if s.get(ProviderORM, npi) is None: s.add(ProviderORM( npi=npi, label=label, legal_name="TOC, Inc.", tax_id="721587149", taxonomy_code="251E00000X", address_line1="1100 East Main St", address_line2="Suite A", city="Montrose", state="CO", zip="814014063", is_active=1, created_at=now, updated_at=now, )) # Seed CO_TXIX payer (idempotent) if s.get(PayerORM, "CO_TXIX") is None: s.add(PayerORM( payer_id="CO_TXIX", name="Colorado Medical Assistance Program", receiver_name="COLORADO MEDICAL ASSISTANCE PROGRAM", receiver_id="COMEDASSISTPROG", is_active=1, created_at=now, updated_at=now, )) # 837P config block s.add(PayerConfigORM( payer_id="CO_TXIX", transaction_type="837P", config_json={ "submitter_name": "Dzinesco", "submitter_contact_name": "Tyler Martinez", "submitter_contact_email": "tyler@dzinesco.com", "receiver_name": "COLORADO MEDICAL ASSISTANCE PROGRAM", "receiver_id_qualifier": "46", "receiver_id": "COMEDASSISTPROG", "bht06_allowed": ["CH", "RP"], "bht06_default": "CH", "sbr09_default": "MC", "sbr09_allowed": ["MC", "16", "MA", "MB", "ZZ"], "payer_id_qualifier": "PI", "payer_id": "CO_TXIX", "pwk_supported": False, "cas_2320_group_allowed": False, "claim_type_codes": {"11": "Office", "12": "Home", "99": "Other"}, }, updated_at=now, )) # 835 config block s.add(PayerConfigORM( payer_id="CO_TXIX", transaction_type="835", config_json={ "expected_payer_tax_ids": [ "81-1725341", "811725341", "84-0644739", "840644739", "1811725341", ], "expected_payer_health_plan_id": "7912900843", "payer_name_pattern": "^CO_(TXIX|BHA)$", }, updated_at=now, )) s.commit() # --------------------------------------------------------------------------- # SP9: ORM-to-Pydantic conversion helpers # --------------------------------------------------------------------------- def _provider_orm_to_dict(row) -> dict: return { "npi": row.npi, "label": row.label, "legal_name": row.legal_name, "tax_id": row.tax_id, "taxonomy_code": row.taxonomy_code, "address_line1": row.address_line1, "address_line2": row.address_line2, "city": row.city, "state": row.state, "zip": row.zip, "is_active": bool(row.is_active), "created_at": row.created_at, "updated_at": row.updated_at, } def _payer_orm_to_dict(row) -> dict: return { "payer_id": row.payer_id, "name": row.name, "receiver_name": row.receiver_name, "receiver_id": row.receiver_id, "is_active": bool(row.is_active), "created_at": row.created_at, "updated_at": row.updated_at, } # Module-level singleton — same import path the old InMemoryStore used. store = CycloneStore()