File size: 1,153 Bytes
3ec36d7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 | """
Workflow progress tracking.
Persists the current node to the workflow_checkpoints table so the
frontend can poll and show live stage progression during execution.
"""
from __future__ import annotations
from app.database.database import SessionLocal
from app.models.workflow_checkpoint import WorkflowCheckpoint
def report_progress(workflow_run_id, stage_name: str):
"""
Write/update the current stage to the database.
Uses its own session so it commits immediately and is visible to API polls.
"""
db = SessionLocal()
try:
# Upsert: replace the PROGRESS checkpoint
db.query(WorkflowCheckpoint).filter(
WorkflowCheckpoint.workflow_run_id == workflow_run_id,
WorkflowCheckpoint.agent_name == "PROGRESS",
).delete(synchronize_session=False)
checkpoint = WorkflowCheckpoint(
workflow_run_id=workflow_run_id,
agent_name="PROGRESS",
state={"current_node": stage_name},
message=stage_name,
)
db.add(checkpoint)
db.commit()
except Exception:
db.rollback()
finally:
db.close()
|