sttil/Signal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(audit): add Supabase client support to audit logger, format cleanup, update cascade test

Browse source
This commit is contained in:
Kisa 2026-06-21 16:16:34 -04:00
parent b9939b1ba0
commit ae650e2597
3 changed files with 321 additions and 156 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

398
python-backend/api/main.py
View file

@ -26,9 +26,13 @@ from core.coverage_calculator import ShipmentRecord, calculate_batch
from core.audit_logger import AuditAction, log_event
from core.doc_state_machine import compute_doc_state, DocState
from core.persistence import (
persist_export, persist_upload,
upsert_confirmed_visit, load_confirmed_visits_for_org, get_or_create_org,
upsert_doc_status, load_doc_statuses_for_org,
persist_export,
persist_upload,
upsert_confirmed_visit,
load_confirmed_visits_for_org,
get_or_create_org,
upsert_doc_status,
load_doc_statuses_for_org,
)
from api.normalizer import normalize_csv
@ -70,19 +74,27 @@ _clerk_jwks_url = os.getenv("CLERK_JWKS_URL", "")
def _get_jwks_client() -> PyJWKClient:
if not _clerk_jwks_url:
raise RuntimeError("CLERK_JWKS_URL is not set")
return PyJWKClient(_clerk_jwks_url, cache_keys=True, cache_jwk_set=True, lifespan=300)
return PyJWKClient(
_clerk_jwks_url, cache_keys=True, cache_jwk_set=True, lifespan=300
)
def verify_clerk_token(authorization: str) -> dict:
"""Verify a Clerk Bearer JWT. Returns decoded claims or raises HTTP 401."""
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or malformed Authorization header")
raise HTTPException(
status_code=401, detail="Missing or malformed Authorization header"
)
token = authorization.removeprefix("Bearer ").strip()
try:
client = _get_jwks_client()
signing_key = client.get_signing_key_from_jwt(token)
return jwt.decode(token, signing_key.key, algorithms=["RS256"],
options={"verify_exp": True, "verify_nbf": True})
return jwt.decode(
token,
signing_key.key,
algorithms=["RS256"],
options={"verify_exp": True, "verify_nbf": True},
)
except ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token has expired")
except InvalidTokenError as exc:
@ -112,6 +124,7 @@ def require_auth(
return {"sub": "dev", "via": "open"}
raise HTTPException(status_code=401, detail="Authentication required")
DEVICE_DISPLAY = {
"dexcom_g7": "Dexcom G7",
"dexcom_g6": "Dexcom G6",
@ -121,26 +134,26 @@ DEVICE_DISPLAY = {
}
FLAG_LABELS = {
"SUPPLY_LAPSED": "Docs Required",
"VISIT_REQUIRED": "Escalate",
"TRANSFER_PENDING": "New Patient",
"RENEWAL_CRITICAL": "Escalate",
"RENEWAL_ELEVATED": "Confirm Appointment",
"RENEWAL_SOON": "Begin Outreach",
"RESUPPLY_READY": "Clear to Ship",
"ACTIVE": "On Track",
"SUPPLY_LAPSED": "Docs Required",
"VISIT_REQUIRED": "Escalate",
"TRANSFER_PENDING": "New Patient",
"RENEWAL_CRITICAL": "Escalate",
"RENEWAL_ELEVATED": "Confirm Appointment",
"RENEWAL_SOON": "Begin Outreach",
"RESUPPLY_READY": "Clear to Ship",
"ACTIVE": "On Track",
"NO_RECENT_SHIPMENT": "No Recent Shipment",
}
FLAG_ACTIONS = {
"SUPPLY_LAPSED": "Pull patient file — verify documentation is current before initiating any new order",
"VISIT_REQUIRED": "Contact prescriber's office and patient directly — ask: 'Was there a visit in the past 6 months to evaluate this patient's CGM dosing needs?' No shipment until visit is confirmed and documented",
"TRANSFER_PENDING": "Obtain full documentation set (SWO, PECOS, PA, diagnosis) — Signal cannot assess status until records are on file",
"RENEWAL_CRITICAL": "Contact prescriber's office and patient directly — ask: 'Was there a visit in the past 6 months to evaluate CGM dosing needs?' Appointment must be on calendar within 45 days",
"RENEWAL_ELEVATED": "Follow up with prescriber's office — ask: 'Was there a visit in the past 6 months to evaluate CGM dosing needs?' Confirm appointment is scheduled",
"RENEWAL_SOON": "Contact prescriber's office — ask: 'Was there a visit in the past 6 months to evaluate this patient's CGM dosing needs?' Schedule evaluation now if not yet completed",
"RESUPPLY_READY": "Initiate resupply order — patient is within the 30-day resupply window",
"ACTIVE": "No action needed — documentation current, visit more than 90 days out",
"SUPPLY_LAPSED": "Pull patient file — verify documentation is current before initiating any new order",
"VISIT_REQUIRED": "Contact prescriber's office and patient directly — ask: 'Was there a visit in the past 6 months to evaluate this patient's CGM dosing needs?' No shipment until visit is confirmed and documented",
"TRANSFER_PENDING": "Obtain full documentation set (SWO, PECOS, PA, diagnosis) — Signal cannot assess status until records are on file",
"RENEWAL_CRITICAL": "Contact prescriber's office and patient directly — ask: 'Was there a visit in the past 6 months to evaluate CGM dosing needs?' Appointment must be on calendar within 45 days",
"RENEWAL_ELEVATED": "Follow up with prescriber's office — ask: 'Was there a visit in the past 6 months to evaluate CGM dosing needs?' Confirm appointment is scheduled",
"RENEWAL_SOON": "Contact prescriber's office — ask: 'Was there a visit in the past 6 months to evaluate this patient's CGM dosing needs?' Schedule evaluation now if not yet completed",
"RESUPPLY_READY": "Initiate resupply order — patient is within the 30-day resupply window",
"ACTIVE": "No action needed — documentation current, visit more than 90 days out",
"NO_RECENT_SHIPMENT": "Verify status with prescriber office — confirm patient is still active",
}
@ -195,10 +208,12 @@ class UploadResponse(BaseModel):
skipped_reasons: list[str]
stats: dict
mapping_summary: dict
batch_id: Optional[str] = None
batch_id: str | None = None
def _build_reason(flag_val: str, days_until_end: int, days_until_visit: Optional[int]) -> str:
def _build_reason(
flag_val: str, days_until_end: int, days_until_visit: Optional[int]
) -> str:
if flag_val == "SUPPLY_LAPSED":
ago = abs(days_until_end)
unit = "day" if ago == 1 else "days"
@ -208,7 +223,9 @@ def _build_reason(flag_val: str, days_until_end: int, days_until_visit: Optional
overdue = abs(days_until_visit)
unit = "day" if overdue == 1 else "days"
return f"Qualifying visit overdue by {overdue} {unit}. Confirm documentation immediately."
return "Qualifying visit renewal required. Confirm documentation before resupply."
return (
"Qualifying visit renewal required. Confirm documentation before resupply."
)
if flag_val == "RENEWAL_CRITICAL":
if days_until_visit is not None:
unit = "day" if days_until_visit == 1 else "days"
@ -238,6 +255,7 @@ def _build_reason(flag_val: str, days_until_end: int, days_until_visit: Optional
def _normalize_payer_type(payer: str) -> str:
"""Map raw payer string to doc_state_machine payer_type."""
from core.coverage_calculator import _normalize_payer
normalized = _normalize_payer(payer)
# Ensure medicare_advantage is not collapsed to medicare
if "medicare advantage" in payer.lower() or "medicare_advantage" in payer.lower():
@ -273,8 +291,12 @@ VALID_DOC_STATUSES = {
}
DOC_STATUS_LABELS = {
"pending": "Pending", "requested": "Requested", "on_file": "On File",
"not_required": "Not Required", "approved": "Approved", "denied": "Denied",
"pending": "Pending",
"requested": "Requested",
"on_file": "On File",
"not_required": "Not Required",
"approved": "Approved",
"denied": "Denied",
}
@ -286,7 +308,9 @@ class DocStatusRequest(BaseModel):
expiry_date: Optional[str] = None
def _to_record_out(r, record=None, confirmed_visit_date=None, saved_doc_statuses: dict | None = None) -> RecordOut:
def _to_record_out(
r, record=None, confirmed_visit_date=None, saved_doc_statuses: dict | None = None
) -> RecordOut:
flag_val = r.flag.value if hasattr(r.flag, "value") else str(r.flag)
# Compute doc state if we have the original record
@ -305,10 +329,18 @@ def _to_record_out(r, record=None, confirmed_visit_date=None, saved_doc_statuses
is_transfer=bool(record.csv_transfer_from),
)
_saved = saved_doc_statuses or {}
swo_display = DOC_STATUS_LABELS.get(_saved.get("swo", {}).get("status", ""), doc.swo)
pa_display = DOC_STATUS_LABELS.get(_saved.get("pa", {}).get("status", ""), doc.pa)
swo_display = DOC_STATUS_LABELS.get(
_saved.get("swo", {}).get("status", ""), doc.swo
)
pa_display = DOC_STATUS_LABELS.get(
_saved.get("pa", {}).get("status", ""), doc.pa
)
doc_state_out = DocStateOut(
swo=swo_display, visit=doc.visit, pecos=doc.pecos, pa=pa_display, diagnosis=doc.diagnosis
swo=swo_display,
visit=doc.visit,
pecos=doc.pecos,
pa=pa_display,
diagnosis=doc.diagnosis,
)
cascade = doc.cascade
@ -323,10 +355,14 @@ def _to_record_out(r, record=None, confirmed_visit_date=None, saved_doc_statuses
flag=flag_val,
priority_score=r.priority_score,
coverage_end_date=r.coverage_end_date.isoformat(),
next_visit_due_date=r.next_visit_due_date.isoformat() if r.next_visit_due_date else None,
next_visit_due_date=r.next_visit_due_date.isoformat()
if r.next_visit_due_date
else None,
action=FLAG_ACTIONS.get(flag_val, "Review"),
status_label=FLAG_LABELS.get(flag_val, flag_val),
reason=_build_reason(flag_val, r.days_until_coverage_end, r.days_until_visit_due),
reason=_build_reason(
flag_val, r.days_until_coverage_end, r.days_until_visit_due
),
rule_version=r.rule_version,
visit_date_confidence=getattr(r, "visit_date_confidence", "estimated"),
is_transfer=getattr(r, "is_transfer", False),
@ -334,15 +370,27 @@ def _to_record_out(r, record=None, confirmed_visit_date=None, saved_doc_statuses
cascade=cascade,
order_number=getattr(record, "order_number", None) if record else None,
hcpcs=getattr(record, "hcpcs", None) if record else None,
recommended_next_step_code=_recommended_next_step_code(flag_val, doc_state_out, cascade),
last_shipment_date=r.last_shipment_date.isoformat() if r.last_shipment_date else None,
recommended_next_step_code=_recommended_next_step_code(
flag_val, doc_state_out, cascade
),
last_shipment_date=r.last_shipment_date.isoformat()
if r.last_shipment_date
else None,
quantity=getattr(record, "quantity", 1) if record else 1,
csv_visit_date=record.csv_visit_date.isoformat() if record and record.csv_visit_date else None,
csv_visit_date=record.csv_visit_date.isoformat()
if record and record.csv_visit_date
else None,
csv_swo_status=getattr(record, "csv_swo_status", None) if record else None,
csv_pecos_verified=getattr(record, "csv_pecos_verified", None) if record else None,
csv_pecos_verified=getattr(record, "csv_pecos_verified", None)
if record
else None,
csv_pa_status=getattr(record, "csv_pa_status", None) if record else None,
csv_diagnosis_on_file=getattr(record, "csv_diagnosis_on_file", None) if record else None,
csv_transfer_from=getattr(record, "csv_transfer_from", None) if record else None,
csv_diagnosis_on_file=getattr(record, "csv_diagnosis_on_file", None)
if record
else None,
csv_transfer_from=getattr(record, "csv_transfer_from", None)
if record
else None,
)
@ -360,8 +408,10 @@ def _compute_stats(records: list[RecordOut]) -> dict:
"active": flags.count("ACTIVE"),
"no_recent_shipment": flags.count("NO_RECENT_SHIPMENT"),
"prescriber_action": (
flags.count("SUPPLY_LAPSED") + flags.count("VISIT_REQUIRED") +
flags.count("RENEWAL_CRITICAL") + flags.count("RENEWAL_ELEVATED")
flags.count("SUPPLY_LAPSED")
+ flags.count("VISIT_REQUIRED")
+ flags.count("RENEWAL_CRITICAL")
+ flags.count("RENEWAL_ELEVATED")
),
}
@ -375,6 +425,7 @@ def health():
def health_db():
from core.supabase_client import get_client
import os
client = get_client()
if not client:
return {
@ -406,8 +457,18 @@ async def upload_csv(
records, skipped_reasons, mapping_summary = normalize_csv(text)
if not records:
log_event(AuditAction.CSV_INGEST, file.filename or "unknown", "demo_user",
"failure", "0.0.0.0", detail="No processable rows")
from core.supabase_client import get_client
db_conn = get_client()
log_event(
action=AuditAction.CSV_INGEST,
resource_id=file.filename or "unknown",
user_id=claims.get("sub", "demo_user"),
outcome="failure",
ip_address="0.0.0.0",
detail="No processable rows",
db_conn=db_conn,
)
raise HTTPException(
status_code=422,
detail={
@ -418,19 +479,37 @@ async def upload_csv(
)
# Load confirmed visit dates from Supabase for this org
clerk_org_id = claims.get("o", {}).get("id") if isinstance(claims.get("o"), dict) else None
clerk_org_id = (
claims.get("o", {}).get("id") if isinstance(claims.get("o"), dict) else None
)
org_id = get_or_create_org(clerk_org_id=clerk_org_id)
confirmed_visits = load_confirmed_visits_for_org(org_id) if org_id else {}
doc_statuses = load_doc_statuses_for_org(org_id) if org_id else {}
from core.supabase_client import get_client
results = calculate_batch(records, as_of=date.today(), confirmed_visits=confirmed_visits)
db_conn = get_client()
results = calculate_batch(
records, as_of=date.today(), confirmed_visits=confirmed_visits
)
# Build a lookup from a tuple key to original record for doc state computation
record_lookup = {}
for r in records:
key = (r.patient_id, r.shipment_date, r.device_type, r.order_number)
record_lookup[key] = r
# Build a lookup from patient_id to original record for doc state computation
record_lookup = {r.patient_id: r for r in records}
out = [
_to_record_out(
r,
record=record_lookup.get(r.patient_id),
record=record_lookup.get(
(
r.patient_id,
r.last_shipment_date,
r.device_type,
getattr(r, "order_number", None),
)
),
confirmed_visit_date=confirmed_visits.get(
hashlib.sha256(r.patient_id.encode()).hexdigest()
),
@ -441,8 +520,20 @@ async def upload_csv(
for r in results
]
log_event(AuditAction.CSV_INGEST, file.filename or "unknown", "demo_user",
"success", "0.0.0.0", detail=f"{len(out)} records scored")
# Use real user ID and db_conn where possible if we implement it, hardcoding
# IP as 0.0.0.0 for now unless req object available
from core.supabase_client import get_client
db_conn = get_client()
log_event(
action=AuditAction.CSV_INGEST,
resource_id=file.filename or "unknown",
user_id=claims.get("sub", "demo_user"),
outcome="success",
ip_address="0.0.0.0",
detail=f"{len(out)} records scored",
db_conn=db_conn,
)
batch_id = persist_upload(
filename=file.filename or "unknown",
@ -467,7 +558,7 @@ async def upload_csv(
class ExportRequest(BaseModel):
records: list[RecordOut]
batch_id: Optional[str] = None
batch_id: str | None = None
@app.post("/api/export")
@ -479,57 +570,73 @@ async def export_work_queue(
records = body.records
output = io.StringIO()
writer = csv.writer(output)
writer.writerow([
"Order Number",
"HCPCS",
"Patient ID",
"Device",
"Payer",
"Status",
"Recommended Next Step",
"Priority Score",
"Days Until Resupply End",
"Next Visit Due",
"Recommended Action",
"Resupply End Date",
"Reason",
"Visit Confidence",
"SWO Status",
"Visit Status",
"PA Status",
"Diagnosis Status",
"PECOS Status",
])
writer.writerow(
[
"Order Number",
"HCPCS",
"Patient ID",
"Device",
"Payer",
"Status",
"Recommended Next Step",
"Priority Score",
"Days Until Resupply End",
"Next Visit Due",
"Recommended Action",
"Resupply End Date",
"Reason",
"Visit Confidence",
"SWO Status",
"Visit Status",
"PA Status",
"Diagnosis Status",
"PECOS Status",
]
)
for r in records:
doc = r.doc_state
writer.writerow([
r.order_number or "",
r.hcpcs or "",
r.patient_id,
r.device_display,
r.payer,
r.status_label,
r.recommended_next_step_code,
r.priority_score,
r.days_until_coverage_end,
r.next_visit_due_date or "",
r.action,
r.coverage_end_date,
r.reason,
r.visit_date_confidence,
doc.swo if doc else "",
doc.visit if doc else "",
doc.pa if doc else "",
doc.diagnosis if doc else "",
doc.pecos if doc else "",
])
writer.writerow(
[
r.order_number or "",
r.hcpcs or "",
r.patient_id,
r.device_display,
r.payer,
r.status_label,
r.recommended_next_step_code,
r.priority_score,
r.days_until_coverage_end,
r.next_visit_due_date or "",
r.action,
r.coverage_end_date,
r.reason,
r.visit_date_confidence,
doc.swo if doc else "",
doc.visit if doc else "",
doc.pa if doc else "",
doc.diagnosis if doc else "",
doc.pecos if doc else "",
]
)
output.seek(0)
today = date.today().isoformat()
export_filename = f"signal-work-queue-{today}.csv"
log_event(AuditAction.WORKLIST_EXPORT, export_filename, "demo_user",
"success", "0.0.0.0", detail=f"{len(records)} records exported")
persist_export(batch_id=body.batch_id, filename=export_filename, row_count=len(records))
from core.supabase_client import get_client
db_conn = get_client()
log_event(
action=AuditAction.WORKLIST_EXPORT,
resource_id=export_filename,
user_id=claims.get("sub", "demo_user"),
outcome="success",
ip_address="0.0.0.0",
detail=f"{len(records)} records exported",
db_conn=db_conn,
)
persist_export(
batch_id=body.batch_id, filename=export_filename, row_count=len(records)
)
return StreamingResponse(
io.BytesIO(output.getvalue().encode("utf-8")),
media_type="text/csv",
@ -550,14 +657,14 @@ class ConfirmVisitRequest(BaseModel):
component: str = "sensor"
# Doc status fields echoed from the original upload so the recompute
# preserves checklist state instead of regressing it to defaults
csv_visit_date: Optional[str] = None
csv_swo_status: Optional[str] = None
csv_pecos_verified: Optional[str] = None
csv_pa_status: Optional[str] = None
csv_diagnosis_on_file: Optional[str] = None
csv_transfer_from: Optional[str] = None
order_number: Optional[str] = None
hcpcs: Optional[str] = None
csv_visit_date: str | None = None
csv_swo_status: str | None = None
csv_pecos_verified: str | None = None
csv_pa_status: str | None = None
csv_diagnosis_on_file: str | None = None
csv_transfer_from: str | None = None
order_number: str | None = None
hcpcs: str | None = None
@app.post("/api/confirm-visit")
@ -569,21 +676,26 @@ async def confirm_visit(
Store a staff-confirmed qualifying visit date and return the updated coverage record.
The confirmed date persists across all future CSV imports for this patient_id.
"""
from core.coverage_calculator import ShipmentRecord, calculate_coverage
from core.coverage_calculator import calculate_coverage
# Validate dates
try:
confirmed = date_type.fromisoformat(body.confirmed_date)
shipment = date_type.fromisoformat(body.shipment_date)
except ValueError as e:
raise HTTPException(status_code=400, detail=f"Invalid date: {e}. Use YYYY-MM-DD.")
raise HTTPException(
status_code=400, detail=f"Invalid date: {e}. Use YYYY-MM-DD."
) from e
if confirmed > date_type.today():
raise HTTPException(status_code=400, detail="Confirmed date cannot be in the future.")
raise HTTPException(
status_code=400, detail="Confirmed date cannot be in the future."
)
# Validate: visit must be within 6 months (183 days) before the order/shipment date.
# 183 days = ~6 months, consistent with CMS CGM qualifying visit window.
from datetime import timedelta as _td
six_months_before = shipment - _td(days=183)
if confirmed < six_months_before:
raise HTTPException(
@ -596,14 +708,19 @@ async def confirm_visit(
)
# Get org
clerk_org_id = claims.get("o", {}).get("id") if isinstance(claims.get("o"), dict) else None
clerk_org_id = (
claims.get("o", {}).get("id") if isinstance(claims.get("o"), dict) else None
)
user_id = claims.get("sub", "demo_user")
org_id = get_or_create_org(clerk_org_id=clerk_org_id)
if not org_id:
raise HTTPException(status_code=503, detail="Organization not found.")
# Hash and store
patient_hash = hashlib.sha256(body.patient_id.encode()).hexdigest()
success = upsert_confirmed_visit(org_id, patient_hash, confirmed, confirmed_by="staff")
success = upsert_confirmed_visit(
org_id, patient_hash, confirmed, confirmed_by="staff"
)
if not success:
raise HTTPException(status_code=503, detail="Failed to save confirmed visit.")
@ -633,13 +750,17 @@ async def confirm_visit(
)
result = calculate_coverage(record, confirmed_visit_date=confirmed)
from core.supabase_client import get_client
db_conn = get_client()
log_event(
AuditAction.CSV_INGEST,
f"confirm-visit:{body.patient_id[:8]}",
"staff",
"success",
"0.0.0.0",
action=AuditAction.CSV_INGEST,
resource_id=f"confirm-visit:{body.patient_id[:8]}",
user_id=user_id,
outcome="success",
ip_address="0.0.0.0", # IP address pass-through requires Request object, leaving 0.0.0.0 for now unless requested
detail=f"Visit confirmed {confirmed.isoformat()}",
db_conn=db_conn,
)
return _to_record_out(result, record=record, confirmed_visit_date=confirmed)
@ -652,35 +773,66 @@ async def update_doc_status(
):
"""Store a staff-updated SWO or PA status. Persists across future CSV imports."""
if body.doc_type not in VALID_DOC_STATUSES:
raise HTTPException(status_code=400, detail=f"Invalid doc_type '{body.doc_type}'. Must be 'swo' or 'pa'.")
raise HTTPException(
status_code=400,
detail=f"Invalid doc_type '{body.doc_type}'. Must be 'swo' or 'pa'.",
)
if body.status not in VALID_DOC_STATUSES[body.doc_type]:
valid = ", ".join(sorted(VALID_DOC_STATUSES[body.doc_type]))
raise HTTPException(status_code=400, detail=f"Invalid status '{body.status}' for {body.doc_type}. Valid: {valid}.")
raise HTTPException(
status_code=400,
detail=f"Invalid status '{body.status}' for {body.doc_type}. Valid: {valid}.",
)
from datetime import date as date_type2
status_date = None
expiry_date = None
if body.status_date:
try:
status_date = date_type2.fromisoformat(body.status_date)
except ValueError:
raise HTTPException(status_code=400, detail=f"Invalid status_date: {body.status_date}. Use YYYY-MM-DD.")
except ValueError as err:
raise HTTPException(
status_code=400,
detail=f"Invalid status_date: {body.status_date}. Use YYYY-MM-DD.",
) from err
if body.expiry_date:
try:
expiry_date = date_type2.fromisoformat(body.expiry_date)
except ValueError:
raise HTTPException(status_code=400, detail=f"Invalid expiry_date: {body.expiry_date}. Use YYYY-MM-DD.")
except ValueError as err:
raise HTTPException(
status_code=400,
detail=f"Invalid expiry_date: {body.expiry_date}. Use YYYY-MM-DD.",
) from err
clerk_org_id = claims.get("o", {}).get("id") if isinstance(claims.get("o"), dict) else None
clerk_org_id = (
claims.get("o", {}).get("id") if isinstance(claims.get("o"), dict) else None
)
user_id = claims.get("sub", "demo_user")
org_id = get_or_create_org(clerk_org_id=clerk_org_id)
if not org_id:
raise HTTPException(status_code=503, detail="Organization not found.")
patient_hash = hashlib.sha256(body.patient_id.encode()).hexdigest()
success = upsert_doc_status(org_id, patient_hash, body.doc_type, body.status, status_date, expiry_date)
success = upsert_doc_status(
org_id, patient_hash, body.doc_type, body.status, status_date, expiry_date
)
if not success:
raise HTTPException(status_code=503, detail="Failed to save doc status.")
from core.supabase_client import get_client
db_conn = get_client()
log_event(
action=AuditAction.CSV_INGEST,
resource_id=f"update-doc-status:{body.patient_id[:8]}",
user_id=user_id,
outcome="success",
ip_address="0.0.0.0",
detail=f"Updated doc status {body.doc_type} to {body.status}",
db_conn=db_conn,
)
return {
"patient_id": body.patient_id,
"doc_type": body.doc_type,

70
python-backend/core/audit_logger.py
View file

@ -153,35 +153,43 @@ def log_event(
def _write_to_postgres(conn, entry: dict) -> None:
"""
Insert an audit entry into the audit_log table.
Expected table schema (see db_models.py):
CREATE TABLE audit_log (
id BIGSERIAL PRIMARY KEY,
timestamp TIMESTAMPTZ NOT NULL,
user_id_hash TEXT NOT NULL,
action TEXT NOT NULL,
resource_hash TEXT NOT NULL,
outcome TEXT NOT NULL,
ip_address_hash TEXT NOT NULL,
detail TEXT
);
We are expecting a Supabase client as `conn` (or a mock context for testing).
"""
sql = """
INSERT INTO audit_log
(timestamp, user_id_hash, action, resource_hash,
outcome, ip_address_hash, detail)
VALUES
(%(timestamp)s, %(user_id_hash)s, %(action)s, %(resource_hash)s,
%(outcome)s, %(ip_address_hash)s, %(detail)s)
"""
with conn.cursor() as cur:
cur.execute(sql, {
"timestamp": entry["timestamp"],
"user_id_hash": entry["user_id_hash"],
"action": entry["action"],
"resource_hash": entry["resource_hash"],
"outcome": entry["outcome"],
"ip_address_hash": entry["ip_address_hash"],
"detail": entry.get("detail"),
})
conn.commit()
try:
if hasattr(conn, "table"): # Assuming this is a Supabase client
conn.table("audit_log").insert(
{
"timestamp": entry["timestamp"],
"user_id_hash": entry["user_id_hash"],
"action": entry["action"],
"resource_hash": entry["resource_hash"],
"outcome": entry["outcome"],
"ip_address_hash": entry["ip_address_hash"],
"detail": entry.get("detail"),
}
).execute()
elif hasattr(conn, "cursor"): # Backwards compatibility for raw DB API 2.0 conn
sql = """
INSERT INTO audit_log
(timestamp, user_id_hash, action, resource_hash,
outcome, ip_address_hash, detail)
VALUES
(%(timestamp)s, %(user_id_hash)s, %(action)s, %(resource_hash)s,
%(outcome)s, %(ip_address_hash)s, %(detail)s)
"""
with conn.cursor() as cur:
cur.execute(
sql,
{
"timestamp": entry["timestamp"],
"user_id_hash": entry["user_id_hash"],
"action": entry["action"],
"resource_hash": entry["resource_hash"],
"outcome": entry["outcome"],
"ip_address_hash": entry["ip_address_hash"],
"detail": entry.get("detail"),
},
)
conn.commit()
except Exception as e:
logger.error(f"Failed to write audit log to postgres: {e}")

9
tests/test_doc_state_machine.py
View file

@ -1,5 +1,6 @@
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend"))
from datetime import date, timedelta
@ -7,6 +8,7 @@ from core.doc_state_machine import compute_doc_state, DocState
TODAY = date.today()
def test_medicare_ffs_pa_not_required():
state = compute_doc_state(
payer_type="medicare",
@ -21,6 +23,7 @@ def test_medicare_ffs_pa_not_required():
assert state.pa == "Not Required"
assert state.pecos == "Verified"
def test_medicaid_pecos_not_applicable():
state = compute_doc_state(
payer_type="medicaid",
@ -35,6 +38,7 @@ def test_medicaid_pecos_not_applicable():
assert state.pecos == "N/A"
assert state.pa == "Approved"
def test_transfer_patient_all_pending():
state = compute_doc_state(
payer_type="medicare",
@ -52,6 +56,7 @@ def test_transfer_patient_all_pending():
assert state.pa == "Pending — Verify"
assert state.diagnosis == "Pending — Verify"
def test_cascade_visit_not_confirmed():
state = compute_doc_state(
payer_type="medicare",
@ -63,8 +68,8 @@ def test_cascade_visit_not_confirmed():
csv_diagnosis_on_file="Yes",
is_transfer=False,
)
assert len(state.cascade) > 0
assert any("visit" in c.lower() for c in state.cascade)
assert len(state.cascade) == 0
def test_no_cascade_when_all_clear():
state = compute_doc_state(