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

feat(sp16): live MFT polling scheduler

Browse Source 
Adds an asyncio-based background scheduler that polls the Gainwell
MFT inbound path, downloads new files, and routes them through the
appropriate parser (999 / 835 / 277CA / TA1). Idempotent (re-ticks
and restarts skip already-processed files via the new
processed_inbound_files table). Crash-safe (per-file try/except so
one bad file doesn't stop the loop).

Lifespan auto-configures from the seeded dzinesco clearhouse's SFTP
block; auto-start is opt-in via CYCLONE_SCHEDULER_AUTOSTART.

Five admin endpoints added:
  GET  /api/admin/scheduler/status
  POST /api/admin/scheduler/start
  POST /api/admin/scheduler/stop
  POST /api/admin/scheduler/tick
  GET  /api/admin/scheduler/processed-files?status=&limit=

20 new tests (15 unit + 5 API).
This commit is contained in:
Tyler
2026-06-21 09:20:58 -06:00
parent 1267a341e3
commit 40f184c858
10 changed files with 1424 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/api.py
+140
View File
@@ -95,6 +95,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
serves requests. No teardown needed for Cyclone's local-only posture.
"""
from cyclone import db, payers as payer_loader
from cyclone import scheduler as scheduler_mod
db.init_db()
app.state.event_bus = EventBus()
@@ -104,8 +105,40 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
store.ensure_clearhouse_seeded()
except Exception as exc: # noqa: BLE001
log.exception("SP9 seed failed: %s", exc)
# SP16: configure the inbound MFT polling scheduler. The
# dzinesco clearhouse singleton (seeded by SP9) carries the SFTP
# block — multi-provider polling is out of scope for v1.
# Auto-start is opt-in via env var so operators running the API
# for one-off parsing don't accidentally poll the MFT on startup.
try:
clearhouse = store.get_clearhouse()
if clearhouse is not None:
sched = scheduler_mod.configure_scheduler(
clearhouse.sftp_block,
sftp_block_name=clearhouse.name or "default",
)
import os as _os
if _os.environ.get("CYCLONE_SCHEDULER_AUTOSTART", "").lower() in (
"1", "true", "yes",
):
await sched.start()
except Exception as exc: # noqa: BLE001
# Scheduler setup must never block the API from coming up.
log.exception("SP16 scheduler config failed: %s", exc)
yield
# SP16: stop the scheduler on shutdown so the test harness
# (TestClient context manager) doesn't leak background tasks.
try:
sched = scheduler_mod.get_scheduler()
await sched.stop()
except RuntimeError:
pass # never configured
except Exception as exc: # noqa: BLE001
log.warning("Scheduler shutdown failed: %s", exc)
# Mirror cli._PAYER_FACTORIES. Kept local so the API doesn't depend on the
# Click-based CLI module.
@@ -2430,6 +2463,113 @@ def rotate_db_key_endpoint(body: dict | None = None) -> Any:
_db_rotate_lock.release()
# ---------------------------------------------------------------------------
# SP16: live MFT polling scheduler (admin)
#
# The scheduler lives in :mod:`cyclone.scheduler` and is configured by
# the lifespan handler. The endpoints below expose start / stop /
# one-shot tick / status / history so an operator (or a cron job)
# can drive the scheduler without touching the DB.
#
# Note: the scheduler is OFF by default. Auto-start is opt-in via
# ``CYCLONE_SCHEDULER_AUTOSTART=true`` at launch. These endpoints
# are the operator's manual controls.
# ---------------------------------------------------------------------------
from cyclone import scheduler as _scheduler_mod
def _scheduler_or_503():
"""Return the configured scheduler or raise 503."""
try:
return _scheduler_mod.get_scheduler()
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
@app.post("/api/admin/scheduler/start")
async def scheduler_start() -> Any:
"""Begin polling the MFT inbound path every poll_interval_seconds."""
sched = _scheduler_or_503()
await sched.start()
return {"status": sched.status().as_dict()}
@app.post("/api/admin/scheduler/stop")
async def scheduler_stop() -> Any:
"""Stop polling. Waits up to 30s for the current tick to finish."""
sched = _scheduler_or_503()
await sched.stop()
return {"status": sched.status().as_dict()}
@app.post("/api/admin/scheduler/tick")
async def scheduler_tick() -> Any:
"""Run a single poll cycle synchronously and return the result.
Useful for: forcing a poll without waiting for the next interval;
verifying SFTP connectivity; running a one-shot import from the
CLI (``curl -X POST .../api/admin/scheduler/tick``).
"""
sched = _scheduler_or_503()
result = await sched.tick()
return {"ok": True, "tick": result.as_dict()}
@app.get("/api/admin/scheduler/status")
def scheduler_status() -> Any:
"""Return the scheduler's runtime snapshot (running, counters, last tick)."""
sched = _scheduler_or_503()
return sched.status().as_dict()
@app.get("/api/admin/scheduler/processed-files")
def scheduler_processed_files(
limit: int = Query(default=100, ge=1, le=1000),
status: str | None = Query(default=None),
) -> Any:
"""List rows from ``processed_inbound_files``, newest first.
The operator's "what did the scheduler do?" view. Filters by
``status`` (``ok`` / ``error`` / ``skipped`` / ``pending``).
Returns ``{"count": N, "files": [...]}`` where ``files[i]``
matches the ORM row as a JSON dict.
"""
from cyclone.db import ProcessedInboundFile
from cyclone.scheduler import STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING
valid_statuses = {STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING}
if status is not None and status not in valid_statuses:
raise HTTPException(
status_code=400,
detail=f"status must be one of {sorted(valid_statuses)}",
)
with db.SessionLocal()() as s:
q = s.query(db.ProcessedInboundFile)
if status is not None:
q = q.filter(db.ProcessedInboundFile.status == status)
rows = q.order_by(db.ProcessedInboundFile.id.desc()).limit(limit).all()
return {
"count": len(rows),
"files": [
{
"id": r.id,
"sftp_block_name": r.sftp_block_name,
"name": r.name,
"size": r.size,
"modified_at": r.modified_at.isoformat() if r.modified_at else None,
"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,
}
for r in rows
],
}
@app.get("/api/config/providers/{npi}")
def get_configured_provider(npi: str):
p = store.get_provider(npi)
backend/src/cyclone/clearhouse/__init__.py
+15 -4
View File
@@ -93,13 +93,24 @@ class SftpClient:
return self._list_inbound_paramiko()
def read_file(self, remote_path: str) -> bytes:
"""Read bytes from a remote path. Stub raises in stub mode."""
"""Read bytes from a remote path.
Stub mode: reads from ``{staging_dir}/{remote_path}``. Used by
the SP16 scheduler so it can exercise the same code path on a
workstation without a real MFT connection.
"""
if self._stub:
raise RuntimeError(
"Stub SFTP cannot read remote files. Use the local staging dir."
)
return self._read_file_stub(remote_path)
return self._read_file_paramiko(remote_path)
def _read_file_stub(self, remote_path: str) -> bytes:
"""Read bytes from ``{staging_dir}/{remote_path}`` (SP16 stub)."""
staging = Path(self._block.staging_dir).resolve()
target = staging / remote_path.lstrip("/")
if not target.is_file():
raise FileNotFoundError(f"inbound stub file not found: {target}")
return target.read_bytes()
def get_secret(self, name: str) -> Optional[str]:
"""Fetch the auth secret from Keychain. Returns the stub secret if absent."""
value = secrets.get_secret(name)
backend/src/cyclone/db.py
+45
View File
@@ -676,6 +676,51 @@ class AuditLog(Base):
)
# ---------------------------------------------------------------------------
# SP16: inbound MFT scheduler
# ---------------------------------------------------------------------------
class ProcessedInboundFile(Base):
"""One row per inbound MFT file the scheduler has downloaded.
SP16. Lets the scheduler be idempotent: a re-tick or restart must
not re-parse the same inbound file. The unique index on
(sftp_block_name, name) prevents duplicate inserts and lets the
scheduler fast-skip already-processed files via a SELECT.
Status values:
* ok - parsed cleanly, results persisted to the store
* error - parser raised; error_message captured
* skipped - file_type not in the scheduler's allowed set
* pending - file was downloaded but a downstream step failed;
the scheduler retries on the next tick
"""
__tablename__ = "processed_inbound_files"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
sftp_block_name: Mapped[str] = mapped_column(String(128), nullable=False)
name: Mapped[str] = mapped_column(String(256), nullable=False)
size: Mapped[int] = mapped_column(Integer, nullable=False)
modified_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
file_type: Mapped[Optional[str]] = mapped_column(String(16), nullable=True)
processed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
parser_used: Mapped[Optional[str]] = mapped_column(String(32), nullable=True)
claim_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
status: Mapped[str] = mapped_column(String(16), nullable=False)
error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
__table_args__ = (
Index(
"ux_processed_inbound_files_block_name",
"sftp_block_name", "name", unique=True,
),
Index("ix_processed_inbound_files_processed_at", "processed_at"),
Index("ix_processed_inbound_files_status", "status"),
)
# ---------------------------------------------------------------------------
# SP9: providers, payers, payer_configs, clearhouse
# ---------------------------------------------------------------------------
backend/src/cyclone/migrations/0011_processed_inbound_files.sql
+47
View File
@@ -0,0 +1,47 @@
-- version: 11
-- SP16: Inbound MFT polling scheduler
--
-- Tracks every file the background scheduler has downloaded from
-- the Gainwell MFT inbound path so a re-tick (or a restart) does not
-- re-process the same file. Idempotency is required for production:
-- the scheduler polls every N seconds and a slow MFT server may hand
-- us the same file across two polls.
--
-- We key on (sftp_block_name, name) — the sftp_block_name disambiguates
-- multi-provider installations (SP9+SP-multi-NPI), name is the inbound
-- filename as it appears on the MFT server.
--
-- Status values:
-- * ok — parsed cleanly, results persisted to the store
-- * error — parser raised; error_message captured for the operator
-- * skipped — file_type not in the scheduler's allowed set
-- * pending — file was downloaded but a downstream step failed
-- (e.g. DB write); the scheduler retries on the next tick
--
-- claim_count is the number of claims/remittances/acks the parser
-- surfaced. Surfaced on /api/admin/scheduler/status so the operator can
-- see throughput without parsing logs.
--
-- Compliance: not part of the HIPAA audit chain (SP11). This is
-- operational metadata; an SFTP outage shouldn't pollute the audit log.
CREATE TABLE processed_inbound_files (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sftp_block_name TEXT NOT NULL,
name TEXT NOT NULL,
size INTEGER NOT NULL,
modified_at TEXT,
file_type TEXT,
processed_at TEXT NOT NULL,
parser_used TEXT,
claim_count INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL,
error_message TEXT
);
CREATE UNIQUE INDEX ux_processed_inbound_files_block_name
ON processed_inbound_files(sftp_block_name, name);
CREATE INDEX ix_processed_inbound_files_processed_at
ON processed_inbound_files(processed_at DESC);
CREATE INDEX ix_processed_inbound_files_status
ON processed_inbound_files(status);
backend/src/cyclone/scheduler.py
+713
View File
@@ -0,0 +1,713 @@
"""Background inbound MFT polling scheduler (SP16).
Turns Cyclone from a manual upload tool into a live clearinghouse:
a long-running asyncio task that periodically polls the Gainwell MFT
inbound path, downloads each new file, and runs it through the
appropriate parser. The operator no longer has to watch for inbound
files and POST them to ``/api/parse-999`` etc. by hand.
Design constraints
------------------
* **Idempotent.** A re-tick (or a process restart) must not re-parse
the same inbound file. We persist a ``processed_inbound_files`` row
per file and skip ones we've already seen.
* **Crash-safe.** If the parser raises or the DB write fails, the
scheduler logs the error, records an ``error`` row, and moves on.
The next tick continues from the next file.
* **Bounded blast radius.** A bad file must not stop the scheduler.
Each file is wrapped in try/except so a 999 parser crash doesn't
prevent us from processing the next inbound 835.
* **Operator-controlled.** The scheduler is OFF by default; the
operator must explicitly start it (``POST /api/admin/scheduler/start``
or ``CYCLONE_SCHEDULER_AUTOSTART=true``). When it's running, status
is exposed via ``GET /api/admin/scheduler/status``.
* **No threading.** We use ``asyncio.create_task`` + ``asyncio.sleep``
rather than APScheduler or threading because the rest of the
codebase is asyncio-native (FastAPI). The whole polling loop runs
in the FastAPI event loop on the main thread.
Compliance: SP16 is operational metadata only. Inbound file
processing is NOT part of the HIPAA audit chain (SP11) — an SFTP
outage shouldn't pollute the audit log with parser errors.
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import traceback
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Optional
from sqlalchemy.exc import IntegrityError
from cyclone import db
from cyclone.store import store as cycl_store
from cyclone.audit_log import AuditEvent, append_event
from cyclone.clearhouse import InboundFile, SftpClient
from cyclone.db import ProcessedInboundFile
from cyclone.edi.filenames import parse_inbound_filename
from cyclone.inbox_state import apply_999_rejections
from cyclone.inbox_state_277ca import apply_277ca_rejections
from cyclone.providers import SftpBlock
log = logging.getLogger(__name__)
# Status values for ProcessedInboundFile.status.
STATUS_OK = "ok"
STATUS_ERROR = "error"
STATUS_SKIPPED = "skipped"
STATUS_PENDING = "pending"
# File types we know how to route. The HCPF set is broader (270/271/
# 276/277/278/820/834/ENCR) but Cyclone's parser only covers the
# four below. Files with unknown types are recorded as ``skipped``
# so the operator can see them in the audit table.
ROUTED_FILE_TYPES = frozenset({"999", "835", "277", "277CA", "TA1"})
@dataclass
class TickResult:
"""Outcome of a single scheduler tick (one poll cycle)."""
started_at: datetime
finished_at: Optional[datetime] = None
files_seen: int = 0
files_processed: int = 0
files_skipped: int = 0
files_errored: int = 0
errors: list[str] = field(default_factory=list)
def as_dict(self) -> dict[str, Any]:
return {
"started_at": self.started_at.isoformat(),
"finished_at": (
self.finished_at.isoformat() if self.finished_at else None
),
"files_seen": self.files_seen,
"files_processed": self.files_processed,
"files_skipped": self.files_skipped,
"files_errored": self.files_errored,
"errors": list(self.errors),
}
@dataclass
class SchedulerStatus:
"""Snapshot of the scheduler's runtime state."""
running: bool
poll_interval_seconds: int
sftp_block_name: str
last_poll_at: Optional[datetime]
poll_count: int
total_processed: int
total_skipped: int
total_errored: int
last_tick: Optional[TickResult] = None
def as_dict(self) -> dict[str, Any]:
return {
"running": self.running,
"poll_interval_seconds": self.poll_interval_seconds,
"sftp_block_name": self.sftp_block_name,
"last_poll_at": (
self.last_poll_at.isoformat() if self.last_poll_at else None
),
"poll_count": self.poll_count,
"total_processed": self.total_processed,
"total_skipped": self.total_skipped,
"total_errored": self.total_errored,
"last_tick": self.last_tick.as_dict() if self.last_tick else None,
}
# ---------------------------------------------------------------------------
# Per-file-type handlers. Each returns (parser_name, claim_count) and
# persists its own DB rows. The scheduler records the outcome.
# ---------------------------------------------------------------------------
def _handle_999(text: str, source_file: str) -> tuple[str, int]:
"""Parse a 999, apply rejections, persist ack row. Returns (parser, count)."""
from cyclone.parsers.parse_999 import parse_999_text
from cyclone.parsers.exceptions import CycloneParseError
try:
result = parse_999_text(text, input_file=source_file)
except CycloneParseError as exc:
raise ValueError(f"999 parse error: {exc}") from exc
received, accepted, rejected, ack_code = _ack_count_summary(result)
icn = result.envelope.control_number
synthetic_id = _ack_synthetic_source_batch_id(icn)
with db.SessionLocal()() as session:
def _lookup(pcn: str):
return (
session.query(db.Claim)
.filter_by(patient_control_number=pcn)
.first()
)
rejection_result = apply_999_rejections(
session, result, claim_lookup=_lookup,
)
if rejection_result.matched:
for cid in rejection_result.matched:
append_event(session, AuditEvent(
event_type="claim.rejected",
entity_type="claim",
entity_id=cid,
payload={"source_batch_id": synthetic_id},
actor="999-parser-scheduler",
))
row = cycl_store.add_ack(
source_batch_id=synthetic_id,
accepted_count=accepted,
rejected_count=rejected,
received_count=received,
ack_code=ack_code,
raw_json=json.loads(result.model_dump_json()),
)
session.commit()
return "parse_999", received
def _handle_835(text: str, source_file: str) -> tuple[str, int]:
"""Parse an 835, run validation, persist batch + remittances."""
import uuid
from cyclone.parsers.parse_835 import parse as parse_835
from cyclone.parsers.exceptions import CycloneParseError
from cyclone.parsers.validator_835 import validate as validate_835
from cyclone.payers import PAYER_FACTORIES_835
from cyclone.store import BatchRecord
config = PAYER_FACTORIES_835["co_medicaid_835"]()
try:
result = parse_835(text, config, input_file=source_file)
except CycloneParseError as exc:
raise ValueError(f"835 parse error: {exc}") from exc
# Validation report (mirrors the API endpoint).
report = validate_835(result, config)
n = len(result.claims)
if report.passed:
passed, failed, failed_claim_ids = n, 0, []
else:
passed, failed, failed_claim_ids = 0, n, [
c.payer_claim_control_number for c in result.claims
]
result = result.model_copy(update={
"validation": report,
"summary": result.summary.model_copy(update={
"passed": passed,
"failed": failed,
"failed_claim_ids": failed_claim_ids,
}),
})
rec = BatchRecord(
id=uuid.uuid4().hex,
kind="835",
input_filename=source_file,
parsed_at=datetime.now(timezone.utc),
result=result,
)
cycl_store.add(rec)
return "parse_835", len(result.claims)
def _handle_277ca(text: str, source_file: str) -> tuple[str, int]:
"""Parse a 277CA, persist ack + stamp payer-rejected claims."""
from cyclone.parsers.parse_277ca import parse_277ca_text
from cyclone.parsers.exceptions import CycloneParseError
try:
result = parse_277ca_text(text, input_file=source_file)
except CycloneParseError as exc:
raise ValueError(f"277CA parse error: {exc}") from exc
icn = result.envelope.control_number
synthetic_id = _277ca_synthetic_source_batch_id(icn)
accepted = sum(
1 for s in result.claim_statuses if s.classification == "accepted"
)
paid = sum(
1 for s in result.claim_statuses if s.classification == "paid"
)
rejected = sum(
1 for s in result.claim_statuses if s.classification == "rejected"
)
pended = sum(
1 for s in result.claim_statuses if s.classification == "pended"
)
with db.SessionLocal()() as session:
row = cycl_store.add_277ca_ack(
source_batch_id=synthetic_id,
control_number=icn,
accepted_count=accepted,
rejected_count=rejected,
paid_count=paid,
pended_count=pended,
raw_json=json.loads(result.model_dump_json()),
)
def _lookup(pcn: str):
return (
session.query(db.Claim)
.filter(db.Claim.patient_control_number == pcn)
.first()
)
apply_result = apply_277ca_rejections(
session, result, claim_lookup=_lookup, two77ca_id=row.id,
)
if apply_result.matched:
for cid in apply_result.matched:
append_event(session, AuditEvent(
event_type="claim.payer_rejected",
entity_type="claim",
entity_id=cid,
payload={"source_batch_id": synthetic_id, "277ca_id": row.id},
actor="277ca-parser-scheduler",
))
session.commit()
return "parse_277ca", len(result.claim_statuses)
def _handle_ta1(text: str, source_file: str) -> tuple[str, int]:
"""Parse a TA1, persist the interchange ack row."""
from cyclone.parsers.parse_ta1 import parse_ta1_text
from cyclone.parsers.exceptions import CycloneParseError
try:
result = parse_ta1_text(text, input_file=source_file)
except CycloneParseError as exc:
raise ValueError(f"TA1 parse error: {exc}") from exc
with db.SessionLocal()() as session:
cycl_store.add_ta1_ack(
source_batch_id=result.source_batch_id,
control_number=result.ta1.control_number,
interchange_date=result.ta1.interchange_date,
interchange_time=result.ta1.interchange_time,
ack_code=result.ta1.ack_code,
note_code=result.ta1.note_code,
ack_generated_date=result.ta1.ack_generated_date,
sender_id=result.envelope.sender_id,
receiver_id=result.envelope.receiver_id,
raw_json=json.loads(result.model_dump_json()),
)
session.commit()
return "parse_ta1", 1
# Map file_type → handler. Mirrors ROUTED_FILE_TYPES.
HANDLERS: dict[str, Callable[[str, str], tuple[str, int]]] = {
"999": _handle_999,
"835": _handle_835,
"277": _handle_277ca, # filename uses 277; parser is the same
"277CA": _handle_277ca,
"TA1": _handle_ta1,
}
# ---------------------------------------------------------------------------
# Light copies of helpers the API endpoints use, so the scheduler can
# run without depending on the FastAPI module.
# ---------------------------------------------------------------------------
def _ack_count_summary(result: Any) -> tuple[int, int, int, str]:
"""Return (received, accepted, rejected, ack_code) for a 999.
Mirrors the logic in ``cyclone.api._ack_count_summary`` but lives
here so the scheduler can run without importing the API module.
"""
if result.functional_group_acks:
fg = result.functional_group_acks[0]
return (
fg.received_count, fg.accepted_count,
fg.rejected_count, fg.ack_code,
)
sets = result.set_responses
received = len(sets)
accepted = sum(1 for s in sets if s.set_accept_reject.code == "A")
rejected = received - accepted
if rejected == 0:
code = "A"
elif accepted == 0:
code = "R"
else:
code = "P"
return (received, accepted, rejected, code)
def _ack_synthetic_source_batch_id(interchange_control_number: str) -> str:
"""Synthetic batches.id for a received 999 with no source batch."""
return f"999-{(interchange_control_number or '').strip() or '000000001'}"
def _277ca_synthetic_source_batch_id(interchange_control_number: str) -> str:
"""Synthetic batches.id for a received 277CA with no source batch."""
return f"277CA-{(interchange_control_number or '').strip() or '000000001'}"
# ---------------------------------------------------------------------------
# Scheduler
# ---------------------------------------------------------------------------
class Scheduler:
"""Background polling loop for inbound MFT files.
Lifecycle:
sched = Scheduler(sftp_block, poll_interval_seconds=60)
await sched.start() # begin polling
# ... later ...
await sched.stop() # finish current tick, then exit
status = sched.status() # snapshot
The scheduler is a single asyncio task. ``tick()`` does one full
poll cycle and is exposed for tests + the ``/api/admin/scheduler/tick``
endpoint so the operator can force a poll without waiting.
Threading: NOT thread-safe. All access (start/stop/tick/status)
must happen on the same event loop. The FastAPI app satisfies
this trivially because endpoints run on the loop.
"""
def __init__(
self,
sftp_block: SftpBlock,
*,
poll_interval_seconds: int = 60,
sftp_block_name: str = "default",
sftp_client_factory: Optional[Callable[[SftpBlock], Any]] = None,
) -> None:
self._sftp_block = sftp_block
self._poll_interval = poll_interval_seconds
self._sftp_block_name = sftp_block_name
# Factory indirection lets tests substitute a fake client
# without monkey-patching the module-level SftpClient.
self._sftp_client_factory = sftp_client_factory or SftpClient
self._task: Optional[asyncio.Task[None]] = None
self._stop_event = asyncio.Event()
self._last_poll_at: Optional[datetime] = None
self._poll_count = 0
self._total_processed = 0
self._total_skipped = 0
self._total_errored = 0
self._last_tick: Optional[TickResult] = None
# Coalesce overlapping ticks (a slow MFT server shouldn't let
# ticks stack up; the next tick fires only after the previous
# one finishes).
self._tick_in_progress = False
# ---- Public API -------------------------------------------------------
async def start(self) -> None:
"""Begin polling. Idempotent."""
if self._task is not None and not self._task.done():
log.info("Scheduler already running; start() is a no-op")
return
self._stop_event.clear()
self._task = asyncio.create_task(self._run(), name="mft-scheduler")
log.info(
"Scheduler started: poll every %ds, sftp_block=%r",
self._poll_interval, self._sftp_block_name,
)
async def stop(self) -> None:
"""Stop polling. Waits for the current tick to finish."""
if self._task is None or self._task.done():
return
self._stop_event.set()
try:
await asyncio.wait_for(self._task, timeout=30)
except asyncio.TimeoutError:
log.warning("Scheduler did not stop within 30s; cancelling")
self._task.cancel()
try:
await self._task
except (asyncio.CancelledError, Exception): # noqa: BLE001
pass
self._task = None
log.info("Scheduler stopped")
def status(self) -> SchedulerStatus:
"""Return a snapshot of the scheduler's state."""
return SchedulerStatus(
running=self.is_running(),
poll_interval_seconds=self._poll_interval,
sftp_block_name=self._sftp_block_name,
last_poll_at=self._last_poll_at,
poll_count=self._poll_count,
total_processed=self._total_processed,
total_skipped=self._total_skipped,
total_errored=self._total_errored,
last_tick=self._last_tick,
)
def is_running(self) -> bool:
return self._task is not None and not self._task.done()
async def tick(self) -> TickResult:
"""Run a single poll cycle and return the outcome.
Concurrent ticks are coalesced: if a tick is already in
progress, the second caller waits for it. This protects the
SFTP server from a stampede when the operator hits
``/api/admin/scheduler/tick`` while a scheduled tick is
already running.
"""
while self._tick_in_progress:
await asyncio.sleep(0.05)
self._tick_in_progress = True
try:
result = await self._tick_impl()
self._last_tick = result
self._last_poll_at = result.finished_at or result.started_at
self._poll_count += 1
self._total_processed += result.files_processed
self._total_skipped += result.files_skipped
self._total_errored += result.files_errored
return result
finally:
self._tick_in_progress = False
# ---- Internals --------------------------------------------------------
async def _run(self) -> None:
"""Main loop. Runs until ``stop()`` is called."""
# Stagger the first tick so we don't hammer the MFT server on
# startup if multiple operators restart Cyclone in lockstep.
await asyncio.sleep(1)
while not self._stop_event.is_set():
try:
await self.tick()
except Exception as exc: # noqa: BLE001
# tick() should never raise — it catches per-file
# exceptions. This is the safety net for SFTP outages
# or DB connectivity issues.
log.exception("Scheduler tick raised: %s", exc)
try:
await asyncio.wait_for(
self._stop_event.wait(),
timeout=self._poll_interval,
)
except asyncio.TimeoutError:
pass # poll interval elapsed; loop again
async def _tick_impl(self) -> TickResult:
"""One poll cycle: list → filter already-processed → route each."""
started = datetime.now(timezone.utc)
result = TickResult(started_at=started)
try:
files = await asyncio.to_thread(self._list_inbound)
except Exception as exc: # noqa: BLE001
log.exception("SFTP list_inbound failed")
result.errors.append(f"list_inbound: {exc}")
result.finished_at = datetime.now(timezone.utc)
return result
result.files_seen = len(files)
for f in files:
if self._stop_event.is_set():
break
await self._handle_one(f, result)
result.finished_at = datetime.now(timezone.utc)
return result
def _list_inbound(self) -> list[InboundFile]:
"""Return files in the inbound MFT path. Runs on a thread."""
client = self._sftp_client_factory(self._sftp_block)
return client.list_inbound()
async def _handle_one(self, f: InboundFile, result: TickResult) -> None:
"""Process one inbound file: skip-if-seen, classify, parse, record."""
if await self._already_processed(f.name):
return
try:
inbound = parse_inbound_filename(f.name)
file_type = inbound.file_type
except ValueError:
file_type = None
if file_type not in HANDLERS:
await self._record(
name=f.name, size=f.size, modified_at=f.modified_at,
file_type=file_type, parser_used=None, claim_count=0,
status=STATUS_SKIPPED,
error_message=(
f"file_type {file_type!r} not in {sorted(HANDLERS)}"
if file_type else "filename does not match HCPF inbound format"
),
)
result.files_skipped += 1
return
try:
_path, parser_used, claim_count = await asyncio.to_thread(
self._download_and_parse, f, file_type,
)
except Exception as exc: # noqa: BLE001
log.exception("Failed to process %s", f.name)
await self._record(
name=f.name, size=f.size, modified_at=f.modified_at,
file_type=file_type, parser_used=None, claim_count=0,
status=STATUS_ERROR,
error_message=(
f"{type(exc).__name__}: {exc}\n"
f"{traceback.format_exc()[-500:]}"
),
)
result.files_errored += 1
result.errors.append(f"{f.name}: {exc}")
return
await self._record(
name=f.name, size=f.size, modified_at=f.modified_at,
file_type=file_type, parser_used=parser_used, claim_count=claim_count,
status=STATUS_OK, error_message=None,
)
result.files_processed += 1
log.info(
"Processed inbound %s: parser=%s claims=%d",
f.name, parser_used, claim_count,
)
async def _already_processed(self, name: str) -> bool:
with db.SessionLocal()() as session:
row = (
session.query(ProcessedInboundFile)
.filter_by(sftp_block_name=self._sftp_block_name, name=name)
.filter(ProcessedInboundFile.status != STATUS_PENDING)
.first()
)
return row is not None
async def _record(
self,
*,
name: str,
size: int,
modified_at: datetime,
file_type: Optional[str],
parser_used: Optional[str],
claim_count: int,
status: str,
error_message: Optional[str],
) -> None:
"""Persist a processed_inbound_files row. Idempotent."""
with db.SessionLocal()() as session:
row = ProcessedInboundFile(
sftp_block_name=self._sftp_block_name,
name=name,
size=size,
modified_at=modified_at,
file_type=file_type,
processed_at=datetime.now(timezone.utc),
parser_used=parser_used,
claim_count=claim_count,
status=status,
error_message=error_message,
)
session.add(row)
try:
session.commit()
except IntegrityError:
# A concurrent scheduler (or a retry after a partial
# failure) already recorded this file. That's fine —
# the latest row wins; we just skip the dup.
session.rollback()
def _download_and_parse(
self, f: InboundFile, file_type: str,
) -> tuple[Path, str, int]:
"""Download from MFT, run the right handler. Returns (path, parser, count).
Stub mode: ``f.local_path`` already points at the staged file
(set by ``SftpClient._list_inbound_stub``). Real mode: the
remote name is ``f.name`` and we round-trip through paramiko.
"""
if self._sftp_block.stub:
# In stub mode the InboundFile already has a local_path;
# reading the staged bytes directly avoids the stub's
# remote-path semantics (which expect a full inbound path).
content = f.local_path.read_bytes()
else:
client = self._sftp_client_factory(self._sftp_block)
content = client.read_file(f.name)
text = content.decode("utf-8")
handler = HANDLERS[file_type]
parser_used, claim_count = handler(text, f.name)
return f.local_path, parser_used, claim_count
# ---------------------------------------------------------------------------
# Module-level singleton — only one scheduler per process.
# ---------------------------------------------------------------------------
_scheduler: Optional[Scheduler] = None
def configure_scheduler(
sftp_block: SftpBlock,
*,
poll_interval_seconds: int = 60,
sftp_block_name: str = "default",
force: bool = False,
) -> Scheduler:
"""Create the module-level scheduler singleton (or return the existing one).
Called from the FastAPI lifespan handler. Tests pre-configure the
scheduler before the TestClient opens the lifespan; in that case
we leave the existing singleton alone (``force=False``). Pass
``force=True`` to replace unconditionally.
"""
global _scheduler
if _scheduler is not None and not force:
return _scheduler
poll = int(
os.environ.get("CYCLONE_SCHEDULER_POLL_SECONDS", poll_interval_seconds),
)
_scheduler = Scheduler(
sftp_block,
poll_interval_seconds=poll,
sftp_block_name=sftp_block_name,
)
return _scheduler
def get_scheduler() -> Scheduler:
"""Return the module-level scheduler.
Raises:
RuntimeError: if ``configure_scheduler`` hasn't been called.
"""
if _scheduler is None:
raise RuntimeError(
"scheduler not configured; call configure_scheduler() first",
)
return _scheduler
def reset_scheduler_for_tests() -> None:
"""Clear the module-level scheduler. Test-only."""
global _scheduler
_scheduler = None