docs: surface SP14 (Payer-Rejected UI + acknowledge) + SP15 (key rotation) #3

Closed
tyler wants to merge 14 commits from docs/sp14-15-readme-sync into main
13 changed files with 2187 additions and 1591 deletions
+156 -4
View File
@@ -155,6 +155,7 @@ parses and rejects claims, the inbox reflects the new
| POST | `/api/inbox/candidates/{remit_id}/match` | Manual match. `409` if the claim state moved out from under us. | | POST | `/api/inbox/candidates/{remit_id}/match` | Manual match. `409` if the claim state moved out from under us. |
| POST | `/api/inbox/candidates/dismiss` | `{pairs: [{claim_id, remit_id}]}`. Session-scoped. | | POST | `/api/inbox/candidates/dismiss` | `{pairs: [{claim_id, remit_id}]}`. Session-scoped. |
| POST | `/api/inbox/rejected/resubmit` | `{claim_ids: [...]}`. `200` with `conflicts` for non-rejected. | | POST | `/api/inbox/rejected/resubmit` | `{claim_ids: [...]}`. `200` with `conflicts` for non-rejected. |
| POST | `/api/inbox/payer-rejected/acknowledge` | `{claim_ids: [...], actor: "..."}`. Bulk-acknowledge payer rejections. Idempotent. `200` with `transitioned` / `already_acked` / `not_found` / `not_rejected` counts. Writes an audit event per transition; never overwrites the underlying 277CA rejection. |
| GET | `/api/inbox/export.csv?lane=<lane>` | Streams CSV of the lane's rows. | | GET | `/api/inbox/export.csv?lane=<lane>` | Streams CSV of the lane's rows. |
## Outbound 837 Serializer ## Outbound 837 Serializer
@@ -321,8 +322,9 @@ both be true for the same claim.
The `audit_log` table is the canonical record of every state transition The `audit_log` table is the canonical record of every state transition
the system has ever observed — claim lifecycle, reconciliation the system has ever observed — claim lifecycle, reconciliation
decisions, config reloads, SFTP submissions, 277CA rejects. SP11 made decisions, config reloads, SFTP submissions, 277CA rejects, Payer-Rejected
it tamper-evident: every row carries a SHA-256 hash of acknowledgements, SQLCipher key rotations. SP11 made it tamper-evident:
every row carries a SHA-256 hash of
`(prev_hash || row_payload)`, forming a chain back to a genesis row. `(prev_hash || row_payload)`, forming a chain back to a genesis row.
Any `INSERT`, `UPDATE`, or `DELETE` that breaks the chain is detectable Any `INSERT`, `UPDATE`, or `DELETE` that breaks the chain is detectable
in a single walk. in a single walk.
@@ -357,6 +359,40 @@ operator has created the Keychain entry on first run. See
for the one-time setup recipe and the HIPAA Security Rule §164.312(a)(2)(iv) for the one-time setup recipe and the HIPAA Security Rule §164.312(a)(2)(iv)
mapping. mapping.
### Key rotation
The DB encryption key is rotated in place via `POST /api/admin/db/rotate-key`.
The handler:
1. Generates a fresh 256-bit CSPRNG key (`db_crypto.generate_db_key()`).
2. Persists the new key to the Keychain under the same
`cyclone.db.key` account (overwriting the old one).
3. Disposes the SQLAlchemy engine so pooled connections release the
file (SQLCipher refuses to `PRAGMA rekey` while another connection
holds the DB).
4. Opens with the old key, issues `PRAGMA rekey`, and verifies the
schema survived (table-count sanity check).
5. Rebuilds the engine with the new key.
6. Writes a `db.key_rotated` audit event carrying the SHA-256
fingerprints of the old and new keys (first 8 hex chars) plus the
post-rotation `table_count`.
When SQLCipher is enabled the engine uses SQLAlchemy's `NullPool`
instead of the default `QueuePool`. `QueuePool` returns connections to
a shared queue that any thread can pull from, which breaks SQLCipher's
thread affinity. `NullPool` trades connection reuse for thread safety
— the only correct behavior under FastAPI's per-request threadpool.
The rotation endpoint is serialized with a module-level
`threading.Lock` (one rotation in flight at a time), returns `409` if
a rotation is already running, `400` if encryption is not enabled,
and `503` with a `reason` on `PRAGMA rekey` or Keychain failure so the
operator can take the next step without parsing the traceback.
| Method | Path | Notes |
| ------ | --------------------------------- | -------------------------------------------------------------- |
| POST | `/api/admin/db/rotate-key` | Rotate the SQLCipher key in place. Audit-logged. |
## SFTP Wire-Up (paramiko) ## SFTP Wire-Up (paramiko)
The `clearhouse.submit` endpoint uses `paramiko` to push a batch of The `clearhouse.submit` endpoint uses `paramiko` to push a batch of
@@ -382,6 +418,58 @@ was a one-file change (`sftp_paramiko.py` replacing `sftp_stub.py`).
the `paramiko` extras aren't installed so the test suite stays green on the `paramiko` extras aren't installed so the test suite stays green on
Linux dev boxes. Linux dev boxes.
## Inbound MFT Scheduler (SP16)
The dzinesco SFTP server accepts inbound files (835 remits, 999/TA1
acks, 271 eligibility responses, 277/277CA claim acknowledgments) that
need to be pulled on a cadence, parsed, and persisted. SP16 ships the
**operator-facing control surface** for the polling loop, even though
the actual background scheduler is still in flight:
| Method | Path | Notes |
| ------ | --------------------------------------------- | ---------------------------------------------------------------- |
| POST | `/api/admin/scheduler/start` | Start the background MFT polling loop (idempotent). |
| POST | `/api/admin/scheduler/stop` | Stop the background loop. Waits for the current tick to finish. |
| GET | `/api/admin/scheduler/status` | Running state + last tick summary. |
| POST | `/api/admin/scheduler/tick` | Force a single poll cycle. Coalesces with an in-flight tick. |
| GET | `/api/admin/scheduler/processed-files` | Recently processed inbound files. `?status=ok|error|skipped|pending`, `?limit=` (1500, default 50). |
The endpoints read from a `cyclone.scheduler` module and a
`cyclone.db.ProcessedInboundFile` ORM model that are being landed
alongside SP16 — the admin router's import surface is in place, the
implementation is in progress. The control surface is forward-compatible
with the eventual scheduler: starting / stopping / ticking / listing
processed files will all work as soon as the loop lands.
The `paramiko` SFTP client from [SP13](#sftp-wire-up-paramiko) is the
underlying transport; the scheduler reuses it to list + download files
under the inbound SFTP path, then dispatches each downloaded file to
the matching parser based on the `<FileType>` token in the filename
(see [SP13 file-naming template](#sftp-wire-up-paramiko)).
## Batches, Reconciliation, and Activity
These read/write endpoints are the core operator surface for browsing
the parsed record and acting on matches. They predate the per-SP
endpoint reference sections in the **Roadmap** and are listed here in
one place so the route inventory stays discoverable.
| Method | Path | Notes |
| ------ | --------------------------------------------- | ---------------------------------------------------------------- |
| GET | `/api/batches` | All parsed batches, newest first. `?limit=` (11000, default 100). |
| GET | `/api/batches/{batch_id}` | One batch detail (envelope, claims / remits, validation summary). |
| GET | `/api/batch-diff?a=<id>&b=<id>` | Side-by-side diff of two batches: `added` / `removed` / `changed` claims + envelope metadata for each. Both query params required. |
| GET | `/api/reconciliation/unmatched` | `{"claims": [...], "remittances": [...]}` — every Claim with no paired Remittance and vice versa. The two lists are always present (empty list, never absent) so the UI can index unconditionally. |
| POST | `/api/reconciliation/match` | Body `{claim_id, remit_id}`. Manually pair a claim with a remit. `400` on missing ids, `404` on unknown id, `409` on already-matched or terminal-state claim. |
| POST | `/api/reconciliation/unmatch` | Body `{claim_id}`. Remove the current match and reset the claim to `submitted`. `404` on unknown id, `409` on no current match. |
| GET | `/api/providers` | Distinct providers across parsed claims. `?npi=`, `?state=`, `?limit=`, `?offset=`. Distinct from `/api/config/providers/{npi}` (SP9 config table) — this endpoint surfaces providers derived from the parsed claim stream. |
| GET | `/api/activity` | Recent activity events. `?kind=`, `?since=`, `?limit=` (1500, default 200). Powers the Activity page; the streaming counterpart `/api/activity/stream` is documented under **Live updates**. |
The per-SP endpoint blocks at the bottom of the **Roadmap** cover the
SP-specific routes (parse, 999/TA1/277CA, Inbox, claim drawer, line
reconciliation, outbound 837, multi-payer config, audit log, 277CA,
key rotation, payer-rejected acknowledge).
## Persistence ## Persistence
Parsed batches, claims, remittances, matches, 277CA rejections, Parsed batches, claims, remittances, matches, 277CA rejections,
@@ -415,8 +503,31 @@ backup API).
. .
├── backend/ ├── backend/
│ ├── src/cyclone/ │ ├── src/cyclone/
│ │ ├── api.py # FastAPI app, GET + parse routes, /api/{resource}/stream │ │ ├── api.py # FastAPI app + middleware; mounts api_routers/* sub-apps
│ │ ├── api_helpers.py # NDJSON / content-negotiation / live-tail helpers │ │ ├── api_helpers.py # NDJSON / content-negotiation / live-tail helpers
│ │ ├── api_routers/ # FastAPI APIRouters extracted from api.py
│ │ │ ├── health.py # GET /api/health
│ │ │ ├── acks.py # GET /api/acks, /api/acks/{id}
│ │ │ ├── ta1_acks.py # GET /api/ta1-acks, /api/ta1-acks/{id}
│ │ │ ├── activity.py # GET /api/activity, /api/activity/stream
│ │ │ ├── batches.py # GET /api/batches, /api/batches/{id}
│ │ │ ├── claims.py # GET /api/claims, /api/claims/stream, /api/claims/{id},
│ │ │ │ # /api/claims/{id}/serialize-837, /api/claims/{id}/line-reconciliation
│ │ │ ├── remittances.py # GET /api/remittances, /api/remittances/stream, /api/remittances/{id}
│ │ │ ├── providers.py # GET /api/providers
│ │ │ ├── inbox.py # GET /api/inbox/lanes, POST /api/inbox/* (match/dismiss/acknowledge/resubmit),
│ │ │ │ # GET /api/inbox/export.csv
│ │ │ ├── reconciliation.py # GET /api/reconciliation/unmatched, /api/batch-diff,
│ │ │ │ # POST /api/reconciliation/match, /api/reconciliation/unmatch
│ │ │ ├── eligibility.py # POST /api/eligibility/request, /api/eligibility/parse-271
│ │ │ ├── clearhouse.py # GET /api/clearhouse, POST /api/clearhouse/submit
│ │ │ ├── config.py # /api/config/providers[/...], /api/config/payers[/...], /api/admin/reload-config
│ │ │ ├── parse_835.py # POST /api/parse-835
│ │ │ ├── parse_837.py # POST /api/parse-837
│ │ │ ├── parse_999.py # POST /api/parse-999
│ │ │ ├── parse_ta1.py # POST /api/parse-ta1
│ │ │ └── admin.py # /api/admin/audit-log[/verify], /api/admin/db/rotate-key,
│ │ │ # /api/admin/scheduler/* (SP16)
│ │ ├── pubsub.py # in-process EventBus (drop-oldest, per-kind fan-out) │ │ ├── pubsub.py # in-process EventBus (drop-oldest, per-kind fan-out)
│ │ ├── store.py # CycloneStore, mappers, publish-on-write │ │ ├── store.py # CycloneStore, mappers, publish-on-write
│ │ ├── db.py # SQLAlchemy engine, session factory, ORM models │ │ ├── db.py # SQLAlchemy engine, session factory, ORM models
@@ -473,7 +584,7 @@ backup API).
## Roadmap ## Roadmap
Sub-projects 2 through 13 are **shipped**. See the [completeness Sub-projects 2 through 15 are **shipped**. See the [completeness
review](docs/reviews/2026-06-20-cyclone-completeness-review.md) for review](docs/reviews/2026-06-20-cyclone-completeness-review.md) for
the honest gap analysis against the industry definition of a HIPAA the honest gap analysis against the industry definition of a HIPAA
clearinghouse — the short version is that the local-only, clearinghouse — the short version is that the local-only,
@@ -484,6 +595,19 @@ scope.
Shipped sub-projects (most recent first): Shipped sub-projects (most recent first):
- **Sub-project 15 (shipped) — SQLCipher key rotation.** In-place
rotation via `PRAGMA rekey`, serialized through a module-level
`threading.Lock` and a SQLAlchemy `NullPool` to keep SQLCipher
thread-affine under FastAPI's per-request threadpool. Writes a
`db.key_rotated` audit event with old + new key fingerprints and
post-rotation `table_count`. See
[Encryption at Rest — Key rotation](#key-rotation).
- **Sub-project 14 (shipped) — 5-lane Inbox UI.** The Payer-Rejected
lane is now rendered in the Inbox alongside Rejected / Candidates /
Unmatched / Done today. New bulk action
`POST /api/inbox/payer-rejected/acknowledge` drops claims from the
working surface without erasing the original 277CA rejection event
(audit log stays intact, SP11).
- **Sub-project 13 (shipped) — SFTP wire-up.** `paramiko`-backed - **Sub-project 13 (shipped) — SFTP wire-up.** `paramiko`-backed
`SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint `SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint
actually pushes to actually pushes to
@@ -752,6 +876,34 @@ the one-time setup recipe.
`mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`. `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`.
SFTP credentials are fetched from the macOS Keychain at call time. SFTP credentials are fetched from the macOS Keychain at call time.
### SP14 endpoints (5-lane Inbox UI + acknowledge)
- `GET /api/inbox/lanes` — same shape as SP6; the `payer_rejected`
lane payload now also includes
`payer_rejected_acknowledged_at` + `payer_rejected_acknowledged_actor`
per row so the UI can badge acknowledged claims (forward-compat for
a future "Recently acknowledged" view).
- `POST /api/inbox/payer-rejected/acknowledge` — bulk-acknowledge
Payer-Rejected claims. Body: `{claim_ids: [...], actor: "..."}`.
Idempotent. Response: `200` with
`{transitioned, already_acked, not_found, not_rejected}`. Writes an
`inbox.payer_rejected_acknowledged` audit event per transition.
Acknowledgement hides a claim from the lane but never overwrites
`payer_rejected_at` / `payer_rejected_reason` /
`payer_rejected_by_277ca_id` — the original 277CA evidence stays
intact in the audit log.
### SP15 endpoints (SQLCipher key rotation)
- `POST /api/admin/db/rotate-key` — rotate the SQLCipher DB key in
place. Generates a fresh 256-bit key, writes it to the Keychain
(overwriting `cyclone.db.key`), disposes the engine, issues
`PRAGMA rekey`, verifies the schema, rebuilds the engine. Writes
a `db.key_rotated` audit event with old + new key fingerprints
and `table_count`. Returns `409` when a rotation is already in
flight, `400` when encryption is not enabled, `503` with a
`reason` on `PRAGMA rekey` or Keychain failure.
## License ## License
No license file yet; this is internal-use software. Add a `LICENSE` file No license file yet; this is internal-use software. Add a `LICENSE` file
+62 -1587
View File
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,97 @@
"""``/api/activity`` — list & live-tail of recorded Activity events.
The Activity log is the cross-resource audit feed: every parser write,
inbox state change, match/dismiss/resubmit, clearhouse submission, etc.
records one row keyed by ``kind``. The list endpoint returns the most
recent ``limit`` events with optional ``kind`` / ``since`` filters; the
stream endpoint emits a snapshot followed by live updates from the
``EventBus``.
The default ``limit`` on the list endpoint is 200 (matches the prior
inline behavior); the stream endpoint defaults to 50 because activity is
high-volume — callers usually want the most recent handful, not a full
replay.
"""
from __future__ import annotations
from typing import Any, AsyncIterator
from fastapi import APIRouter, Query, Request
from fastapi.responses import StreamingResponse
from cyclone.api_helpers import (
ndjson_line,
ndjson_stream_list,
tail_events,
wants_ndjson,
)
from cyclone.pubsub import EventBus
from cyclone.store import store
router = APIRouter()
@router.get("/api/activity")
def list_activity_endpoint(
request: Request,
kind: str | None = Query(None),
since: str | None = Query(None),
limit: int = Query(200, ge=1, le=500),
) -> Any:
"""Return the recent Activity log, newest first.
Optional ``kind`` filters by ``event["kind"]`` (e.g. ``claim_submitted``).
Optional ``since`` is an inclusive ISO timestamp lower bound applied
to ``event["timestamp"]``.
"""
events = store.recent_activity(limit=limit)
if kind is not None:
events = [e for e in events if e["kind"] == kind]
if since is not None:
events = [e for e in events if e["timestamp"] >= since]
total = len(events)
has_more = False
if wants_ndjson(request):
return StreamingResponse(
ndjson_stream_list(events, total, total, has_more),
media_type="application/x-ndjson",
)
return {
"items": events,
"total": total,
"returned": total,
"has_more": has_more,
}
@router.get("/api/activity/stream")
async def activity_stream_endpoint(
request: Request,
kind: str | None = Query(None),
since: str | None = Query(None),
limit: int = Query(50, ge=1, le=500),
) -> StreamingResponse:
"""Stream Activity events as NDJSON: snapshot first, then live events.
Subscribes to ``activity_recorded``. The snapshot half reuses the
same in-memory filter as ``list_activity`` so the two endpoints are
interchangeable for the snapshot half.
"""
bus: EventBus = request.app.state.event_bus
async def gen() -> AsyncIterator[bytes]:
events = store.recent_activity(limit=limit)
if kind is not None:
events = [e for e in events if e["kind"] == kind]
if since is not None:
events = [e for e in events if e["timestamp"] >= since]
for ev in events:
yield ndjson_line({"type": "item", "data": ev})
yield ndjson_line({
"type": "snapshot_end", "data": {"count": len(events)},
})
async for chunk in tail_events(request, bus, ["activity_recorded"]):
yield chunk
return StreamingResponse(gen(), media_type="application/x-ndjson")
+332
View File
@@ -0,0 +1,332 @@
"""``/api/admin/*`` — operator-facing admin surface (SP11 + SP15).
Four endpoints:
* ``GET /api/admin/audit-log`` — list audit-log rows with optional
``entity_type`` / ``entity_id`` / ``event_type`` filters (SP11).
* ``GET /api/admin/audit-log/verify`` — walk the audit-log hash chain
and verify every row (SP11).
* ``POST /api/admin/db/rotate-key`` — SP15 SQLCipher key rotation.
Re-encrypts the DB in place with a fresh key, updates the Keychain.
Protected by a module-level ``threading.Lock`` so two concurrent
rotations can't race (a single process restart resets the lock —
intentional, the next start-up opens with whatever key is in the
Keychain).
* ``POST /api/admin/reload-config`` — re-read ``config/payers.yaml``
and revalidate; returns the number of configs loaded.
"""
from __future__ import annotations
import json
import logging
import threading
from typing import Any
from fastapi import APIRouter, HTTPException, Query
from cyclone import db, db_crypto, secrets
from cyclone.audit_log import AuditEvent, append_event, verify_chain
log = logging.getLogger(__name__)
router = APIRouter()
# Module-level lock: protects against concurrent key rotations within
# the same process. A process restart resets it (intentional — the
# operator's next start-up opens with whatever key is in the Keychain).
_db_rotate_lock = threading.Lock()
@router.get("/api/admin/audit-log")
def list_audit_log_endpoint(
entity_type: str | None = Query(default=None),
entity_id: str | None = Query(default=None),
event_type: str | None = Query(default=None),
limit: int = Query(default=100, ge=1, le=1000),
) -> Any:
"""List audit-log rows, newest first, with optional filters.
Filters match the (entity_type, entity_id) pair (typical use:
"show me everything that happened to claim C-123") or a single
event_type (typical use: "show me all clearhouse.submitted
events today").
"""
with db.SessionLocal()() as s:
q = s.query(db.AuditLog)
if entity_type:
q = q.filter(db.AuditLog.entity_type == entity_type)
if entity_id:
q = q.filter(db.AuditLog.entity_id == entity_id)
if event_type:
q = q.filter(db.AuditLog.event_type == event_type)
rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all()
return {
"total": len(rows),
"items": [
{
"id": r.id,
"event_type": r.event_type,
"entity_type": r.entity_type,
"entity_id": r.entity_id,
"actor": r.actor,
"payload": json.loads(r.payload_json) if r.payload_json else None,
"created_at": r.created_at.isoformat() if r.created_at else None,
"prev_hash": r.prev_hash,
"hash": r.hash,
}
for r in rows
],
}
@router.get("/api/admin/audit-log/verify")
def verify_audit_log_endpoint() -> Any:
"""Walk the audit-log chain and verify every row's hash.
Returns ``{"ok": true, "checked": N}`` for a clean chain, or
``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}``
for a broken chain. This is the operator's "did anyone tamper?"
endpoint; run it on demand or via a nightly cron job.
"""
with db.SessionLocal()() as s:
result = verify_chain(s)
return {
"ok": result.ok,
"checked": result.checked,
"first_bad_id": result.first_bad_id,
"reason": result.reason,
}
@router.post("/api/admin/db/rotate-key")
def rotate_db_key_endpoint(body: dict | None = None) -> Any:
"""Generate a fresh DB key, re-encrypt the DB, update the Keychain.
Request body (optional):
actor: who initiated the rotation. Defaults to "operator".
reason: human-readable reason. Written to the audit log.
Returns:
``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}``
on success. On failure (DB not encrypted, rekey failed,
Keychain update failed) returns the same shape with
``ok=false`` and a ``reason``. HTTP 503 is returned if the
rekey fails or encryption is not enabled.
The Keychain write happens *after* the rekey succeeds. If the
Keychain write fails, the DB has the new key but the Keychain
still has the old one — the endpoint returns 503 with a
"keychain update failed" reason and the operator must restore
the old key manually (``cyclone db restore-key <old_key>``) to
avoid being locked out.
"""
body = body or {}
actor = body.get("actor") or "operator"
reason = body.get("reason") or ""
if not db_crypto.is_encryption_enabled():
raise HTTPException(
status_code=400,
detail="encryption not enabled (sqlcipher3 missing or no Keychain key)",
)
# Acquire the lock; non-blocking so a stuck rotation doesn't
# silently hold up other requests.
if not _db_rotate_lock.acquire(blocking=False):
raise HTTPException(
status_code=409,
detail="another key rotation is in progress",
)
try:
url = db._resolve_url()
old_key = db_crypto.get_db_key()
if not old_key:
raise HTTPException(
status_code=400,
detail="no DB key in Keychain; cannot rotate",
)
new_key = db_crypto.generate_db_key()
result = db_crypto.rotate_db_key(
url=url, old_key=old_key, new_key=new_key,
)
if not result.ok:
# Rekey failed. The DB still has the old key. The
# Keychain is unchanged. Caller should NOT retry with
# the same new key (it's lost); generate a fresh one.
log.error("SQLCipher rotate failed: %s", result.reason)
raise HTTPException(
status_code=503,
detail={
"ok": False,
"old_fingerprint": result.old_fingerprint,
"new_fingerprint": result.new_fingerprint,
"rotated_at": result.rotated_at,
"reason": result.reason,
},
)
# Rekey succeeded. Now update the Keychain. If this fails
# the DB is locked behind the new key — operator must
# restore the old key manually.
if not secrets.set_secret(db_crypto.KEYCHAIN_ACCOUNT, new_key):
log.error("Keychain update failed after successful rekey!")
raise HTTPException(
status_code=503,
detail={
"ok": False,
"old_fingerprint": result.old_fingerprint,
"new_fingerprint": result.new_fingerprint,
"rotated_at": result.rotated_at,
"reason": (
"rekey succeeded but Keychain update failed — "
"the DB is now encrypted with the new key but "
"the Keychain still has the old one. "
"Restore the old key to the Keychain to recover."
),
},
)
# Store the old key in the "previous" account for a grace
# period so the operator can roll back if they discover the
# new key is broken (e.g. the Keychain entry got truncated).
secrets.set_secret(db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key)
# Rebuild the engine so subsequent connections use the new
# key. dispose_engine() closes every pooled connection that
# was using the old key; init_db() opens new ones with the
# new key from the (now-updated) Keychain.
db.reinit_engine()
# Audit log the rotation. We do this after the engine is
# rebuilt so the audit event is written with the new key —
# proving that the new key works for new writes.
try:
with db.SessionLocal()() as s:
append_event(s, AuditEvent(
event_type="db.key_rotated",
entity_type="database",
entity_id="cyclone.db",
actor=actor,
payload={
"old_fingerprint": result.old_fingerprint,
"new_fingerprint": result.new_fingerprint,
"table_count": result.table_count,
"reason": reason,
},
))
s.commit()
except Exception as exc: # noqa: BLE001
# Audit append is best-effort; rotation already succeeded.
log.warning("could not write audit event for rotation: %s", exc)
return {
"ok": True,
"old_fingerprint": result.old_fingerprint,
"new_fingerprint": result.new_fingerprint,
"rotated_at": result.rotated_at,
"table_count": result.table_count,
}
finally:
_db_rotate_lock.release()
@router.post("/api/admin/reload-config")
def reload_config():
"""Re-read ``config/payers.yaml`` and revalidate. Returns counts."""
from cyclone import payers as payer_loader
try:
configs = payer_loader.load_payer_configs()
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
return {"ok": True, "loaded": len(configs), "errors": []}
# ---------------------------------------------------------------------------
# SP16: inbound MFT scheduler
# ---------------------------------------------------------------------------
@router.post("/api/admin/scheduler/start")
async def scheduler_start():
"""Start the background MFT polling loop (idempotent)."""
from cyclone import scheduler as sched_mod
sched = sched_mod.get_scheduler()
await sched.start()
return {"ok": True, "status": sched.status().as_dict()}
@router.post("/api/admin/scheduler/stop")
async def scheduler_stop():
"""Stop the background MFT polling loop. Waits for the current tick."""
from cyclone import scheduler as sched_mod
sched = sched_mod.get_scheduler()
await sched.stop()
return {"ok": True, "status": sched.status().as_dict()}
@router.get("/api/admin/scheduler/status")
def scheduler_status():
"""Return the scheduler's running state + last tick summary."""
from cyclone import scheduler as sched_mod
sched = sched_mod.get_scheduler()
return sched.status().as_dict()
@router.post("/api/admin/scheduler/tick")
async def scheduler_tick():
"""Force a single poll cycle. Returns the TickResult.
Useful for the operator's "run now" button or for testing. If a
scheduled tick is already in flight, this call coalesces onto it
(waits for it to finish, returns its result).
"""
from cyclone import scheduler as sched_mod
sched = sched_mod.get_scheduler()
result = await sched.tick()
return {"ok": True, "tick": result.as_dict()}
@router.get("/api/admin/scheduler/processed-files")
def scheduler_processed_files(
limit: int = 50,
status: str | None = None,
):
"""List recently-processed inbound files (newest first).
The operator uses this to confirm the scheduler is keeping up
with MFT traffic. ``status`` filters to one of: ok, error,
skipped, pending.
"""
from sqlalchemy import desc
from cyclone import db
from cyclone.db import ProcessedInboundFile
with db.SessionLocal()() as session:
q = session.query(ProcessedInboundFile).order_by(
desc(ProcessedInboundFile.processed_at),
)
if status:
q = q.filter(ProcessedInboundFile.status == status)
q = q.limit(min(limit, 500))
rows = q.all()
return {
"count": len(rows),
"files": [
{
"id": r.id,
"name": r.name,
"size": r.size,
"file_type": r.file_type,
"processed_at": r.processed_at.isoformat() if r.processed_at else None,
"parser_used": r.parser_used,
"claim_count": r.claim_count,
"status": r.status,
"error_message": (
r.error_message[:200] + "..."
if r.error_message and len(r.error_message) > 200
else r.error_message
),
}
for r in rows
],
}
@@ -0,0 +1,75 @@
"""``/api/batches`` — list & detail endpoints for parsed batches.
Each ``store.add(...)`` call (837P, 835, 999, TA1, 277CA) produces a
``BatchRecord`` kept in memory by :class:`CycloneStore`. The list
endpoint returns a lightweight summary (id, kind, filename, parsedAt,
claim count) for the recent batches; the detail endpoint returns the
full parsed payload for one batch.
"""
from __future__ import annotations
import json
from typing import Any
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi.responses import StreamingResponse
from cyclone.api_helpers import ndjson_stream_list, wants_ndjson
from cyclone.store import BatchRecord, store
router = APIRouter()
def _batch_summary_claim_count(rec: BatchRecord) -> int:
"""Return the number of claims on a batch, handling both 837P and 835."""
if rec.kind == "837p":
return len(rec.result.claims) # type: ignore[attr-defined]
if rec.kind == "835":
return len(rec.result.claims) # type: ignore[attr-defined]
return 0
@router.get("/api/batches")
def list_batches_endpoint(
request: Request,
limit: int = Query(100, ge=1, le=1000),
) -> Any:
"""Summary of all parsed batches, newest first."""
records = store.list(limit=limit)
items = [
{
"id": r.id,
"kind": r.kind,
"inputFilename": r.input_filename,
"parsedAt": r.parsed_at.isoformat().replace("+00:00", "Z"),
"claimCount": _batch_summary_claim_count(r),
}
for r in records
]
all_records = store.all()
total = len(all_records)
returned = len(items)
has_more = total > returned
if wants_ndjson(request):
return StreamingResponse(
ndjson_stream_list(items, total, returned, has_more),
media_type="application/x-ndjson",
)
return {
"items": items,
"total": total,
"returned": returned,
"has_more": has_more,
}
@router.get("/api/batches/{batch_id}")
def get_batch_endpoint(batch_id: str) -> Any:
"""Return the full parsed payload for one batch."""
rec = store.get(batch_id)
if rec is None:
raise HTTPException(
status_code=404,
detail={"error": "Not found", "detail": f"Batch {batch_id} not found"},
)
return json.loads(rec.result.model_dump_json())
+411
View File
@@ -0,0 +1,411 @@
"""``/api/claims`` — list, live-tail, detail, X12 regenerate, line-reconciliation.
This is the largest single-resource router. Five endpoints:
* ``GET /api/claims`` — paginated list with filters
(status / batch_id / provider_npi / payer / date range / sort).
* ``GET /api/claims/stream`` — NDJSON snapshot + live tail via the
``claim_written`` EventBus kind.
* ``GET /api/claims/{claim_id}`` — drawer detail
(header / state / service lines / diagnoses / parties / validation /
raw segments / stateHistory / matchedRemittance).
* ``GET /api/claims/{claim_id}/serialize-837`` — regenerate X12 for
download (SP8).
* ``GET /api/claims/{claim_id}/line-reconciliation`` — per-line
reconciliation view for the ClaimDrawer tab (spec §5.1).
**Route ordering.** ``/stream`` is declared **before** ``/{claim_id}`` so
FastAPI's first-match routing doesn't swallow the literal ``stream``
segment as a claim id. Same order as the prior inline code.
The two projection helpers ``_claim_line_dict`` / ``_svc_to_dict`` are
private to this router; both only used by the line-reconciliation
endpoint.
"""
from __future__ import annotations
from decimal import Decimal
from typing import Any, AsyncIterator
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi.responses import JSONResponse, Response, StreamingResponse
from cyclone import db
from cyclone.api_helpers import (
ndjson_line,
ndjson_stream_list,
tail_events,
wants_ndjson,
)
from cyclone.db import Claim
from cyclone.parsers.models import ClaimOutput
from cyclone.parsers.serialize_837 import SerializeError as SerializeError837, serialize_837
from cyclone.pubsub import EventBus
from cyclone.store import store
router = APIRouter()
@router.get("/api/claims")
def list_claims_endpoint(
request: Request,
batch_id: str | None = Query(None),
status: str | None = Query(None),
provider_npi: str | None = Query(None),
payer: str | None = Query(None),
date_from: str | None = Query(None),
date_to: str | None = Query(None),
sort: str | None = Query(None),
order: str = Query("desc"),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
) -> Any:
"""Paginated claims list with filters; supports NDJSON streaming.
``has_more`` is computed as ``total > offset + returned`` so paginating
forward keeps returning true until the last page is reached.
"""
common = dict(
batch_id=batch_id,
status=status,
provider_npi=provider_npi,
payer=payer,
date_from=date_from,
date_to=date_to,
)
items = list(store.iter_claims(
sort=sort, order=order, limit=limit, offset=offset, **common,
))
total = len(list(store.iter_claims(**common)))
returned = len(items)
has_more = total > offset + returned
if wants_ndjson(request):
return StreamingResponse(
ndjson_stream_list(items, total, returned, has_more),
media_type="application/x-ndjson",
)
return {
"items": items,
"total": total,
"returned": returned,
"has_more": has_more,
}
@router.get("/api/claims/stream")
async def claims_stream_endpoint(
request: Request,
status: str | None = Query(None),
provider_npi: str | None = Query(None),
payer: str | None = Query(None),
date_from: str | None = Query(None),
date_to: str | None = Query(None),
sort: str | None = Query(None),
order: str = Query("desc"),
limit: int = Query(100, ge=1, le=1000),
) -> StreamingResponse:
"""Stream Claims as NDJSON: snapshot first, then live events.
Wire format:
* ``{"type":"item","data":<claim>}`` per snapshot row, then per
new ``claim_written`` event
* ``{"type":"snapshot_end","data":{"count":N}}`` after the snapshot
* ``{"type":"heartbeat","data":{"ts":<iso>}}`` every
``CYCLONE_TAIL_HEARTBEAT_S`` seconds when idle
Query params mirror :func:`list_claims_endpoint` so a frontend can
swap a one-shot fetch for a tail with no URL surgery.
NOTE: registered before ``/api/claims/{claim_id}`` so the literal
``stream`` path segment doesn't get matched as a claim id.
"""
bus: EventBus = request.app.state.event_bus
async def gen() -> AsyncIterator[bytes]:
# 1. Snapshot (eager — iter_claims returns a list already).
rows = store.iter_claims(
status=status, provider_npi=provider_npi, payer=payer,
date_from=date_from, date_to=date_to,
sort=sort or "-submission_date", order=order, limit=limit,
)
for row in rows:
yield ndjson_line({"type": "item", "data": row})
yield ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}})
# 2. Subscribe + heartbeats.
async for chunk in tail_events(request, bus, ["claim_written"]):
yield chunk
return StreamingResponse(gen(), media_type="application/x-ndjson")
@router.get("/api/claims/{claim_id}")
def get_claim_detail_endpoint(claim_id: str) -> dict:
"""Return one claim with full drawer context (SP4).
Body shape is produced by :meth:`CycloneStore.get_claim_detail`:
header, state, service lines, diagnoses, parties, validation,
raw segments, ``stateHistory`` (most-recent-first, capped at 50),
and a populated ``matchedRemittance`` block when paired.
Returns 404 — never 500 — on a missing claim so the UI can
distinguish "doesn't exist" from a transient fetch error.
"""
body = store.get_claim_detail(claim_id)
if body is None:
raise HTTPException(
status_code=404,
detail={
"error": "Not found",
"detail": f"Claim {claim_id} not found",
},
)
return body
@router.get("/api/claims/{claim_id}/serialize-837")
def serialize_claim_as_837(claim_id: str):
"""Return the claim as a regenerated X12 837P file (SP8).
Loads the ClaimOutput from the persisted ``raw_json`` and runs the
outbound serializer. Returns 404 if the claim doesn't exist, 422 if
the stored payload has no parseable ClaimOutput (data integrity
issue, not a transient failure).
"""
with db.SessionLocal()() as s:
row = s.get(Claim, claim_id)
if row is None:
return JSONResponse(
{"error": "Not found", "detail": f"Claim {claim_id} not found"},
status_code=404,
)
if not row.raw_json:
return JSONResponse(
{
"error": "Unprocessable",
"detail": f"Claim {claim_id} has no raw_json; cannot serialize",
},
status_code=422,
)
try:
claim_obj = ClaimOutput.model_validate(row.raw_json)
except Exception as exc:
return JSONResponse(
{
"error": "Unprocessable",
"detail": f"Claim {claim_id} raw_json is malformed: {exc}",
},
status_code=422,
)
try:
text = serialize_837(claim_obj)
except SerializeError837 as exc:
return JSONResponse(
{"error": "Unprocessable", "detail": str(exc)},
status_code=422,
)
return Response(
content=text,
media_type="text/x12",
headers={
"Content-Disposition": f'attachment; filename="claim-{claim_id}.x12"'
},
)
@router.get("/api/claims/{claim_id}/line-reconciliation")
def get_claim_line_reconciliation(claim_id: str) -> dict:
"""Per-line reconciliation view for the ClaimDrawer tab.
Spec §5.1. Returns the 837 service lines and 835 SVC composites
side-by-side, with per-line CAS adjustments and a summary block.
Architecture note: 837 service lines live in ``Claim.raw_json``
(not a separate ORM table), so the 837-side rows are read from the
JSON blob; the 835-side rows come from ``ServiceLinePayment`` ORM.
``LineReconciliation.claim_service_line_number`` stores the 1-based
line number to join them.
"""
from sqlalchemy import select
from cyclone.db import (
CasAdjustment,
LineReconciliation,
ServiceLinePayment,
)
with db.SessionLocal()() as s:
claim = s.get(db.Claim, claim_id)
if claim is None:
raise HTTPException(
status_code=404,
detail={"error": "Not found", "detail": f"Claim {claim_id} not found"},
)
# 837 service lines: from raw_json.
raw = claim.raw_json or {}
claim_lines_raw = raw.get("service_lines") or []
# Normalize to dicts for the response.
claim_lines = [_claim_line_dict(d) for d in claim_lines_raw]
# 835 service payments: ORM rows from the matched remit.
remits = list(
s.execute(
select(db.Remittance).where(db.Remittance.claim_id == claim_id)
).scalars().all()
)
svc_payments: list[dict] = []
svc_ids: list[int] = []
if remits:
svc_rows = list(
s.execute(
select(ServiceLinePayment).where(
ServiceLinePayment.remittance_id.in_([r.id for r in remits])
).order_by(ServiceLinePayment.line_number)
).scalars().all()
)
for svc in svc_rows:
d = _svc_to_dict(svc)
svc_payments.append(d)
svc_ids.append(svc.id)
# LineReconciliation rows.
lrs = list(
s.execute(
select(LineReconciliation).where(LineReconciliation.claim_id == claim_id)
).scalars().all()
)
# Index by claim_service_line_number and service_line_payment_id.
lr_by_claim_num: dict[int, LineReconciliation] = {
lr.claim_service_line_number: lr for lr in lrs if lr.claim_service_line_number is not None
}
lr_by_svc: dict[int, LineReconciliation] = {
lr.service_line_payment_id: lr for lr in lrs if lr.service_line_payment_id is not None
}
# CAS rows grouped by svc id.
cas_by_svc: dict[int, list[CasAdjustment]] = {}
if svc_ids:
cas_rows = list(
s.execute(
select(CasAdjustment).where(CasAdjustment.service_line_payment_id.in_(svc_ids))
).scalars().all()
)
for c in cas_rows:
cas_by_svc.setdefault(c.service_line_payment_id, []).append(c)
# Build output lines array, preserving 837 order then 835-only.
svc_by_id: dict[int, dict] = {d["id"]: d for d in svc_payments}
lines_out: list[dict] = []
billed_total = Decimal("0")
paid_total = Decimal("0")
adjustment_total = Decimal("0")
matched_count = 0
used_svc_ids: set[int] = set()
for cl in claim_lines:
billed_total += Decimal(str(cl["charge"]))
lr = lr_by_claim_num.get(cl["line_number"])
if lr is None:
lines_out.append({
"claim_service_line": cl,
"service_line_payment": None,
"status": "unmatched_837_only",
"adjustments": [],
})
continue
svc_id = lr.service_line_payment_id
svc = svc_by_id.get(svc_id) if svc_id else None
if svc_id is not None:
used_svc_ids.add(svc_id)
cas_list = cas_by_svc.get(svc_id, []) if svc_id is not None else []
cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0"))
if svc:
paid_total += Decimal(str(svc["payment"]))
adjustment_total += cas_total
if lr.status == "matched":
matched_count += 1
lines_out.append({
"claim_service_line": cl,
"service_line_payment": svc,
"status": lr.status,
"adjustments": [
{"group_code": c.group_code, "reason_code": c.reason_code,
"amount": str(Decimal(str(c.amount)))}
for c in cas_list
],
})
# 835-only lines (no claim match).
for lr in lrs:
if lr.claim_service_line_number is not None:
continue
svc_id = lr.service_line_payment_id
if svc_id is None:
continue
if svc_id in used_svc_ids:
continue
svc = svc_by_id.get(svc_id)
cas_list = cas_by_svc.get(svc_id, [])
cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0"))
if svc:
paid_total += Decimal(str(svc["payment"]))
adjustment_total += cas_total
lines_out.append({
"claim_service_line": None,
"service_line_payment": svc,
"status": lr.status,
"adjustments": [
{"group_code": c.group_code, "reason_code": c.reason_code,
"amount": str(Decimal(str(c.amount)))}
for c in cas_list
],
})
return {
"claim_id": claim_id,
"summary": {
"billed_total": str(billed_total),
"paid_total": str(paid_total),
"adjustment_total": str(adjustment_total),
"matched_lines": matched_count,
"total_lines": len(claim_lines),
},
"lines": lines_out,
}
def _claim_line_dict(d: dict) -> dict:
"""Project an 837 service-line dict from ``Claim.raw_json`` to wire shape."""
proc = d.get("procedure") or {}
charge = d.get("charge")
units = d.get("units")
return {
"line_number": d.get("line_number"),
"procedure_qualifier": proc.get("qualifier", "HC"),
"procedure_code": proc.get("code", ""),
"modifiers": proc.get("modifiers") or [],
"charge": str(Decimal(str(charge))) if charge is not None else "0",
"units": str(Decimal(str(units))) if units is not None else None,
"unit_type": d.get("unit_type"),
"service_date": d.get("service_date"),
}
def _svc_to_dict(svc) -> dict:
"""Project an ORM ``ServiceLinePayment`` to wire shape."""
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,
}
@@ -0,0 +1,126 @@
"""``/api/clearhouse`` — SP9 single-payer SFTP submit surface.
Two endpoints:
- ``GET /api/clearhouse`` — return the singleton clearhouse config
(dzinesco's identity, SFTP block, filename block). 404 when the
config hasn't been seeded yet.
- ``POST /api/clearhouse/submit`` — submit a batch of claims to the
clearhouse (SFTP). SP9 ships a stub that copies to ``staging_dir``
instead of opening a real SFTP connection; SP13 wires the real
paramiko-backed client behind the same ``make_client()`` factory.
For each claim the endpoint serializes a fresh 837 via
:func:`_serialize_claim_for_submit`, builds an HCPF-compliant outbound
filename, and records an audit-log row on success. Failures are
captured per-claim so the operator sees exactly which claim_ids
didn't go through.
"""
from __future__ import annotations
import json
from fastapi import APIRouter, HTTPException
from cyclone import db
from cyclone.audit_log import AuditEvent, append_event
from cyclone.parsers.parse_837 import parse
from cyclone.parsers.serialize_837 import serialize_837
from cyclone.store import store
router = APIRouter()
@router.get("/api/clearhouse")
def get_clearhouse():
"""Return the singleton clearhouse config (dzinesco's identity, SFTP block, filename block)."""
ch = store.get_clearhouse()
if ch is None:
raise HTTPException(status_code=404, detail="clearhouse not seeded")
return json.loads(ch.model_dump_json())
@router.post("/api/clearhouse/submit")
def submit_to_clearhouse(body: dict):
"""Submit a batch of claims to the clearhouse (SFTP). SP9: stub.
Body: ``{"claim_ids": [...], "payer_id": "CO_TXIX"}``
Stub behavior: serializes each claim via the SP7 serializer, builds
an HCPF-compliant outbound filename, and copies the result to
``{staging_dir}/{outbound_path}/{filename}`` instead of opening a
real SFTP connection. Returns a receipt per claim.
"""
from cyclone.clearhouse import make_client
from cyclone.edi.filenames import build_outbound_filename
claim_ids = body.get("claim_ids", [])
payer_id = body.get("payer_id")
if not claim_ids:
raise HTTPException(status_code=400, detail="claim_ids required")
if not payer_id:
raise HTTPException(status_code=400, detail="payer_id required")
ch = store.get_clearhouse()
if ch is None:
raise HTTPException(status_code=500, detail="clearhouse not seeded")
client = make_client(ch.sftp_block)
results = []
for cid in claim_ids:
try:
x12_text = _serialize_claim_for_submit(cid)
except Exception as exc: # noqa: BLE001
results.append({"claim_id": cid, "ok": False, "error": str(exc)})
continue
filename = build_outbound_filename(ch.tpid, "837P")
remote = f"{ch.sftp_block.paths['outbound']}/{filename}"
staging_path = client.write_file(remote, x12_text.encode("utf-8"))
results.append({
"claim_id": cid,
"ok": True,
"filename": filename,
"staging_path": str(staging_path),
"remote_path": remote,
})
# SP11: audit trail for each successful clearhouse submission.
with db.SessionLocal()() as audit_s:
append_event(audit_s, AuditEvent(
event_type="clearhouse.submitted",
entity_type="claim",
entity_id=cid,
payload={
"filename": filename,
"remote_path": remote,
"tpid": ch.tpid,
"stub": ch.sftp_block.stub,
},
actor="clearhouse-submit",
))
audit_s.commit()
return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub}
def _serialize_claim_for_submit(claim_id: str) -> str:
"""Serialize a claim to X12 for SFTP submission."""
with db.SessionLocal()() as s:
row = s.get(db.Claim, claim_id)
if row is None:
raise ValueError(f"claim {claim_id!r} not found")
raw = row.raw_json or {}
return _serialize_claim_from_raw(row, raw)
def _serialize_claim_from_raw(claim_row, raw: dict) -> str:
"""Best-effort serializer that uses the stored raw_json to emit a fresh 837.
For SP9 this delegates to the existing serialize_837 helper if the
claim has a complete raw_segments array. Otherwise it raises.
"""
# SP9 stub: if the claim has a `raw_json` with `x12_text`, use that.
if isinstance(raw, dict) and raw.get("x12_text"):
result = parse(raw["x12_text"])
if result.claims:
return serialize_837(result.claims[0])
raise RuntimeError(
f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text"
)
+56
View File
@@ -0,0 +1,56 @@
"""``/api/config/*`` — configured-provider / configured-payer lookups.
The ``config`` prefix is the operator-facing read-only window into the
``config/payers.yaml`` static config and the seeded providers table.
Used by the frontend Settings page to show what's available.
The payers endpoint at ``/api/config/payers/{payer_id}/configs`` is
slightly different: it returns both the YAML-loaded config and any
DB-overridden config (so a runtime override wins over the static
fallback). 404 / 200 / array shape match the rest of the API.
"""
from __future__ import annotations
import json
from fastapi import APIRouter, HTTPException, Query
from cyclone import payers as payer_loader
from cyclone.store import store
router = APIRouter()
@router.get("/api/config/providers")
def list_configured_providers(is_active: bool | None = Query(default=True)):
"""List the configured provider rows (3 NPIs for SP9)."""
return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)]
@router.get("/api/config/providers/{npi}")
def get_configured_provider(npi: str):
p = store.get_provider(npi)
if p is None:
raise HTTPException(status_code=404, detail=f"provider {npi!r} not found")
return json.loads(p.model_dump_json())
@router.get("/api/config/payers")
def list_configured_payers(is_active: bool | None = Query(default=True)):
return [json.loads(p.model_dump_json()) for p in store.list_payers(is_active=is_active)]
@router.get("/api/config/payers/{payer_id}/configs")
def list_payer_configs(payer_id: str):
"""List all (transaction_type, config_json) blocks for a payer."""
configs = [
{"transaction_type": tx, "config_json": block, "source": "yaml"}
for (pid, tx), block in payer_loader.all_configs().items()
if pid == payer_id
]
# Also check the DB for runtime-overridden configs
for tx in ("837P", "835", "277CA", "999", "TA1"):
live = store.get_payer_config(payer_id, tx)
if live is not None:
configs.append({"transaction_type": tx, "config_json": live, "source": "db"})
return configs
@@ -0,0 +1,220 @@
"""``/api/eligibility/*`` — 270/271 eligibility surface (SP3 P4 T23T24).
API-only — no DB persistence per spec section 3.4. The operator builds
a 270 inquiry from a small JSON body, downloads the raw text to paste
into a payer portal, then re-pastes the 271 response and parses it
back into structured fields on demand.
Two endpoints:
* ``POST /api/eligibility/request`` — build a 270 inquiry from a JSON
body and return ``{raw_270_text, parsed}``.
* ``POST /api/eligibility/parse-271`` — accept a raw 271 file upload
and return the parsed ``coverage_benefits`` / ``subscriber`` /
``summary`` / envelope triples.
The ``_validate_eligibility_request`` helper is private to this router
— it builds the Pydantic ``ParseResult270`` and is only used by the
``/request`` endpoint.
"""
from __future__ import annotations
import json
import logging
from datetime import date as _date
from typing import Any
from fastapi import APIRouter, File, HTTPException, UploadFile
from fastapi.responses import JSONResponse
from cyclone.parsers.exceptions import CycloneParseError
from cyclone.parsers.models import BatchSummary, Envelope
from cyclone.parsers.models_270 import (
EligibilityBenefitInquiry,
InformationReceiver270,
InformationSource270,
ParseResult270,
Subscriber270,
)
from cyclone.parsers.parse_271 import parse as parse_271_text
from cyclone.parsers.serialize_270 import serialize_270
log = logging.getLogger(__name__)
router = APIRouter()
def _validate_eligibility_request(body: dict) -> tuple[ParseResult270, str]:
"""Build a :class:`ParseResult270` from a request body dict.
The body shape is the minimum surface needed to build a valid 270
inquiry (per spec section 3.4 — operator-driven, ephemeral):
::
{
"subscriber": {first_name, last_name, member_id, dob},
"provider": {npi, name},
"payer": {id, name},
"service_type_code": "1"
}
Returns ``(ParseResult270, service_type_code)``. Raises
:class:`HTTPException` (400) when the body is missing required
fields.
"""
subscriber_in = body.get("subscriber") or {}
provider_in = body.get("provider") or {}
payer_in = body.get("payer") or {}
service_type_code = (body.get("service_type_code") or "").strip()
# Required-field checks. We surface a single 400 with the first
# missing field name to match the rest of the API's error contract.
if not service_type_code:
raise HTTPException(
status_code=400,
detail={"error": "Bad request", "detail": "service_type_code is required"},
)
if not subscriber_in.get("member_id"):
raise HTTPException(
status_code=400,
detail={"error": "Bad request", "detail": "subscriber.member_id is required"},
)
if not provider_in.get("npi"):
raise HTTPException(
status_code=400,
detail={"error": "Bad request", "detail": "provider.npi is required"},
)
if not payer_in.get("name"):
raise HTTPException(
status_code=400,
detail={"error": "Bad request", "detail": "payer.name is required"},
)
subscriber_dob_raw = subscriber_in.get("dob")
subscriber_dob: _date | None = None
if subscriber_dob_raw:
try:
subscriber_dob = _date.fromisoformat(subscriber_dob_raw)
except (TypeError, ValueError) as exc:
raise HTTPException(
status_code=400,
detail={
"error": "Bad request",
"detail": f"subscriber.dob must be YYYY-MM-DD: {exc}",
},
) from exc
result = ParseResult270(
envelope=Envelope(
sender_id="SUBMITTERID",
receiver_id=str(payer_in.get("id") or "RECEIVERID"),
control_number="000000001",
transaction_date=_date.today(),
implementation_guide="005010X279A1",
),
information_source=InformationSource270(
name=str(payer_in["name"]),
id=str(payer_in.get("id") or "") or None,
),
information_receiver=InformationReceiver270(
name=str(provider_in.get("name") or ""),
npi=str(provider_in["npi"]),
),
subscriber=Subscriber270(
member_id=str(subscriber_in["member_id"]),
first_name=str(subscriber_in.get("first_name") or "") or None,
last_name=str(subscriber_in.get("last_name") or "") or None,
dob=subscriber_dob,
),
inquiries=[EligibilityBenefitInquiry(service_type_code=service_type_code)],
summary=BatchSummary(
input_file="eligibility_request",
control_number="000000001",
transaction_date=_date.today(),
total_claims=1,
passed=1,
failed=0,
),
)
return result, service_type_code
@router.post("/api/eligibility/request")
def post_eligibility_request(body: dict) -> Any:
"""Build a 270 eligibility inquiry from a small JSON body.
Returns ``{"raw_270_text": <X12>, "parsed": <ParseResult270>}``
so the operator can either download the raw text (paste into a
payer portal) or render the parsed fields directly. Per spec
section 3.4, nothing is persisted to the DB.
"""
try:
result, _ = _validate_eligibility_request(body)
except HTTPException:
raise
except (KeyError, TypeError, ValueError) as exc:
raise HTTPException(
status_code=400,
detail={"error": "Bad request", "detail": f"Malformed body: {exc}"},
) from exc
raw_270_text = serialize_270(result)
return {
"raw_270_text": raw_270_text,
"parsed": json.loads(result.model_dump_json()),
}
@router.post("/api/eligibility/parse-271")
async def post_eligibility_parse_271(
file: UploadFile = File(...),
) -> Any:
"""Parse a 271 eligibility response and return the structured summary.
Accepts the raw 271 text as a file upload (multipart/form-data),
mirrors the ``/api/parse-999`` contract. Per spec section 3.4 the
result is NOT persisted — the operator re-pastes the 271 each
time they need a fresh read.
The response body is a JSON object with three top-level keys:
``coverage_benefits``, ``subscriber``, and ``summary``. 400 is
returned on empty / undecodable / malformed EDI; 200 on success.
"""
raw = await file.read()
if not raw:
return JSONResponse(
status_code=400,
content={"error": "Empty file", "detail": "Uploaded file contained no bytes."},
)
try:
text = raw.decode("utf-8")
except UnicodeDecodeError as exc:
return JSONResponse(
status_code=400,
content={"error": "Encoding error", "detail": str(exc)},
)
try:
result = parse_271_text(text, input_file=file.filename or "")
except CycloneParseError as exc:
return JSONResponse(
status_code=400,
content={"error": "Parse error", "detail": str(exc)},
)
except Exception as exc: # pragma: no cover - safety net
log.exception("Unexpected parser failure on 271")
return JSONResponse(
status_code=500,
content={"error": "Internal server error", "detail": str(exc)},
)
return {
"coverage_benefits": [
json.loads(cb.model_dump_json()) for cb in result.coverage_benefits
],
"subscriber": json.loads(result.subscriber.model_dump_json()),
"summary": json.loads(result.summary.model_dump_json()),
"envelope": json.loads(result.envelope.model_dump_json()),
"information_source": json.loads(result.information_source.model_dump_json()),
"information_receiver": json.loads(result.information_receiver.model_dump_json()),
}
+309
View File
@@ -0,0 +1,309 @@
"""``/api/inbox/*`` — operator-facing inbox surface.
Six endpoints powering the Inbox page (lanes, candidates, payer-rejected
acknowledge, rejected resubmit, CSV export). All endpoints run inside a
``db.SessionLocal()`` scope; the lanes / dismiss endpoints also read
``app.state.dismissed_pairs`` (a session-scoped set so dismissed items
don't re-appear until process restart).
Endpoints:
* ``GET /api/inbox/lanes`` — return all four lanes in one call
(rejected / payer_rejected / candidates / unmatched / done_today).
* ``POST /api/inbox/candidates/{remit_id}/match`` — manually link a
remit to a claim from the candidates lane.
* ``POST /api/inbox/candidates/dismiss`` — add candidate pairs to the
dismissed set so they stop appearing in the candidates lane.
* ``POST /api/inbox/payer-rejected/acknowledge`` — clear SP14
payer-rejected claims from the working surface without deleting the
rejection (idempotent: re-acknowledging is a noop).
* ``POST /api/inbox/rejected/resubmit`` — bulk move REJECTED claims
back to SUBMITTED; with ``?download=true`` returns a ZIP of
regenerated 837s with unique interchange control numbers.
* ``GET /api/inbox/export.csv?lane=…`` — stream a CSV for a single lane.
"""
from __future__ import annotations
import csv
import io
import json
import zipfile
from datetime import datetime, timezone
from fastapi import APIRouter, HTTPException, Query, Request, Response
from fastapi.responses import StreamingResponse
from cyclone import db
from cyclone.audit_log import AuditEvent, append_event
from cyclone.db import Claim, ClaimState, Remittance
from cyclone.parsers.models import ClaimOutput
from cyclone.parsers.serialize_837 import (
SerializeError as SerializeError837,
serialize_837_for_resubmit,
)
router = APIRouter()
@router.get("/api/inbox/lanes")
def inbox_lanes(request: Request) -> dict:
"""Return all Inbox lanes in one call.
Reads the session-scoped ``dismissed_pairs`` set so the
candidates lane excludes pairs the operator already dismissed in
this process lifetime.
"""
dismissed_pairs = getattr(request.app.state, "dismissed_pairs", set())
with db.SessionLocal()() as session:
from cyclone.inbox_lanes import compute_lanes
lanes = compute_lanes(session, dismissed_pairs=dismissed_pairs)
return {
"rejected": lanes.rejected,
# SP10: payer-rejected lane (277CA STC A4/A6/A7). Distinct from
# the 999 envelope rejection in ``rejected`` above.
"payer_rejected": lanes.payer_rejected,
"candidates": lanes.candidates,
"unmatched": lanes.unmatched,
"done_today": lanes.done_today,
}
@router.post("/api/inbox/candidates/{remit_id}/match")
def inbox_match_candidate(remit_id: str, body: dict):
"""Manually link a remit to a claim."""
claim_id = body.get("claim_id")
if not claim_id:
raise HTTPException(400, "claim_id required")
with db.SessionLocal()() as s:
claim = s.get(Claim, claim_id)
remit = s.get(Remittance, remit_id)
if claim is None or remit is None:
raise HTTPException(404, "claim or remit not found")
if claim.matched_remittance_id and claim.matched_remittance_id != remit_id:
raise HTTPException(
409,
detail={
"error": "claim_already_matched",
"current_state": (
claim.state.value if hasattr(claim.state, "value")
else str(claim.state)
),
"matched_remittance_id": claim.matched_remittance_id,
},
)
claim.matched_remittance_id = remit_id
remit.claim_id = claim_id
s.commit()
return {"ok": True, "claim_id": claim_id, "remit_id": remit_id}
@router.post("/api/inbox/candidates/dismiss")
def inbox_dismiss_candidates(request: Request, body: dict):
"""Add candidate pairs to the session-scoped dismissed set."""
pairs = body.get("pairs") or []
if not hasattr(request.app.state, "dismissed_pairs"):
request.app.state.dismissed_pairs = set()
for p in pairs:
cid = p.get("claim_id")
rid = p.get("remit_id")
if cid and rid:
request.app.state.dismissed_pairs.add(frozenset({cid, rid}))
return {"ok": True, "dismissed_count": len(pairs)}
@router.post("/api/inbox/payer-rejected/acknowledge")
def inbox_acknowledge_payer_rejected(body: dict):
"""Mark Payer-Rejected claims as acknowledged by the operator.
Operator hits "Acknowledge" on the Payer-Rejected Inbox lane to clear
the claim from the working surface. We don't delete the rejection
(the original payer_rejected_* fields stay for SP11 audit), we just
set ``payer_rejected_acknowledged_at`` so the lane query filters it
out.
Idempotent: re-acknowledging an already-acknowledged claim is a noop
(the timestamp is not bumped). Returns the count actually transitioned
so the UI can show "3 of 5 were already acknowledged".
"""
claim_ids = body.get("claim_ids") or []
actor = body.get("actor") or "operator"
if not isinstance(claim_ids, list) or not claim_ids:
raise HTTPException(400, "claim_ids must be a non-empty list")
if not all(isinstance(c, str) for c in claim_ids):
raise HTTPException(400, "claim_ids must be a list of strings")
with db.SessionLocal()() as session:
now = datetime.now(timezone.utc)
transitioned = 0
already_acked = 0
not_found = 0
not_rejected = 0
for cid in claim_ids:
claim = session.get(Claim, cid)
if claim is None:
not_found += 1
continue
if claim.payer_rejected_at is None:
not_rejected += 1
continue
if claim.payer_rejected_acknowledged_at is not None:
already_acked += 1
continue
claim.payer_rejected_acknowledged_at = now
claim.payer_rejected_acknowledged_actor = actor
transitioned += 1
# SP11: audit event for the acknowledge action. Best-effort:
# an audit-log failure must not block the operator's action.
try:
append_event(session, AuditEvent(
event_type="claim.payer_rejected_acknowledged",
entity_type="claim",
entity_id=claim.id,
actor=actor,
payload={
"payer_rejected_status_code": claim.payer_rejected_status_code,
"payer_rejected_by_277ca_id": claim.payer_rejected_by_277ca_id,
},
))
except Exception: # noqa: BLE001
pass
if transitioned:
session.commit()
return {
"ok": True,
"transitioned": transitioned,
"already_acked": already_acked,
"not_found": not_found,
"not_rejected": not_rejected,
}
@router.post("/api/inbox/rejected/resubmit")
def inbox_resubmit_rejected(
body: dict,
download: bool = Query(False, description="When true, return a ZIP of regenerated 837 files for the resubmitted claims (instead of JSON)."),
):
"""Bulk move REJECTED claims back to SUBMITTED.
With ``?download=true``, the response is a ``application/zip`` archive
containing one ``claim-{id}.x12`` per successfully resubmitted claim
(regenerated via ``serialize_837_for_resubmit`` so each file gets a
unique interchange/group control number). Conflicts are omitted from
the ZIP — they remain visible to the caller via the JSON shape of the
non-download path. Empty resubmit + download → 200 with an empty zip
so the UI can still hand the user a downloadable artifact.
"""
ids = body.get("claim_ids") or []
if not ids:
raise HTTPException(400, "claim_ids required")
accepted: list[str] = []
conflicts: list[dict] = []
# Track which claims are about to be resubmitted (and their index in
# the bundle) so the download path can serialize them with unique
# control numbers — back-to-back resubmits in the same file would
# otherwise all share ISA13/GS06 = "000000001".
accepted_with_rows: list[tuple[str, Claim]] = []
with db.SessionLocal()() as s:
for cid in ids:
c = s.get(Claim, cid)
if c is None:
continue
if c.state != ClaimState.REJECTED:
conflicts.append({
"claim_id": cid,
"current_state": (
c.state.value if hasattr(c.state, "value")
else str(c.state)
),
})
continue
c.state = ClaimState.SUBMITTED
c.state_changed_at = datetime.now(timezone.utc)
c.rejection_reason = None
c.rejected_at = None
c.resubmit_count = (c.resubmit_count or 0) + 1
accepted.append(cid)
accepted_with_rows.append((cid, c))
s.commit()
if not download:
return {"ok": True, "resubmitted": accepted, "conflicts": conflicts}
# Build a ZIP of regenerated 837s for the accepted claims. Conflicts
# and missing ids are deliberately excluded — the user already saw
# them in the JSON response on prior actions; the download is the
# "give me the files I asked for" payload.
buf = io.BytesIO()
serialize_errors: list[dict] = []
with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
for idx, (cid, c) in enumerate(accepted_with_rows, start=1):
if not c.raw_json:
serialize_errors.append({"claim_id": cid, "reason": "no raw_json"})
continue
try:
claim_obj = ClaimOutput.model_validate(c.raw_json)
except Exception as exc:
serialize_errors.append({"claim_id": cid, "reason": f"raw_json invalid: {exc}"})
continue
try:
text = serialize_837_for_resubmit(claim_obj, interchange_index=idx)
except SerializeError837 as exc:
serialize_errors.append({"claim_id": cid, "reason": str(exc)})
continue
zf.writestr(f"claim-{cid}.x12", text)
buf.seek(0)
headers = {
"Content-Disposition": (
f'attachment; filename="resubmit-{len(accepted)}-claims.zip"'
),
}
# Surface per-claim serialization failures as a custom response header
# so the UI can show "10 resubmitted, 2 couldn't be regenerated" without
# parsing the binary. The header value is JSON-encoded; the UI is
# expected to JSON.parse it after a fetch with response.ok.
if serialize_errors:
headers["X-Cyclone-Serialize-Errors"] = json.dumps(serialize_errors)
return Response(
content=buf.getvalue(),
media_type="application/zip",
headers=headers,
)
@router.get("/api/inbox/export.csv")
def inbox_export_csv(request: Request, lane: str):
"""Stream a CSV for a single lane."""
if lane not in {"rejected", "candidates", "unmatched", "done_today"}:
raise HTTPException(400, f"unknown lane: {lane}")
dismissed_pairs = getattr(request.app.state, "dismissed_pairs", set())
with db.SessionLocal()() as session:
from cyclone.inbox_lanes import compute_lanes
lanes = compute_lanes(session, dismissed_pairs=dismissed_pairs)
rows = getattr(lanes, lane)
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow([
"id", "kind", "patient_control_number", "charge_amount",
"payer_id", "provider_npi", "state", "rejection_reason",
"service_date", "score",
])
for r in rows:
writer.writerow([
r.get("id") or r.get("payer_claim_control_number"),
r.get("kind"),
r.get("patient_control_number"),
r.get("charge_amount"),
r.get("payer_id"),
r.get("provider_npi") or r.get("rendering_provider_npi"),
r.get("state"),
r.get("rejection_reason"),
r.get("service_date_from") or r.get("service_date"),
r.get("score"),
])
buf.seek(0)
return StreamingResponse(
iter([buf.getvalue()]),
media_type="text/csv",
headers={"Content-Disposition": f'attachment; filename="inbox-{lane}.csv"'},
)
@@ -0,0 +1,49 @@
"""``GET /api/providers`` — distinct providers observed across claims.
The list is built by ``store.distinct_providers()`` which scans claim
rows and returns one entry per unique ``billing_provider_npi``. The
``npi`` and ``state`` query params filter client-side. ``limit`` /
``offset`` paginate. Accepts ``application/x-ndjson`` like the other
list endpoints — see ``api_helpers.ndjson_stream_list``.
"""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Query, Request
from fastapi.responses import StreamingResponse
from cyclone.api_helpers import ndjson_stream_list, wants_ndjson
from cyclone.store import store
router = APIRouter()
@router.get("/api/providers")
def list_providers(
request: Request,
npi: str | None = Query(None),
state: str | None = Query(None),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
) -> Any:
items = store.distinct_providers()
if npi is not None:
items = [p for p in items if p["npi"] == npi]
if state is not None:
items = [p for p in items if p.get("state") == state]
paged = items[offset:offset + limit]
total = len(items)
returned = len(paged)
has_more = total > offset + returned
if wants_ndjson(request):
return StreamingResponse(
ndjson_stream_list(paged, total, returned, has_more),
media_type="application/x-ndjson",
)
return {
"items": paged,
"total": total,
"returned": returned,
"has_more": has_more,
}
@@ -0,0 +1,164 @@
"""``/api/reconciliation/*`` + ``/api/batch-diff`` — operator review surface.
Four endpoints powering the reconciliation page:
* ``GET /api/reconciliation/unmatched`` — return unmatched Claims
(left) and unmatched Remittances (right).
* ``POST /api/reconciliation/match`` — manually pair a Claim with a
Remittance (operator override).
* ``POST /api/reconciliation/unmatch`` — remove the current match and
reset the Claim to submitted.
* ``GET /api/batch-diff`` — side-by-side diff of two batches (SP3 P4 /
T18).
The two POST endpoints delegate to ``store.manual_match`` /
``store.manual_unmatch``; the router is responsible for translating the
store's exception hierarchy (``AlreadyMatchedError``, ``InvalidStateError``,
``NotMatchedError``, ``LookupError``) into the HTTP error contract.
"""
from __future__ import annotations
from fastapi import APIRouter, HTTPException, Query
from cyclone.store import (
AlreadyMatchedError,
InvalidStateError,
store,
)
router = APIRouter()
@router.get("/api/reconciliation/unmatched")
def get_reconciliation_unmatched() -> dict:
"""Return unmatched Claims (left) and unmatched Remittances (right).
Powers the reconciliation review surface: every Claim with no
paired Remittance appears on the left, every Remittance with no
paired Claim appears on the right. The two lists are always present
(empty list, never absent) so the UI can index unconditionally.
"""
return store.list_unmatched(kind="both")
@router.get("/api/batch-diff")
def get_batch_diff(
a: str | None = Query(None),
b: str | None = Query(None),
) -> dict:
"""Return a side-by-side diff of two batches identified by id.
Query params: ``a=<batch_id>``, ``b=<batch_id>`` (both required).
Response body (snake_case keys, see :mod:`cyclone.batch_diff` for the
projector shapes):
- ``a`` / ``b`` — small metadata blocks (id, kind, parsedAt,
inputFilename, claimCount)
- ``added`` — claims present in B but not A
- ``removed`` — claims present in A but not B
- ``changed`` — claims present in both, with field deltas
- ``summary`` — precomputed counts
Errors:
- 400 — missing ``a`` or ``b``
- 404 — either batch id is unknown
Pure read endpoint — never mutates the store. Both 837P and 835
batches are accepted (mixed-kind diffs are valid: comparing the
submitted claims against the matching remittances).
"""
if not a or not b:
raise HTTPException(
status_code=400,
detail={"error": "Missing param", "detail": "Both ?a=<batch_id> and ?b=<batch_id> are required."},
)
try:
a_rec, b_rec = store.load_two_for_diff(a, b)
except LookupError as exc:
raise HTTPException(
status_code=404,
detail={"error": "Not found", "detail": str(exc)},
)
# Lazy import — keeps the module's import surface small until the
# endpoint is actually hit.
from cyclone.batch_diff import diff_batches_to_wire
return diff_batches_to_wire(a_rec, b_rec)
@router.post("/api/reconciliation/match")
def post_reconciliation_match(body: dict) -> dict:
"""Manually pair a Claim with a Remittance (operator override).
Body: ``{"claim_id": ..., "remit_id": ...}``. Returns
``{"claim": <ui>, "match": <ui>}`` on success. Errors:
- 400: missing ``claim_id`` or ``remit_id``
- 404: claim or remittance not found
- 409: claim already matched, or apply_* returned a noop
(claim in terminal state) — detail echoes ``current_state``
and ``activity_kind`` so the UI can render a precise message.
"""
claim_id = body.get("claim_id")
remit_id = body.get("remit_id")
if not claim_id or not remit_id:
raise HTTPException(
status_code=400,
detail="claim_id and remit_id required",
)
try:
return store.manual_match(claim_id, remit_id)
except AlreadyMatchedError as e:
raise HTTPException(
status_code=409,
detail={"error": "already_matched", "message": str(e)},
)
except InvalidStateError as e:
raise HTTPException(
status_code=409,
detail={
"error": "invalid_state",
"current_state": e.current_state,
"activity_kind": e.activity_kind,
},
)
except LookupError:
# manual_match raises LookupError when the claim or remittance
# row is missing (we catch the parent class so any future
# KeyError subclasses in the store get the same treatment).
raise HTTPException(
status_code=404,
detail="claim_or_remit_not_found",
)
@router.post("/api/reconciliation/unmatch")
def post_reconciliation_unmatch(body: dict) -> dict:
"""Remove the current match for a Claim; reset Claim to submitted.
Body: ``{"claim_id": ...}``. Returns
``{"claim": <ui>, "deletedMatches": <count>}``. Errors:
- 400: missing ``claim_id``
- 404: claim not found
- 409: claim has no current match (NotMatchedError is mapped
by the store; we surface 409 to match the manual_match contract)
"""
from cyclone.store import NotMatchedError
claim_id = body.get("claim_id")
if not claim_id:
raise HTTPException(
status_code=400,
detail="claim_id required",
)
try:
return store.manual_unmatch(claim_id)
except NotMatchedError as e:
raise HTTPException(
status_code=409,
detail={"error": "not_matched", "message": str(e)},
)
except LookupError:
raise HTTPException(
status_code=404,
detail="claim_not_found",
)
@@ -0,0 +1,130 @@
"""``/api/remittances`` — list, live-tail, detail endpoints for 835 ERAs.
Mirror of the ``claims`` router for the 835 / Remittance resource:
* ``GET /api/remittances`` — paginated list with filters
(batch_id / payer / claim_id / date range / sort).
* ``GET /api/remittances/stream`` — NDJSON snapshot + live tail via the
``remittance_written`` EventBus kind.
* ``GET /api/remittances/{remittance_id}`` — detail with the labeled
CAS ``adjustments`` array.
**Route ordering.** ``/stream`` is declared **before** ``/{remittance_id}``
so FastAPI's first-match routing doesn't swallow the literal ``stream``
segment as a remittance id. Same order as the prior inline code.
"""
from __future__ import annotations
from typing import Any, AsyncIterator
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi.responses import StreamingResponse
from cyclone.api_helpers import (
ndjson_line,
ndjson_stream_list,
tail_events,
wants_ndjson,
)
from cyclone.pubsub import EventBus
from cyclone.store import store
router = APIRouter()
@router.get("/api/remittances")
def list_remittances_endpoint(
request: Request,
batch_id: str | None = Query(None),
payer: str | None = Query(None),
claim_id: str | None = Query(None),
date_from: str | None = Query(None),
date_to: str | None = Query(None),
sort: str | None = Query(None),
order: str = Query("desc"),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
) -> Any:
"""Paginated remittances list with filters; supports NDJSON streaming.
``has_more`` is computed as ``total > offset + returned`` so paginating
forward keeps returning true until the last page is reached.
"""
common = dict(
batch_id=batch_id,
payer=payer,
claim_id=claim_id,
date_from=date_from,
date_to=date_to,
)
items = list(store.iter_remittances(
sort=sort, order=order, limit=limit, offset=offset, **common,
))
total = len(list(store.iter_remittances(**common)))
returned = len(items)
has_more = total > offset + returned
if wants_ndjson(request):
return StreamingResponse(
ndjson_stream_list(items, total, returned, has_more),
media_type="application/x-ndjson",
)
return {
"items": items,
"total": total,
"returned": returned,
"has_more": has_more,
}
@router.get("/api/remittances/stream")
async def remittances_stream_endpoint(
request: Request,
payer: str | None = Query(None),
claim_id: str | None = Query(None),
date_from: str | None = Query(None),
date_to: str | None = Query(None),
sort: str | None = Query(None),
order: str = Query("desc"),
limit: int = Query(100, ge=1, le=1000),
) -> StreamingResponse:
"""Stream Remittances as NDJSON: snapshot first, then live events.
Subscribes to ``remittance_written``. Default sort is
``-received_date`` (newest-first), matching the list endpoint's
most common sort.
NOTE: registered before ``/api/remittances/{remittance_id}`` so
the literal ``stream`` path segment doesn't get matched as a
remittance id.
"""
bus: EventBus = request.app.state.event_bus
async def gen() -> AsyncIterator[bytes]:
rows = store.iter_remittances(
payer=payer, claim_id=claim_id,
date_from=date_from, date_to=date_to,
sort=sort or "-received_date", order=order, limit=limit,
)
for row in rows:
yield ndjson_line({"type": "item", "data": row})
yield ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}})
async for chunk in tail_events(request, bus, ["remittance_written"]):
yield chunk
return StreamingResponse(gen(), media_type="application/x-ndjson")
@router.get("/api/remittances/{remittance_id}")
def get_remittance_endpoint(remittance_id: str) -> dict:
"""Return one remittance with its labeled CAS ``adjustments`` array.
Returns 404 when the remittance is missing — never 500.
"""
body = store.get_remittance(remittance_id)
if body is None:
raise HTTPException(
status_code=404,
detail={"error": "Not found", "detail": f"Remittance {remittance_id} not found"},
)
return body