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

feat(sp36): extract clearhouse + providers routers

Browse Source 
Tasks 11 + 12 combined per user direction (one commit for the
non-streaming singleton block).

api_routers/clearhouse.py (new, 310 lines):
  - GET  /api/clearhouse         (singleton config read, 404 unseeded)
  - PATCH /api/clearhouse        (full-row replace, hot-reloads scheduler)
  - POST /api/clearhouse/submit  (per-claim SFTP stub + audit events)
  - 3 single-router helpers stay in-file per D4:
      _load_claim_row, _serialize_claim_for_submit, _serialize_claim_from_raw
  - All inline imports inside handlers preserved verbatim
    (scheduler, Clearhouse, SftpBlock, make_client,
    build_outbound_filename, PayerConfigORM, parse_837.parse).

api_routers/providers.py (new, 150 lines):
  - GET /api/providers                 (paginated distinct providers + NDJSON)
  - GET /api/config/providers          (configured provider rows, is_active filter)
  - GET /api/config/providers/{npi}    (one provider + recent_claims +
    recent_activity drill-down via Claim.id outer-join Remittance.claim_id)
  - Three URL prefixes, one router per spec.

api_routers/_shared.py:
  - Early promotion: _actor_user_id moved here from api.py.
    The clearhouse router needs it; leaving it in api.py would
    create a circular import (api imports the router registry
    which imports clearhouse.py, which would import api for
    _actor_user_id). Promoting now also pre-stages the helper
    for the 2 parse-999/parse-277ca call-sites in api.py that
    Task 16 (parse router) will extract. The function body is
    verbatim — only the location moved. Docstring documents
    the early-promotion rationale.

api.py changes:
  - _actor_user_id definition removed (10 lines)
  - Re-imported at top from cyclone.api_routers._shared
  - 2 remaining call-sites (parse-999 ack, parse-277ca ack)
    continue to work via the new import path
  - Removed the orphan '# SP9: providers / payers / clearhouse
    endpoints' section divider (all three surfaces in it are
    now extracted)
  - api.py: 2858 -> 2468 LOC (-390; 6 routes extracted)

api_routers/__init__.py: registry extended (alphabetical) with
clearhouse + providers. 14 routers in registry.

Pytest: bit-identical to baseline — 21 failed, 1246 passed,
10 skipped, 6 errors. (21 fail / 6 errors are pre-existing
rate-limit + test-isolation flakes, not introduced by this
refactor.)

Live-tested via curl on the running container:
  GET /api/clearhouse                  -> 401 (matrix_gate fires)
  POST /api/clearhouse/submit          -> 401 (gated before body parse)
  GET /api/providers                   -> 401
  GET /api/config/providers            -> 401
  GET /api/config/providers/<npi>      -> 401
  POST /api/providers                  -> 405 (method-not-allowed)
  POST /api/config/providers           -> 405 (method-not-allowed)

