Nora
243 lines
7.5 KiB
Python
243 lines
7.5 KiB
Python
"""ACK persistence — 999 / TA1 / 277CA acknowledgements.
|
|
|
|
Each ACK type has its own ORM row (Ack / Ta1Ack / Two77caAck). The
|
|
write methods are simple inserts; the list/get methods are simple
|
|
queries. Fail-soft: persistence errors are logged but do not block
|
|
parsing.
|
|
|
|
SP25: every write path publishes one event on the EventBus so the
|
|
Acks page live-tail can subscribe — mirrors the existing
|
|
``claim_written`` / ``remittance_written`` / ``activity_recorded``
|
|
publish pattern (see ``cyclone.store.write``). Publish is best-effort:
|
|
a failing subscriber MUST NOT roll back the persisted row.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import date
|
|
from typing import TYPE_CHECKING
|
|
|
|
from cyclone import db
|
|
from cyclone.db import Ack
|
|
|
|
from . import utcnow
|
|
from .ui import to_ui_ack, to_ui_ta1_ack, to_ui_two77ca_ack
|
|
from .write import _sync_publish
|
|
|
|
if TYPE_CHECKING:
|
|
from cyclone.pubsub import EventBus
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _safe_publish(event_bus: "EventBus | None", kind: str, payload: dict) -> None:
|
|
"""Best-effort publish wrapped in try/except.
|
|
|
|
Mirrors ``CycloneStore._publish_events_sync`` — never raises, never
|
|
rolls back the persisted row. Falls back to skipping silently when
|
|
the bus doesn't implement the private ``_sync_publish`` interface
|
|
(e.g. test stubs).
|
|
"""
|
|
if event_bus is None:
|
|
return
|
|
try:
|
|
_sync_publish(event_bus, kind, payload)
|
|
except Exception: # noqa: BLE001
|
|
log.exception("acks store: %s publish failed", kind)
|
|
|
|
|
|
# -- 999 ACKs (SP3 P3 T13) -------------------------------------------
|
|
|
|
def add_999_ack(
|
|
*,
|
|
source_batch_id: str,
|
|
accepted_count: int,
|
|
rejected_count: int,
|
|
received_count: int,
|
|
ack_code: str,
|
|
raw_json: dict,
|
|
event_bus: "EventBus | None" = None,
|
|
) -> db.Ack:
|
|
"""Persist a 999 ACK row and return it.
|
|
|
|
``source_batch_id`` must reference an existing ``batches.id``.
|
|
For received 999s with no source batch the caller should pass a
|
|
synthetic id (e.g. ``"999-<interchange_control_number>"``) —
|
|
see ``/api/parse-999`` for that policy.
|
|
|
|
``raw_json`` is the full ``ParseResult999`` model dump; the
|
|
detail endpoint surfaces it without re-parsing the original
|
|
X12 text.
|
|
|
|
SP25: when ``event_bus`` is provided, publishes one
|
|
``ack_received`` event (with the full ``to_ui_ack`` row shape)
|
|
after the row commits so the Acks page live-tail sees new 999s
|
|
the moment they land.
|
|
"""
|
|
with db.SessionLocal()() as s:
|
|
row = Ack(
|
|
source_batch_id=source_batch_id,
|
|
accepted_count=accepted_count,
|
|
rejected_count=rejected_count,
|
|
received_count=received_count,
|
|
ack_code=ack_code,
|
|
parsed_at=utcnow(),
|
|
raw_json=raw_json,
|
|
)
|
|
s.add(row)
|
|
s.commit()
|
|
s.refresh(row)
|
|
row_id = row.id
|
|
|
|
if event_bus is not None:
|
|
with db.SessionLocal()() as s:
|
|
payload = to_ui_ack(s.get(Ack, row_id))
|
|
_safe_publish(event_bus, "ack_received", payload)
|
|
|
|
return row
|
|
|
|
def list_acks() -> list[db.Ack]:
|
|
"""Return every 999 ACK row, newest first (auto-increment id desc)."""
|
|
with db.SessionLocal()() as s:
|
|
return (
|
|
s.query(Ack)
|
|
.order_by(Ack.id.desc())
|
|
.all()
|
|
)
|
|
|
|
def get_ack( ack_id: int) -> db.Ack | None:
|
|
"""Return a single ACK row by id, or ``None`` if not found."""
|
|
with db.SessionLocal()() as s:
|
|
return s.get(Ack, ack_id)
|
|
|
|
# -- TA1 (Interchange Acknowledgment) -------------------------------
|
|
|
|
def add_ta1_ack(
|
|
*,
|
|
source_batch_id: str,
|
|
control_number: str,
|
|
interchange_date: date | None,
|
|
interchange_time: str | None,
|
|
ack_code: str,
|
|
note_code: str | None,
|
|
ack_generated_date: date | None,
|
|
sender_id: str,
|
|
receiver_id: str,
|
|
raw_json: dict,
|
|
event_bus: "EventBus | None" = None,
|
|
) -> db.Ta1Ack:
|
|
"""Persist a TA1 (Interchange Acknowledgment) row and return it.
|
|
|
|
Mirrors :meth:`add_999_ack` for the lower-level envelope ack. The
|
|
flat columns are promoted out of ``raw_json`` so the list
|
|
endpoint can sort/filter without a JSON parse; the full
|
|
``ParseResultTa1`` stays in ``raw_json`` for the detail endpoint.
|
|
|
|
SP25: when ``event_bus`` is provided, publishes one
|
|
``ta1_ack_received`` event after the row commits.
|
|
"""
|
|
with db.SessionLocal()() as s:
|
|
row = db.Ta1Ack(
|
|
source_batch_id=source_batch_id,
|
|
control_number=control_number,
|
|
interchange_date=interchange_date,
|
|
interchange_time=interchange_time,
|
|
ack_code=ack_code,
|
|
note_code=note_code,
|
|
ack_generated_date=ack_generated_date,
|
|
sender_id=sender_id,
|
|
receiver_id=receiver_id,
|
|
parsed_at=utcnow(),
|
|
raw_json=raw_json,
|
|
)
|
|
s.add(row)
|
|
s.commit()
|
|
s.refresh(row)
|
|
row_id = row.id
|
|
|
|
if event_bus is not None:
|
|
with db.SessionLocal()() as s:
|
|
payload = to_ui_ta1_ack(s.get(db.Ta1Ack, row_id))
|
|
_safe_publish(event_bus, "ta1_ack_received", payload)
|
|
|
|
return row
|
|
|
|
def list_ta1_acks() -> list[db.Ta1Ack]:
|
|
"""Return every TA1 ACK row, newest first (auto-increment id desc).
|
|
|
|
Mirrors :meth:`list_acks` — the API endpoint slices to its own
|
|
``limit`` so the ``total`` field reflects the full row count.
|
|
"""
|
|
with db.SessionLocal()() as s:
|
|
return (
|
|
s.query(db.Ta1Ack)
|
|
.order_by(db.Ta1Ack.id.desc())
|
|
.all()
|
|
)
|
|
|
|
def get_ta1_ack( ack_id: int) -> db.Ta1Ack | None:
|
|
"""Return a single TA1 ACK row by id, or ``None`` if not found."""
|
|
with db.SessionLocal()() as s:
|
|
return s.get(db.Ta1Ack, ack_id)
|
|
|
|
# -- 277CA (SP10) --------------------------------------------------
|
|
|
|
def add_277ca_ack(
|
|
*,
|
|
source_batch_id: str,
|
|
control_number: str,
|
|
accepted_count: int,
|
|
rejected_count: int,
|
|
paid_count: int,
|
|
pended_count: int,
|
|
raw_json: dict,
|
|
event_bus: "EventBus | None" = None,
|
|
) -> db.Two77caAck:
|
|
"""Persist a 277CA (Claim Acknowledgment) row and return it.
|
|
|
|
Mirrors :meth:`add_999_ack` but for the claim-level ack. The
|
|
per-claim status detail stays in ``raw_json``; only the four
|
|
counts are promoted so the list endpoint stays fast.
|
|
|
|
SP25: when ``event_bus`` is provided, publishes one
|
|
``two77ca_ack_received`` event after the row commits.
|
|
"""
|
|
with db.SessionLocal()() as s:
|
|
row = db.Two77caAck(
|
|
source_batch_id=source_batch_id,
|
|
control_number=control_number,
|
|
accepted_count=accepted_count,
|
|
rejected_count=rejected_count,
|
|
paid_count=paid_count,
|
|
pended_count=pended_count,
|
|
parsed_at=utcnow(),
|
|
raw_json=raw_json,
|
|
)
|
|
s.add(row)
|
|
s.commit()
|
|
s.refresh(row)
|
|
row_id = row.id
|
|
|
|
if event_bus is not None:
|
|
with db.SessionLocal()() as s:
|
|
payload = to_ui_two77ca_ack(s.get(db.Two77caAck, row_id))
|
|
_safe_publish(event_bus, "two77ca_ack_received", payload)
|
|
|
|
return row
|
|
|
|
def list_277ca_acks() -> list[db.Two77caAck]:
|
|
"""Return every 277CA ACK row, newest first (auto-increment id desc)."""
|
|
with db.SessionLocal()() as s:
|
|
return (
|
|
s.query(db.Two77caAck)
|
|
.order_by(db.Two77caAck.id.desc())
|
|
.all()
|
|
)
|
|
|
|
def get_277ca_ack( ack_id: int) -> db.Two77caAck | None:
|
|
"""Return a single 277CA ACK row by id, or ``None`` if not found."""
|
|
with db.SessionLocal()() as s:
|
|
return s.get(db.Two77caAck, ack_id)
|
|
|