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

Add cyclone seed CLI subcommand for dev DB population

Browse Source 
Now that Dashboard reads live API hooks instead of the in-memory
sample store, a fresh dev DB renders as $0 KPIs and 'No activity
yet.' for a long time. `python -m cyclone seed` inserts a
deterministic batch of 96 claims + 28 activity events so the
Dashboard / Claims / Activity Log pages have something to render
without going through the Inbox / EDI parser pipeline.

  python -m cyclone seed           # insert (no-op if seeded)
  python -m cyclone seed --reset   # wipe + re-insert
  python -m cyclone seed --count 10
  python -m cyclone seed --status  # counts only

The seed mirrors the frontend `sampleData.ts` fixtures (3 TX
providers, 7 payers, 8 CPTs, 96 claims spread across the last 200
days) so dev dashboards look identical to the previous in-memory
fixture mode. `raw_json` and `payload_json` are populated in the
shape that `store.to_ui_claim_from_orm` / `recent_activity` parse
back into the UI wire format — wire parity, not shortcuts.

`--reset` cleans by id prefix (`SEED-` / `CLM-S`) rather than
relying on SQLite's `ON DELETE CASCADE`, which the dev session
doesn't enforce (no `PRAGMA foreign_keys = ON` at the SQLAlchemy
session layer). Safer than silent orphans.

