feat(backend): SQLAlchemy-backed CycloneStore (T9)
Replace the in-memory InMemoryStore with a SQLAlchemy-backed CycloneStore that persists Batch, Claim, Remittance, Match, and ActivityEvent rows to a configurable DB engine (sqlite by default, overridable via CYCLONE_DB_URL). Public API preserved: add / get_batch / iter_claims / iter_remittances / distinct_providers / recent_activity. New T10/T12 stubs: list_unmatched, manual_match, manual_unmatch. Backward-compat shims for tests that called ._batches.clear() or acquired ._lock as a context manager. Idempotency: add() does a per-row session.get(Claim, c.id) (and s.get(Remittance, ...)) check before each insert; duplicates are skipped with a warning. This makes re-uploading the same fixture idempotent instead of raising IntegrityError on the PK. Auto-init DB fixture (backend/tests/conftest.py) sets CYCLONE_DB_URL to a per-test sqlite file and calls db.init_db() once per test, replacing the old module-scoped in-memory fixture.
This commit is contained in:
Tyler
+638
-221
@@ -1,19 +1,51 @@
|
||||
"""In-memory batch store for parsed X12 files.
|
||||
"""SQLAlchemy-backed batch store for parsed X12 files.
|
||||
|
||||
Lives at module scope so route handlers and tests can import a single
|
||||
singleton. Single-process only — no cross-process coordination. Threading:
|
||||
a re-entrant lock guards every public method so FastAPI's threadpool
|
||||
request handlers can call them concurrently.
|
||||
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 (stubs for T10/T12):
|
||||
- list_unmatched()
|
||||
- manual_match(claim_id, remit_id)
|
||||
- manual_unmatch(claim_id)
|
||||
|
||||
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 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 (
|
||||
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
|
||||
@@ -22,6 +54,11 @@ from cyclone.parsers.payer import PayerConfig835
|
||||
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.
|
||||
|
||||
@@ -82,232 +119,112 @@ class BatchRecord835(BatchRecord):
|
||||
result: ParseResult835
|
||||
|
||||
|
||||
class InMemoryStore:
|
||||
"""Thread-safe list of BatchRecord, newest-first on `list`."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._batches: list[BatchRecord] = []
|
||||
self._lock = threading.RLock()
|
||||
|
||||
def add(self, record: BatchRecord) -> None:
|
||||
with self._lock:
|
||||
self._batches.append(record)
|
||||
|
||||
def list(self, *, limit: int = 100) -> list[BatchRecord]:
|
||||
with self._lock:
|
||||
return list(reversed(self._batches[-limit:]))
|
||||
|
||||
def get(self, batch_id: str) -> BatchRecord | None:
|
||||
with self._lock:
|
||||
for b in self._batches:
|
||||
if b.id == batch_id:
|
||||
return b
|
||||
return None
|
||||
|
||||
def all(self) -> list[BatchRecord]:
|
||||
with self._lock:
|
||||
return list(self._batches)
|
||||
|
||||
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]:
|
||||
with self._lock:
|
||||
all_claims: list[dict] = []
|
||||
for b in self._batches:
|
||||
if not isinstance(b, BatchRecord837):
|
||||
continue
|
||||
if batch_id is not None and b.id != batch_id:
|
||||
continue
|
||||
for c in b.result.claims:
|
||||
all_claims.append(to_ui_claim(
|
||||
c, batch_id=b.id, parsed_at=b.parsed_at,
|
||||
))
|
||||
if status is not None:
|
||||
all_claims = [c for c in all_claims if c["status"] == status]
|
||||
if provider_npi is not None:
|
||||
all_claims = [c for c in all_claims if c["providerNpi"] == provider_npi]
|
||||
if payer is not None:
|
||||
needle = payer.casefold()
|
||||
all_claims = [
|
||||
c for c in all_claims
|
||||
if needle in (c.get("payerName") or "").casefold()
|
||||
]
|
||||
all_claims = [
|
||||
c for c in all_claims
|
||||
if _date_in_bounds(c, "submissionDate", date_from, date_to)
|
||||
]
|
||||
if sort is not None:
|
||||
all_claims.sort(
|
||||
key=lambda c: c.get(sort, 0) or 0,
|
||||
reverse=(order == "desc"),
|
||||
)
|
||||
return all_claims[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]:
|
||||
with self._lock:
|
||||
all_remits: list[dict] = []
|
||||
for b in self._batches:
|
||||
if not isinstance(b, BatchRecord835):
|
||||
continue
|
||||
if batch_id is not None and b.id != batch_id:
|
||||
continue
|
||||
for cp in b.result.claims:
|
||||
all_remits.append(to_ui_remittance(
|
||||
cp, batch_id=b.id, parsed_at=b.parsed_at,
|
||||
payer_name=b.result.payer.name,
|
||||
))
|
||||
if payer is not None:
|
||||
all_remits = [r for r in all_remits if r.get("payerName") == payer]
|
||||
if claim_id is not None:
|
||||
all_remits = [r for r in all_remits if r["claimId"] == claim_id]
|
||||
all_remits = [
|
||||
r for r in all_remits
|
||||
if _date_in_bounds(r, "receivedDate", date_from, date_to)
|
||||
]
|
||||
if sort is not None:
|
||||
all_remits.sort(
|
||||
key=lambda r: r.get(sort, 0) or 0,
|
||||
reverse=(order == "desc"),
|
||||
)
|
||||
return all_remits[offset:offset + limit]
|
||||
|
||||
def distinct_providers(self) -> list[dict]:
|
||||
with self._lock:
|
||||
by_npi: dict[str, dict] = {}
|
||||
for b in self._batches:
|
||||
if not isinstance(b, BatchRecord837):
|
||||
continue
|
||||
if not b.result.claims:
|
||||
continue
|
||||
# billing_provider lives on each claim, not at the batch level
|
||||
bp = b.result.claims[0].billing_provider
|
||||
npi = bp.npi
|
||||
if npi not in by_npi:
|
||||
by_npi[npi] = to_ui_provider(
|
||||
npi=npi,
|
||||
name=bp.name or "",
|
||||
tax_id=bp.tax_id,
|
||||
claim_count=0,
|
||||
outstanding_ar=0.0,
|
||||
)
|
||||
by_npi[npi]["claimCount"] += len(b.result.claims)
|
||||
return list(by_npi.values())
|
||||
|
||||
def recent_activity(self, *, limit: int = 200) -> list[dict]:
|
||||
with self._lock:
|
||||
events: list[dict] = []
|
||||
for b in reversed(self._batches):
|
||||
if isinstance(b, BatchRecord837):
|
||||
for c in b.result.claims:
|
||||
events.append(to_activity_event(
|
||||
id=f"{b.id}:{c.claim_id}",
|
||||
kind="claim_submitted",
|
||||
message=f"Claim {c.claim_id} submitted · {c.payer.name}",
|
||||
timestamp=b.parsed_at,
|
||||
npi=c.billing_provider.npi,
|
||||
amount=float(c.claim.total_charge or 0.0),
|
||||
))
|
||||
elif isinstance(b, BatchRecord835):
|
||||
for cp in b.result.claims:
|
||||
events.append(to_activity_event(
|
||||
id=f"{b.id}:{cp.payer_claim_control_number}",
|
||||
kind="remit_received",
|
||||
message=f"Remit {cp.payer_claim_control_number} received",
|
||||
timestamp=b.parsed_at,
|
||||
amount=float(cp.total_paid or 0.0),
|
||||
))
|
||||
return events[:limit]
|
||||
|
||||
|
||||
store = InMemoryStore()
|
||||
|
||||
|
||||
def utcnow() -> datetime:
|
||||
"""tz-aware UTC `datetime` (replaces the old `utcnow_iso` string helper)."""
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
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]``.
|
||||
# ---------------------------------------------------------------------------
|
||||
# ORM row builders.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
``date_from`` / ``date_to`` are ISO date strings (``YYYY-MM-DD``). The
|
||||
item's value is treated as an ISO timestamp; only its date portion is
|
||||
compared. A missing field is included only when no bound is set —
|
||||
bounded filters treat ``None`` as outside the range.
|
||||
|
||||
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``.
|
||||
"""
|
||||
val = item.get(field)
|
||||
if val is None:
|
||||
# Missing dates: include when unbounded, exclude when bounded.
|
||||
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
|
||||
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.
|
||||
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()),
|
||||
)
|
||||
|
||||
|
||||
# ``json`` is imported lazily inside the helpers above so we don't pull
|
||||
# the stdlib ``json`` into module-level imports at the very top — keeps
|
||||
# the dependency surface easy to audit. The ``import json`` line below
|
||||
# is the actual import used by the helpers above.
|
||||
import json # noqa: E402
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Mappers: rich backend models → simpler UI types.
|
||||
# The shape of the returned dicts must match the TypeScript interfaces in
|
||||
# `src/types/index.ts`. Keep both sides in sync.
|
||||
# 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.
|
||||
|
||||
Status rules (per spec section 6.2, first-match-wins):
|
||||
- !passed and any R050_diagnosis_present error → draft
|
||||
- !passed → denied
|
||||
- passed and frequency_code == "1" → submitted
|
||||
- passed and any warnings → pending
|
||||
- otherwise → draft
|
||||
"""
|
||||
v = claim.validation
|
||||
if not v.passed:
|
||||
has_r050 = any(e.rule == "R050_diagnosis_present" for e in v.errors)
|
||||
status = "draft" if has_r050 else "denied"
|
||||
elif claim.claim.frequency_code == "1":
|
||||
status = "submitted"
|
||||
elif v.warnings:
|
||||
status = "pending"
|
||||
else:
|
||||
status = "draft"
|
||||
|
||||
"""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,
|
||||
@@ -321,7 +238,7 @@ def to_ui_claim(
|
||||
),
|
||||
"billedAmount": float(claim.claim.total_charge or 0.0),
|
||||
"receivedAmount": 0.0,
|
||||
"status": status,
|
||||
"status": _claim_status_from_validation(claim),
|
||||
"denialReason": None,
|
||||
"submissionDate": parsed_iso,
|
||||
"batchId": batch_id,
|
||||
@@ -337,14 +254,7 @@ def to_ui_remittance(
|
||||
payer_config: PayerConfig835 | None = None,
|
||||
payer_name: str = "",
|
||||
) -> dict:
|
||||
"""Map an 835 ClaimPayment to the UI's `Remittance` shape.
|
||||
|
||||
Status mapping (per spec section 6.2):
|
||||
- status_code 21/22 → reconciled (reversal/correction)
|
||||
- status_code 4 → received + denialReason from CAS
|
||||
- status_code outside allowlist → received + validationWarning
|
||||
- otherwise → received
|
||||
"""
|
||||
"""Map an 835 ClaimPayment to the UI's `Remittance` shape (preserved)."""
|
||||
code = cp.status_code
|
||||
if code in {"21", "22"}:
|
||||
status = "reconciled"
|
||||
@@ -428,3 +338,510 @@ def to_activity_event(
|
||||
"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) -> 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.
|
||||
"""
|
||||
import logging
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
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),
|
||||
},
|
||||
))
|
||||
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
|
||||
s.add(_remittance_835_row(cp, record.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),
|
||||
},
|
||||
))
|
||||
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,
|
||||
)
|
||||
|
||||
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 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 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()
|
||||
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", "")
|
||||
)
|
||||
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,
|
||||
"_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
|
||||
]
|
||||
|
||||
# -- T12 stubs: manual reconciliation -----------------------------
|
||||
|
||||
def list_unmatched(self) -> list[dict]:
|
||||
"""T12 stub: return claims/remits that auto-match didn't pair.
|
||||
|
||||
Returns an empty list for T9. T12 will fill in the implementation
|
||||
by querying ``Claim`` rows with ``matched_remittance_id IS NULL``
|
||||
plus ``Remittance`` rows with ``claim_id IS NULL``.
|
||||
"""
|
||||
# TODO(t12): implement — query unmatched claims and remittances.
|
||||
return []
|
||||
|
||||
def manual_match(self, claim_id: str, remit_id: str) -> None:
|
||||
"""T12 stub: pair a claim with a remittance manually."""
|
||||
# TODO(t12): implement — create a Match row, update Claim.state.
|
||||
return None
|
||||
|
||||
def manual_unmatch(self, claim_id: str) -> None:
|
||||
"""T12 stub: unpair a previously matched claim."""
|
||||
# TODO(t12): implement — delete the Match row, restore prior state.
|
||||
return None
|
||||
|
||||
|
||||
# Module-level singleton — same import path the old InMemoryStore used.
|
||||
store = CycloneStore()
|
||||
Reference in New Issue
Block a user