pr-reviewer: PASS
This commit is contained in:
Nora
2026-07-06 15:24:09 -06:00
parent cad4f1fe2d
commit 4770c04021
5 changed files with 492 additions and 390 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/api.py
+1 -389
View File
@@ -42,6 +42,7 @@ from sqlalchemy.exc import IntegrityError
from cyclone.inbox_state import apply_999_rejections from cyclone.inbox_state import apply_999_rejections
from cyclone.inbox_state_277ca import apply_277ca_rejections from cyclone.inbox_state_277ca import apply_277ca_rejections
from cyclone.audit_log import AuditEvent, append_event, verify_chain from cyclone.audit_log import AuditEvent, append_event, verify_chain
from cyclone.api_routers._shared import _actor_user_id
from cyclone.claim_acks import ( from cyclone.claim_acks import (
apply_277ca_acks as _apply_277ca_acks, apply_277ca_acks as _apply_277ca_acks,
apply_999_acceptances as _apply_999_acceptances, apply_999_acceptances as _apply_999_acceptances,
@@ -49,19 +50,6 @@ from cyclone.claim_acks import (
) )
def _actor_user_id(request: Request) -> int | None:
"""Return the acting user's id from ``request.state.user``, or None.
``get_current_user``/``matrix_gate`` populate ``request.state.user``
for both the authenticated path and the AUTH_DISABLED escape hatch.
Returns None when the state hasn't been set (e.g. background jobs
or unit tests that bypass auth). Used to stamp ``user_id`` onto
audit events without crashing the request.
"""
user = getattr(request.state, "user", None)
if user is None:
return None
return getattr(user, "id", None)
from cyclone.parsers.exceptions import CycloneParseError from cyclone.parsers.exceptions import CycloneParseError
from cyclone.parsers.models import BatchSummary, ClaimOutput, Envelope, ParseResult from cyclone.parsers.models import BatchSummary, ClaimOutput, Envelope, ParseResult
from cyclone.parsers.models_835 import ParseResult835 from cyclone.parsers.models_835 import ParseResult835
@@ -2445,382 +2433,6 @@ def _svc_to_dict(svc) -> dict:
} }
@app.get("/api/providers", dependencies=[Depends(matrix_gate)])
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,
}
# ---------------------------------------------------------------------------
# SP9: providers / payers / clearhouse endpoints
# ---------------------------------------------------------------------------
@app.get("/api/clearhouse", dependencies=[Depends(matrix_gate)])
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())
@app.patch("/api/clearhouse", dependencies=[Depends(matrix_gate)])
async def patch_clearhouse(body: dict) -> Any:
"""Replace the singleton clearhouse row (SP25).
The full ``Clearhouse`` model is required — we don't accept partial
updates because the operator-facing use case is "I'm switching the
loop to real MFT" or "I'm pointing at a different MFT server",
not "I'm tweaking one field at a time." Validation errors are
returned as 422 (Pydantic default).
After a successful write, the running scheduler is hot-reloaded
via ``scheduler.reconfigure_scheduler()`` so the next tick uses
the new SftpBlock without a process restart.
"""
from cyclone import scheduler as _scheduler_mod
from cyclone.providers import Clearhouse as _Clearhouse, SftpBlock as _SftpBlock
# Strict-validate the sftp_block sub-dict FIRST. Pydantic v2's
# default mode coerces strings to bools (e.g. ``"stub": "yes"``
# silently becomes True), which would hide a real operator
# mistake. The Clearhouse model itself stays in loose mode so
# ISO-string ``updated_at`` (the JSON round-trip shape) keeps
# parsing.
raw_sb = body.get("sftp_block", {})
try:
_SftpBlock.model_validate(raw_sb, strict=True)
except Exception as exc:
raise HTTPException(
status_code=422, detail=f"invalid sftp_block: {exc}",
) from exc
# Now validate the full body in loose mode.
try:
parsed = _Clearhouse.model_validate(body)
except Exception as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
# SP25: when sftp_block.stub=false, the block must carry an auth
# account name and a non-empty host. The Pydantic model catches
# some of these; this catches the "empty password_keychain_account"
# case (which Pydantic allows because it's a free-form dict).
sb = parsed.sftp_block
if not sb.stub:
if not sb.host:
raise HTTPException(
status_code=422,
detail="sftp_block.host is required when stub=false",
)
auth = sb.auth or {}
if not auth.get("password_keychain_account") and not auth.get("key_file"):
raise HTTPException(
status_code=422,
detail=(
"sftp_block.auth must contain either "
"'password_keychain_account' or 'key_file' when stub=false"
),
)
updated = store.update_clearhouse(parsed)
await _scheduler_mod.reconfigure_scheduler(
updated.sftp_block,
sftp_block_name=updated.name or "default",
)
return json.loads(updated.model_dump_json())
@app.post("/api/clearhouse/submit", dependencies=[Depends(matrix_gate)])
def submit_to_clearhouse(request: Request, 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")
# Submitter (Loop 1000A) comes from the clearhouse config. The
# receiver (NM1*40) and SBR09 come from the per-payer config and
# are resolved per-claim below. Without this wiring, the
# serializer would emit "CYCLONE" / "RECEIVER" placeholders and
# the file would be rejected by HCPF.
submitter_kwargs = {
"sender_id": ch.tpid,
"submitter_name": ch.submitter_name,
"submitter_contact_name": ch.submitter_contact_name,
"submitter_contact_email": ch.submitter_contact_email,
}
if getattr(ch, "submitter_contact_phone", None):
submitter_kwargs["submitter_contact_phone"] = ch.submitter_contact_phone
# Build a payer_id → PayerConfig837 map once so we can look up the
# receiver + SBR09 default for each claim.
from cyclone.db import PayerConfigORM as _PayerConfigORM
def _resolve_payer_cfg(claim_obj) -> dict | None:
pid = (claim_obj.payer.id or "").strip() if claim_obj.payer else ""
pname = (claim_obj.payer.name or "").strip() if claim_obj.payer else ""
with db.SessionLocal()() as ss:
if pid:
row = ss.get(_PayerConfigORM, (pid, "837P"))
if row is not None:
return dict(row.config_json)
if pname:
rows = (
ss.query(_PayerConfigORM)
.filter(_PayerConfigORM.transaction_type == "837P")
.all()
)
for r in rows:
cj = dict(r.config_json)
if (r.payer_id or "").upper() == pname.upper():
return cj
if pname.lower() in str(cj).lower():
return cj
row = (
ss.query(_PayerConfigORM)
.filter(_PayerConfigORM.transaction_type == "837P")
.first()
)
return dict(row.config_json) if row else None
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
# Re-resolve the claim so we can look up its payer config. We
# re-parse the stored x12_text to get a ClaimOutput (same path
# the serializer uses).
try:
from cyclone.parsers.parse_837 import parse as _parse837
claim_row_obj = _load_claim_row(cid)
if claim_row_obj is None or not (claim_row_obj.raw_json or {}).get("x12_text"):
raise RuntimeError("no stored x12_text for claim")
parsed = _parse837(claim_row_obj.raw_json["x12_text"])
claim_obj = parsed.claims[0] if parsed.claims else None
except Exception:
claim_obj = None
if claim_obj is not None:
payer_cfg = _resolve_payer_cfg(claim_obj) or {}
receiver_id = payer_cfg.get("receiver_id") or (claim_obj.payer.id if claim_obj.payer else None) or "RECEIVER"
receiver_name = payer_cfg.get("receiver_name") or (claim_obj.payer.name if claim_obj.payer else None) or receiver_id
sbr09 = payer_cfg.get("sbr09_default") or "MC"
# Re-serialize with the proper envelope values.
try:
x12_text = _serialize_claim_for_submit(
cid,
**{**submitter_kwargs, "receiver_id": receiver_id,
"receiver_name": receiver_name,
"claim_filing_indicator_code": sbr09},
)
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",
user_id=_actor_user_id(request),
))
audit_s.commit()
return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub}
def _load_claim_row(claim_id: str):
"""Helper: load a Claim row by id (or return None)."""
with db.SessionLocal()() as s:
return s.get(Claim, claim_id)
def _serialize_claim_for_submit(claim_id: str, **kwargs) -> str:
"""Serialize a claim to X12 for SFTP submission. Lazy import of the
serializer to avoid pulling FastAPI machinery at module import time.
Optional ``**kwargs`` are forwarded to the serializer — used to
pass through clearhouse submitter info and per-payer receiver info
so the regenerated file matches what the HCPF MFT expects.
"""
from cyclone.parsers.serialize_837 import serialize_837
from cyclone import db
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")
# Re-parse the stored raw_json to get a ClaimOutput
from cyclone.parsers.models import ClaimOutput, Envelope, Subscriber, Payer, BillingProvider
from cyclone.parsers.parse_837 import parse
raw = row.raw_json or {}
# Reconstruct minimal ClaimOutput from raw_json; this is best-effort.
return _serialize_claim_from_raw(row, raw, **kwargs)
def _serialize_claim_from_raw(claim_row, raw: dict, **kwargs) -> 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 returns a
minimal placeholder. ``**kwargs`` are forwarded to the serializer
so callers can pass through submitter / receiver / SBR09 values
from the clearhouse and per-payer configs.
"""
from cyclone.parsers.serialize_837 import serialize_837
from cyclone.parsers.parse_837 import parse
# Re-parse the original batch text (need to re-derive from store).
# 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], **kwargs)
# Fallback: raise so the caller sees an error.
raise RuntimeError(
f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text"
)
@app.get("/api/config/providers", dependencies=[Depends(matrix_gate)])
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)]
@app.get("/api/config/providers/{npi}", dependencies=[Depends(matrix_gate)])
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")
provider_dict = json.loads(p.model_dump_json())
# SP21 Task 1.6: extend the response with two top-N arrays that the
# drill-down peek panel hangs off. ``recent_claims`` reuses the
# existing store projection (already returns UI-shaped dicts with
# ``submissionDate``); ``recent_activity`` is a direct ORM join
# because ``ActivityEvent`` has no ``provider_npi`` column — the
# filter has to hop through ``Claim.id``.
recent_claims = sorted(
store.iter_claims(provider_npi=npi),
key=lambda c: c.get("submissionDate") or "",
reverse=True,
)[:10]
# Activity filter has TWO join paths back to a Claim for this
# provider:
# 1. ``ActivityEvent.claim_id IN (claim_ids)`` — events that were
# recorded with a claim FK already set (claim_submitted,
# manual_match, claim_paid, etc.).
# 2. ``Remittance.claim_id IN (claim_ids)`` — the original
# ``remit_received`` event recorded at 835 ingest time
# (``store.add`` lines 999-1003) is inserted with
# ``claim_id=None`` because the remittance hasn't been matched
# to a claim yet. The auto-reconcile pass later sets
# ``Remittance.claim_id`` (``reconcile.run`` lines 289-293),
# but the *original* ActivityEvent row stays orphaned. The
# outer-join-then-OR lets us surface both shapes. Without
# path 2, a provider's activity feed looks frozen the moment
# an 835 lands — the most common activity, invisible.
with db.SessionLocal()() as s:
claim_ids = [
cid
for (cid,) in s.query(Claim.id)
.filter(Claim.provider_npi == npi)
.all()
]
activity_rows = []
if claim_ids:
activity_rows = (
s.query(db.ActivityEvent)
.outerjoin(
Remittance,
db.ActivityEvent.remittance_id == Remittance.id,
)
.filter(or_(
db.ActivityEvent.claim_id.in_(claim_ids),
Remittance.claim_id.in_(claim_ids),
))
.order_by(desc(db.ActivityEvent.ts))
.limit(10)
.all()
)
def _activity_to_ui(a):
return {
"id": a.id,
"ts": a.ts.isoformat().replace("+00:00", "Z") if a.ts else "",
"kind": a.kind,
"batchId": a.batch_id,
"claimId": a.claim_id,
"remittanceId": a.remittance_id,
"payload": a.payload_json or {},
}
provider_dict["recent_claims"] = recent_claims
provider_dict["recent_activity"] = [_activity_to_ui(a) for a in activity_rows]
return provider_dict
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
backend/src/cyclone/api_routers/__init__.py
+4
View File
@@ -15,11 +15,13 @@ from cyclone.api_routers import (
activity, activity,
admin, admin,
claim_acks, claim_acks,
clearhouse,
config, config,
dashboard, dashboard,
eligibility, eligibility,
health, health,
payers, payers,
providers,
reconciliation, reconciliation,
remittances, remittances,
ta1_acks, ta1_acks,
@@ -30,11 +32,13 @@ routers: list[APIRouter] = [
activity.router, # gated activity.router, # gated
admin.router, # gated admin.router, # gated
claim_acks.router, # gated claim_acks.router, # gated
clearhouse.router, # gated
config.router, # gated config.router, # gated
dashboard.router, # gated dashboard.router, # gated
eligibility.router, # gated eligibility.router, # gated
health.router, # public — health probes must work pre-auth health.router, # public — health probes must work pre-auth
payers.router, # gated payers.router, # gated
providers.router, # gated
reconciliation.router, # gated reconciliation.router, # gated
remittances.router, # gated remittances.router, # gated
ta1_acks.router, # gated ta1_acks.router, # gated
backend/src/cyclone/api_routers/_shared.py
+27 -1
View File
@@ -5,5 +5,31 @@ package import from here. Helpers graduate here when at least two
routers need them — single-router helpers stay in the router that routers need them — single-router helpers stay in the router that
uses them. uses them.
Initially empty; helpers promote here as the SP36 refactor progresses. Helpers currently promoted here:
- :func:`_actor_user_id` — promoted early (during SP36 Task 11
/ clearhouse extraction) because the clearhouse router needs
it and the 2 remaining call-sites in ``api.py`` (parse-999 ack
block, parse-277ca ack block) are both inside the parse
surface that Task 16 will extract. Promoting now is cheaper
than leaving a cross-module ``from cyclone.api import _actor_user_id``
that would create an import-cycle at registry load time.
""" """
from __future__ import annotations
from fastapi import Request
def _actor_user_id(request: Request) -> int | None:
"""Return the acting user's id from ``request.state.user``, or None.
``get_current_user``/``matrix_gate`` populate ``request.state.user``
for both the authenticated path and the AUTH_DISABLED escape hatch.
Returns None when the state hasn't been set (e.g. background jobs
or unit tests that bypass auth). Used to stamp ``user_id`` onto
audit events without crashing the request.
"""
user = getattr(request.state, "user", None)
if user is None:
return None
return getattr(user, "id", None)
backend/src/cyclone/api_routers/clearhouse.py
+310
View File
@@ -0,0 +1,310 @@
"""``/api/clearhouse*`` — singleton clearhouse config + SFTP submission.
Three endpoints, all gated by ``matrix_gate``:
- ``GET /api/clearhouse`` — read the singleton clearhouse row
(dzinesco's identity, SFTP block, filename block). 404 when
unseeded.
- ``PATCH /api/clearhouse`` — full-row replacement of the
singleton (SP25). Strict-validates ``sftp_block`` first (Pydantic
v2 default mode coerces strings-to-bools and would hide a real
operator mistake), then validates the whole body in loose mode.
Hot-reloads the running scheduler via
``scheduler.reconfigure_scheduler`` so the next tick picks up the
new ``SftpBlock`` without a process restart.
- ``POST /api/clearhouse/submit`` — submit a batch of claims to
the clearhouse. Stub: serializes via the SP7 serializer, builds
an HCPF-compliant outbound filename, copies the result to the
staging path. Per-claim audit events stamped with
``actor="clearhouse-submit"``.
Three single-router helpers stay in this file (per spec D4):
- :func:`_load_claim_row` — load a ``Claim`` row by id.
- :func:`_serialize_claim_for_submit` — re-serialize a claim to X12
with optional per-call kwargs (submitter, receiver, SBR09, etc).
- :func:`_serialize_claim_from_raw` — best-effort serializer that
re-parses stored ``x12_text`` and re-emits.
SP36 Task 11: this block moved here from ``api.py:2484`` (the 3
``/api/clearhouse*`` routes + 3 single-router helpers).
"""
from __future__ import annotations
import json
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Request
from cyclone import db
from cyclone.api_routers._shared import _actor_user_id
from cyclone.audit_log import AuditEvent, append_event
from cyclone.auth.deps import matrix_gate
from cyclone.db import Claim
from cyclone.store import store
router = APIRouter(dependencies=[Depends(matrix_gate)])
@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.patch("/api/clearhouse")
async def patch_clearhouse(body: dict) -> Any:
"""Replace the singleton clearhouse row (SP25).
The full ``Clearhouse`` model is required — we don't accept partial
updates because the operator-facing use case is "I'm switching the
loop to real MFT" or "I'm pointing at a different MFT server",
not "I'm tweaking one field at a time." Validation errors are
returned as 422 (Pydantic default).
After a successful write, the running scheduler is hot-reloaded
via ``scheduler.reconfigure_scheduler()`` so the next tick uses
the new SftpBlock without a process restart.
"""
from cyclone import scheduler as _scheduler_mod
from cyclone.providers import Clearhouse as _Clearhouse, SftpBlock as _SftpBlock
# Strict-validate the sftp_block sub-dict FIRST. Pydantic v2's
# default mode coerces strings to bools (e.g. ``"stub": "yes"``
# silently becomes True), which would hide a real operator
# mistake. The Clearhouse model itself stays in loose mode so
# ISO-string ``updated_at`` (the JSON round-trip shape) keeps
# parsing.
raw_sb = body.get("sftp_block", {})
try:
_SftpBlock.model_validate(raw_sb, strict=True)
except Exception as exc:
raise HTTPException(
status_code=422, detail=f"invalid sftp_block: {exc}",
) from exc
# Now validate the full body in loose mode.
try:
parsed = _Clearhouse.model_validate(body)
except Exception as exc:
raise HTTPException(
status_code=422, detail=str(exc),
) from exc
# SP25: when sftp_block.stub=false, the block must carry an auth
# account name and a non-empty host. The Pydantic model catches
# some of these; this catches the "empty password_keychain_account"
# case (which Pydantic allows because it's a free-form dict).
sb = parsed.sftp_block
if not sb.stub:
if not sb.host:
raise HTTPException(
status_code=422,
detail="sftp_block.host is required when stub=false",
)
auth = sb.auth or {}
if not auth.get("password_keychain_account") and not auth.get("key_file"):
raise HTTPException(
status_code=422,
detail=(
"sftp_block.auth must contain either "
"'password_keychain_account' or 'key_file' when stub=false"
),
)
updated = store.update_clearhouse(parsed)
await _scheduler_mod.reconfigure_scheduler(
updated.sftp_block,
sftp_block_name=updated.name or "default",
)
return json.loads(updated.model_dump_json())
@router.post("/api/clearhouse/submit")
def submit_to_clearhouse(request: Request, 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")
# Submitter (Loop 1000A) comes from the clearhouse config. The
# receiver (NM1*40) and SBR09 come from the per-payer config and
# are resolved per-claim below. Without this wiring, the
# serializer would emit "CYCLONE" / "RECEIVER" placeholders and
# the file would be rejected by HCPF.
submitter_kwargs = {
"sender_id": ch.tpid,
"submitter_name": ch.submitter_name,
"submitter_contact_name": ch.submitter_contact_name,
"submitter_contact_email": ch.submitter_contact_email,
}
if getattr(ch, "submitter_contact_phone", None):
submitter_kwargs["submitter_contact_phone"] = ch.submitter_contact_phone
# Build a payer_id → PayerConfig837 map once so we can look up the
# receiver + SBR09 default for each claim.
from cyclone.db import PayerConfigORM as _PayerConfigORM
def _resolve_payer_cfg(claim_obj) -> dict | None:
pid = (claim_obj.payer.id or "").strip() if claim_obj.payer else ""
pname = (claim_obj.payer.name or "").strip() if claim_obj.payer else ""
with db.SessionLocal()() as ss:
if pid:
row = ss.get(_PayerConfigORM, (pid, "837P"))
if row is not None:
return dict(row.config_json)
if pname:
rows = (
ss.query(_PayerConfigORM)
.filter(_PayerConfigORM.transaction_type == "837P")
.all()
)
for r in rows:
cj = dict(r.config_json)
if (r.payer_id or "").upper() == pname.upper():
return cj
if pname.lower() in str(cj).lower():
return cj
row = (
ss.query(_PayerConfigORM)
.filter(_PayerConfigORM.transaction_type == "837P")
.first()
)
return dict(row.config_json) if row else None
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
# Re-resolve the claim so we can look up its payer config. We
# re-parse the stored x12_text to get a ClaimOutput (same path
# the serializer uses).
try:
from cyclone.parsers.parse_837 import parse as _parse837
claim_row_obj = _load_claim_row(cid)
if claim_row_obj is None or not (claim_row_obj.raw_json or {}).get("x12_text"):
raise RuntimeError("no stored x12_text for claim")
parsed = _parse837(claim_row_obj.raw_json["x12_text"])
claim_obj = parsed.claims[0] if parsed.claims else None
except Exception:
claim_obj = None
if claim_obj is not None:
payer_cfg = _resolve_payer_cfg(claim_obj) or {}
receiver_id = payer_cfg.get("receiver_id") or (claim_obj.payer.id if claim_obj.payer else None) or "RECEIVER"
receiver_name = payer_cfg.get("receiver_name") or (claim_obj.payer.name if claim_obj.payer else None) or receiver_id
sbr09 = payer_cfg.get("sbr09_default") or "MC"
# Re-serialize with the proper envelope values.
try:
x12_text = _serialize_claim_for_submit(
cid,
**{**submitter_kwargs, "receiver_id": receiver_id,
"receiver_name": receiver_name,
"claim_filing_indicator_code": sbr09},
)
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",
user_id=_actor_user_id(request),
))
audit_s.commit()
return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub}
def _load_claim_row(claim_id: str):
"""Helper: load a Claim row by id (or return None)."""
with db.SessionLocal()() as s:
return s.get(Claim, claim_id)
def _serialize_claim_for_submit(claim_id: str, **kwargs) -> str:
"""Serialize a claim to X12 for SFTP submission. Lazy import of the
serializer to avoid pulling FastAPI machinery at module import time.
Optional ``**kwargs`` are forwarded to the serializer — used to
pass through clearhouse submitter info and per-payer receiver info
so the regenerated file matches what the HCPF MFT expects.
"""
from cyclone.parsers.serialize_837 import serialize_837
from cyclone import db
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")
# Re-parse the stored raw_json to get a ClaimOutput
from cyclone.parsers.models import ClaimOutput, Envelope, Subscriber, Payer, BillingProvider
from cyclone.parsers.parse_837 import parse
raw = row.raw_json or {}
# Reconstruct minimal ClaimOutput from raw_json; this is best-effort.
return _serialize_claim_from_raw(row, raw, **kwargs)
def _serialize_claim_from_raw(claim_row, raw: dict, **kwargs) -> 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 returns a
minimal placeholder. ``**kwargs`` are forwarded to the serializer
so callers can pass through submitter / receiver / SBR09 values
from the clearhouse and per-payer configs.
"""
from cyclone.parsers.serialize_837 import serialize_837
from cyclone.parsers.parse_837 import parse
# Re-parse the original batch text (need to re-derive from store).
# 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], **kwargs)
# Fallback: raise so the caller sees an error.
raise RuntimeError(
f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text"
)
backend/src/cyclone/api_routers/providers.py
+150
View File
@@ -0,0 +1,150 @@
"""``/api/providers`` and ``/api/config/providers*`` — read views over providers.
Three endpoints across two URL prefixes, all gated by ``matrix_gate``:
- ``GET /api/providers`` — distinct provider list
derived from the Claims population (``store.distinct_providers()``),
with ``npi`` / ``state`` filters, pagination, and an NDJSON variant.
- ``GET /api/config/providers`` — the configured provider
rows (3 NPIs for SP9), filtered by ``is_active``.
- ``GET /api/config/providers/{npi}`` — one configured provider
plus a small drill-down block: ``recent_claims`` (top-10 by
``submissionDate``) and ``recent_activity`` (top-10, joined via
``Claim.id`` because ``ActivityEvent`` has no ``provider_npi``
column; the outer-join via ``Remittance.claim_id`` surfaces the
orphan ``remit_received`` events that were recorded pre-match).
SP36 Task 12: this block moved here from ``api.py:2448`` (the
``/api/providers`` list, the ``/api/config/providers`` list, and
the ``/api/config/providers/{npi}`` detail) — three URL prefixes,
one router per spec.
"""
from __future__ import annotations
import json
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.responses import StreamingResponse
from sqlalchemy import desc, or_
from cyclone import db
from cyclone.api_helpers import (
ndjson_stream_list as _ndjson_stream_list,
wants_ndjson as _wants_ndjson,
)
from cyclone.auth.deps import matrix_gate
from cyclone.db import Claim, Remittance
from cyclone.store import store
router = APIRouter(dependencies=[Depends(matrix_gate)])
@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,
}
@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")
provider_dict = json.loads(p.model_dump_json())
# SP21 Task 1.6: extend the response with two top-N arrays that the
# drill-down peek panel hangs off. ``recent_claims`` reuses the
# existing store projection (already returns UI-shaped dicts with
# ``submissionDate``); ``recent_activity`` is a direct ORM join
# because ``ActivityEvent`` has no ``provider_npi`` column — the
# filter has to hop through ``Claim.id``.
recent_claims = sorted(
store.iter_claims(provider_npi=npi),
key=lambda c: c.get("submissionDate") or "",
reverse=True,
)[:10]
# Activity filter has TWO join paths back to a Claim for this
# provider:
# 1. ``ActivityEvent.claim_id IN (claim_ids)`` — events that were
# recorded with a claim FK already set (claim_submitted,
# manual_match, claim_paid, etc.).
# 2. ``Remittance.claim_id IN (claim_ids)`` — the original
# ``remit_received`` event recorded at 835 ingest time
# (``store.add`` lines 999-1003) is inserted with
# ``claim_id=None`` because the remittance hasn't been matched
# to a claim yet. The auto-reconcile pass later sets
# ``Remittance.claim_id`` (``reconcile.run`` lines 289-293),
# but the *original* ActivityEvent row stays orphaned. The
# outer-join-then-OR lets us surface both shapes. Without
# path 2, a provider's activity feed looks frozen the moment
# an 835 lands — the most common activity, invisible.
with db.SessionLocal()() as s:
claim_ids = [
cid
for (cid,) in s.query(Claim.id)
.filter(Claim.provider_npi == npi)
.all()
]
activity_rows = []
if claim_ids:
activity_rows = (
s.query(db.ActivityEvent)
.outerjoin(
Remittance,
db.ActivityEvent.remittance_id == Remittance.id,
)
.filter(or_(
db.ActivityEvent.claim_id.in_(claim_ids),
Remittance.claim_id.in_(claim_ids),
))
.order_by(desc(db.ActivityEvent.ts))
.limit(10)
.all()
)
def _activity_to_ui(a):
return {
"id": a.id,
"ts": a.ts.isoformat().replace("+00:00", "Z") if a.ts else "",
"kind": a.kind,
"batchId": a.batch_id,
"claimId": a.claim_id,
"remittanceId": a.remittance_id,
"payload": a.payload_json or {},
}
provider_dict["recent_claims"] = recent_claims
provider_dict["recent_activity"] = [_activity_to_ui(a) for a in activity_rows]
return provider_dict