Verified end-to-end: 986 backend tests pass, dashboard renders
$75,833 / 96 claims / 11.5% denial rate / 10 activity events /
top-3 providers ranked by claim count.
This commit is contained in:
Nora
2026-06-22 17:30:39 -06:00
parent 7e4bb4d2c8
commit 414d2eb722
2 changed files with 379 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/cli.py
+6
View File
@@ -79,6 +79,12 @@ def main(ctx: click.Context, log_format: str | None, log_file: Path | None) -> N
from cyclone.auth.cli import users_cli # noqa: E402
main.add_command(users_cli)
# Register the dev seed subcommand. Imported here for the same lazy-load
# reason as users_cli — keeps passlib/bcrypt + SQLAlchemy out of the
# parse-only path so ``python -m cyclone --help`` stays snappy.
from cyclone.seed_cli import seed_cli # noqa: E402
main.add_command(seed_cli)
@main.command("parse-837")
@click.argument("input_file", type=click.Path(exists=True, dir_okay=False, path_type=Path))
backend/src/cyclone/seed_cli.py
+373
View File
@@ -0,0 +1,373 @@
"""CLI subcommand: ``python -m cyclone seed``.
Populates the local DB with a deterministic batch of sample claims
plus matching activity events, so the Dashboard / Claims / Activity
Log pages have something to render in a fresh dev environment.
This is a dev-only convenience — the production DB never sees this
command (no ``[env: dev]`` gate, but it's never wired into any
container image). It writes rows with a recognizable batch id prefix
(``SEED-``) so ``--reset`` can clean them up without touching real
ingested data.
Why the data shape is hand-rolled here
--------------------------------------
The ``Claim`` ORM table stores its billing_provider / payer /
subscriber / service_lines payloads in a ``raw_json`` blob that the
read path (``store.to_ui_claim_from_orm``) parses back into the UI
shape. Mirroring the 837 parser's structure keeps the UI rendering
identical to a real ingestion — wire format parity, not shortcuts.
Activity rows are similar: ``payload_json`` carries the message
string the Dashboard's activity card shows.
Subcommands
-----------
``python -m cyclone seed`` — insert the default batch.
``python -m cyclone seed --count N`` — insert N claims (default 96).
``python -m cyclone seed --reset`` — wipe previously-seeded rows
first, then insert a fresh batch.
``python -m cyclone seed --status`` — print row counts without
inserting anything.
Exit codes
----------
* 0 — success (or already seeded and not asked to reset).
* 1 — DB error during insert.
* 2 — usage error (bad flag value).
The command is idempotent: re-running without ``--reset`` is a no-op
once a seed batch exists. This keeps ``cyclone``-driven boot scripts
safe to re-run.
"""
from __future__ import annotations
import json
import random
import sys
from datetime import datetime, timedelta, timezone
import click
from cyclone.db import ActivityEvent, Batch, Claim, ClaimState, SessionLocal
SEED_BATCH_PREFIX = "SEED-"
SEED_CLAIM_PREFIX = "CLM-S"
SEED_DEFAULT_COUNT = 96
SEED_ACTIVITY_LIMIT = 28
# Mirror src/data/sampleData.ts on the frontend so the Dashboard looks
# the same in dev mode as it did with the in-memory fixtures.
SAMPLE_PROVIDERS = [
{
"npi": "1730187395",
"name": "Cedar Park Family Medicine",
"tax_id": "47-3829104",
"address": "1401 Medical Pkwy",
"city": "Cedar Park",
"state": "TX",
"zip": "78613",
"phone": "(512) 555-0142",
},
{
"npi": "1528471902",
"name": "Lakeside Orthopedics",
"tax_id": "83-1172654",
"address": "900 W Lake Dr",
"city": "Austin",
"state": "TX",
"zip": "78746",
"phone": "(512) 555-0188",
},
{
"npi": "1982036471",
"name": "Hill Country Pediatrics",
"tax_id": "74-5520183",
"address": "205 State Hwy 27",
"city": "Marble Falls",
"state": "TX",
"zip": "78654",
"phone": "(830) 555-0117",
},
]
SAMPLE_PAYERS = [
"Blue Cross Blue Shield",
"United Healthcare",
"Aetna",
"Cigna",
"Humana",
"Medicare",
"Medicaid TX",
]
SAMPLE_CPTS = ["99213", "99214", "99203", "93000", "85025", "80053", "73721", "20610"]
SAMPLE_FIRST_NAMES = [
"Avery", "Jordan", "Riley", "Casey", "Morgan",
"Quinn", "Reese", "Sasha", "Drew", "Hayden",
]
SAMPLE_LAST_NAMES = [
"Nguyen", "Patel", "Garcia", "Cohen", "Okafor",
"Martinez", "Hwang", "Brooks", "Singh", "Tanaka",
]
# Frontend uses "accepted"/"pending" which the backend ClaimState
# enum doesn't carry. Map them to the closest real states so the
# Dashboard's filters (e.g. "status === 'submitted' || 'pending'")
# still find a match for in-flight work.
STATUS_WEIGHTS: list[tuple[ClaimState, int]] = [
(ClaimState.SUBMITTED, 1),
(ClaimState.RECEIVED, 1),
(ClaimState.DENIED, 1),
(ClaimState.PAID, 3),
(ClaimState.PAID, 3),
(ClaimState.PARTIAL, 1),
]
DENIAL_REASONS = [
"CO-97: Service included in another service",
"CO-16: Claim lacks information",
"CO-50: Non-covered service",
"PR-1: Deductible amount",
]
def _now_utc() -> datetime:
return datetime.now(timezone.utc).replace(tzinfo=None)
def _pick(rng: random.Random, items):
return items[rng.randint(0, len(items) - 1)]
def _build_seed_rows(count: int, *, seed: int = 42) -> tuple[Batch, list[Claim], list[ActivityEvent]]:
"""Build a deterministic Batch + N Claims + matching activity events.
The seed is fixed so re-running produces identical data — keeps
screenshots and dev environments stable.
"""
rng = random.Random(seed)
now = _now_utc()
parsed_at = now - timedelta(minutes=5)
batch = Batch(
id=f"{SEED_BATCH_PREFIX}{now.strftime('%Y%m%d-%H%M%S')}",
kind="837",
input_filename="seed/sample-837.edi",
parsed_at=parsed_at,
totals_json={"claim_count": count},
validation_json={"errors": [], "warnings": []},
raw_result_json={},
)
claims: list[Claim] = []
activity: list[ActivityEvent] = []
seq = 10428 # Match the frontend fixture's id range
for i in range(count):
provider = _pick(rng, SAMPLE_PROVIDERS)
payer = _pick(rng, SAMPLE_PAYERS)
cpt = _pick(rng, SAMPLE_CPTS)
first = _pick(rng, SAMPLE_FIRST_NAMES)
last = _pick(rng, SAMPLE_LAST_NAMES)
state, _ = _pick(rng, STATUS_WEIGHTS)
days_back = rng.randint(0, 200)
submitted = now - timedelta(days=days_back, hours=rng.randint(0, 23))
billed = 80 + rng.randint(0, 1400)
if state == ClaimState.PAID:
received = int(billed * (0.6 + rng.random() * 0.4))
elif state == ClaimState.PARTIAL:
received = int(billed * (0.2 + rng.random() * 0.3))
elif state == ClaimState.DENIED:
received = 0
else:
received = 0
claim_id = f"{SEED_CLAIM_PREFIX}{(seq + i):05d}"
denial_reason = _pick(rng, DENIAL_REASONS) if state == ClaimState.DENIED else None
raw_json = {
"billing_provider": {
"npi": provider["npi"],
"name": provider["name"],
"tax_id": provider["tax_id"],
"address": provider["address"],
"city": provider["city"],
"state": provider["state"],
"zip": provider["zip"],
"phone": provider["phone"],
},
"payer": {"name": payer},
"subscriber": {"first_name": first, "last_name": last},
"service_lines": [
{
"procedure": {"code": cpt},
"charge_amount": float(billed),
"service_date": submitted.date().isoformat(),
}
],
}
claims.append(Claim(
id=claim_id,
batch_id=batch.id,
patient_control_number=claim_id.replace(SEED_CLAIM_PREFIX, "PCN-"),
service_date_from=submitted.date(),
service_date_to=submitted.date(),
charge_amount=billed,
provider_npi=provider["npi"],
payer_id=None,
state=state,
state_changed_at=submitted,
rejection_reason=denial_reason,
resubmit_count=0,
raw_json=raw_json,
))
# First SEED_ACTIVITY_LIMIT claims get an activity event. Mirrors
# the frontend buildActivity() shape (claim_paid / claim_denied /
# claim_accepted / claim_submitted) but maps frontend statuses
# onto the backend's wire enum.
if i < SEED_ACTIVITY_LIMIT:
if state == ClaimState.PAID:
kind = "claim_paid"
verb = "Paid"
elif state == ClaimState.DENIED:
kind = "claim_denied"
verb = "Denied"
elif state in (ClaimState.RECEIVED, ClaimState.PARTIAL):
kind = "claim_accepted"
verb = "Accepted"
else:
kind = "claim_submitted"
verb = "Submitted"
payload = {
"message": f"{verb} {claim_id} · {first} {last}",
"npi": provider["npi"],
"amount": float(billed),
}
activity.append(ActivityEvent(
ts=submitted,
kind=kind,
batch_id=batch.id,
claim_id=claim_id,
remittance_id=None,
payload_json=payload,
))
return batch, claims, activity
def _existing_seed_batch_ids(s) -> list[str]:
rows = s.query(Batch.id).filter(Batch.id.like(f"{SEED_BATCH_PREFIX}%")).all()
return [r[0] for r in rows]
def _delete_seed_rows(s) -> int:
"""Delete every row that was inserted by the seed.
The seed writes rows whose ids begin with ``SEED-`` (batches) or
``CLM-S`` (claims). Activity events are joined to seeded batches
when the batch is still around, but we also clean up orphan
activity rows (no FK from ``activity_events.claim_id`` to
``claims.id``) by ``claim_id`` prefix. Returns the number of
batch rows deleted (the rest of the counts are printed via
``_print_status``).
"""
# Activity first — its rows reference both claims and batches, so
# delete the events tied to seed claims first, then any that were
# only tied to a now-orphaned seed batch.
activity = s.query(ActivityEvent).filter(
ActivityEvent.claim_id.like(f"{SEED_CLAIM_PREFIX}%")
).delete(synchronize_session=False)
activity2 = s.query(ActivityEvent).filter(
ActivityEvent.batch_id.like(f"{SEED_BATCH_PREFIX}%")
).delete(synchronize_session=False)
claims = s.query(Claim).filter(Claim.id.like(f"{SEED_CLAIM_PREFIX}%")).delete(synchronize_session=False)
batches = s.query(Batch).filter(Batch.id.like(f"{SEED_BATCH_PREFIX}%")).delete(synchronize_session=False)
s.commit()
click.echo(f" Removed {batches} seeded batch(es), {claims} claim(s), "
f"{activity + activity2} activity event(s).")
return batches
def _insert(batch: Batch, claims: list[Claim], activity: list[ActivityEvent]) -> None:
with SessionLocal()() as s:
s.add(batch)
s.add_all(claims)
s.add_all(activity)
s.commit()
def _print_status(s) -> None:
from sqlalchemy import func
seed_batches = s.query(func.count(Batch.id)).filter(Batch.id.like(f"{SEED_BATCH_PREFIX}%")).scalar() or 0
seed_claims = s.query(func.count(Claim.id)).filter(Claim.id.like(f"{SEED_CLAIM_PREFIX}%")).scalar() or 0
total_claims = s.query(func.count(Claim.id)).scalar() or 0
total_activity = s.query(func.count(ActivityEvent.id)).scalar() or 0
click.echo(f" Seed batches: {seed_batches}")
click.echo(f" Seed claims: {seed_claims}")
click.echo(f" Total claims: {total_claims}")
click.echo(f" Total activity: {total_activity}")
@click.command("seed")
@click.option(
"--count",
default=SEED_DEFAULT_COUNT,
show_default=True,
type=click.IntRange(min=1, max=10_000),
help="Number of claims to insert in the new batch.",
)
@click.option(
"--reset",
is_flag=True,
help="Delete any existing seeded batch (and its claims/activity) before inserting.",
)
@click.option(
"--status",
"show_status",
is_flag=True,
help="Print current seeded row counts and exit.",
)
def seed_cli(count: int, reset: bool, show_status: bool) -> None:
"""Populate the local DB with a deterministic batch of sample claims."""
with SessionLocal()() as s:
if show_status:
_print_status(s)
return
existing = _existing_seed_batch_ids(s)
if existing and not reset:
click.echo(
f"Seed already present (batch ids: {', '.join(existing)}). "
f"Re-run with --reset to replace, or --status to inspect.",
err=True,
)
sys.exit(0)
if reset and existing:
deleted = _delete_seed_rows(s)
click.echo(f" Removed {deleted} seeded batch(es).")
try:
batch, claims, activity = _build_seed_rows(count)
except Exception as exc:
click.echo(f"Failed to build seed rows: {exc}", err=True)
sys.exit(1)
try:
_insert(batch, claims, activity)
except Exception as exc:
click.echo(f"Failed to insert seed rows: {exc}", err=True)
sys.exit(1)
click.echo(f" Inserted batch {batch.id} with {len(claims)} claims and {len(activity)} activity events.")
_print_status(s)