"""Seeds the "RTI app" tenant with a sample workflow that uses every feature of the workflow engine (app/crud/workflow_engine.py), as drafts: * "RTI — screen application (sample)" — a Task the service step runs; * "RTI — information from another office (sample)" — the sub-workflow; * "RTI request — full process (sample)" — the main workflow: Start with a 30-day case SLA → Task step (with a retry) → decision (life or liberty) → urgent PIO review (48 h SLA, reminder, escalates by moving on) or Clerk seat assignment (24 h SLA, reminder, escalation notifies the Nodal officer) → parallel split: PIO drafts a reply, Approvers clear it (everyone must approve), and — only when needed — one sub-workflow per other office → join → PIO decision (assignee from a field, escalation reassigns to the Appellate authority, an on-enter email notification) → reply sent and a 30-day appeal window (timer), additional fee demanded and a race between "payment received" and a 7-day deadline (join: first one wins), or rejected (ends everything) → appeal decided by a panel (majority vote). Counted on business days. Goes through the API as the RTI designer, like seed_rti_app_records.py. Idempotent: drafts are updated in place; anything already published is left as it is. Nothing is published — do that in the Studio (the Task, then the sub-workflow, then the main workflow) or with --publish. python scripts/seed_rti_sample_workflow.py [--api http://127.0.0.1:8000] [--publish] """ import argparse import asyncio import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) sys.path.insert(0, str(Path(__file__).resolve().parent)) from grc_app.client import Api, user_identity # noqa: E402 TENANT = "RTI app" DESIGNER = "vidyakumar.s@saasvaap.com" TASK_NAME = "RTI — screen application (sample)" CHILD_NAME = "RTI — information from another office (sample)" MAIN_NAME = "RTI request — full process (sample)" def step(key, type_, label, x, y, to=(), **extra) -> dict: nexts = [t if isinstance(t, dict) else {"to": t, "condition": None} for t in to] return {"key": key, "type": type_, "label": label, "position_x": x, "position_y": y, "next_steps": nexts, **extra} def when(field, op, value) -> dict: return {"driver_field": field, "operator": op, "value": value} def inp(name, type_="string") -> dict: return {"name": name, "type": type_, "source_field": name, "required": False} def out(name, type_="string") -> dict: return {"name": name, "type": type_, "source": name} def by_name(items: list[dict], name: str, key: str = "name") -> dict: found = next((i for i in items if i.get(key) == name), None) if found is None: raise SystemExit(f"{name!r} isn't in the RTI app tenant") return found def upsert_task(api: Api) -> dict: nodes = [ {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "screen"}], "position_x": 40, "position_y": 120}, { "key": "screen", "type": "set_variable", "label": "Mark as screened", "input_params": [{"name": "screened", "type": "string", "value": "yes"}, {"name": "screening_note", "type": "string", "value": "Checked by the sample Task"}], "position_x": 320, "position_y": 120, }, ] existing = next((t for t in api.get("/tasks", limit=200) if t["name"] == TASK_NAME), None) if existing is None: print("creating Task", TASK_NAME) return api.post("/tasks", {"name": TASK_NAME, "nodes": nodes}) if existing["status"] == "draft": print("updating Task", TASK_NAME) return api.patch(f"/tasks/{existing['id']}", {"name": TASK_NAME, "nodes": nodes}) print("Task already published — left as it is") return existing def upsert_workflow(api: Api, name: str, steps: list[dict], settings: dict) -> dict: existing = next((w for w in api.get("/workflows", limit=200) if w["name"] == name), None) if existing is None: print("creating workflow", name) return api.post("/workflows", {"name": name, "step_definitions": steps, "settings": settings}) if existing["status"] == "draft": print("updating workflow", name) return api.patch(f"/workflows/{existing['workflow_id']}", {"name": name, "step_definitions": steps, "settings": settings}) print("workflow already published — left as it is:", name) return existing def child_steps(roles: dict, forms: dict) -> list[dict]: return [ step("start", "start", "Request to office", 40, 200, ["provide"], input_params=[inp("office_id", "any"), inp("application_number")]), step( "provide", "task", "Office PIO provides information", 320, 200, ["done"], approver_roles=[roles["PIO"]], sla_hours=72, remind_before_hours=12, target_type="form", target_id=forms["RTI Reply"], output_params=[out("info_summary")], ), step("done", "end", "Information sent", 600, 200), ] def main_steps(roles: dict, forms: dict, task_id: str, child_id: str, smtp_id: str | None) -> list[dict]: pio, clerk, approver = roles["PIO"], roles["Clerk"], roles["Approver"] appellate, nodal, owner = roles["Appellate authority"], roles["Nodal officer"], roles["Process Owner"] notify_pio = ( [{"connector_id": smtp_id, "trigger": "on_enter", "message": "An RTI application is waiting for the PIO's decision.", "recipient_field": None, "recipient_role_ids": [pio], "recipient_user_ids": []}] if smtp_id else [] ) return [ step("urgent_review", "approval", "Urgent PIO review", 830, 60, ["gather"], approver_roles=[pio], sla_hours=48, remind_before_hours=6, escalation={"action": "route", "role_ids": [], "user_ids": [], "route_to": "pio_decision"}), step("draft_reply", "task", "PIO drafts reply", 1370, 60, ["info_join"], approver_roles=[pio], sla_hours=72, target_type="form", target_id=forms["RTI Reply"], output_params=[out("reply_subject")]), step("start", "start", "Application received", 20, 200, ["screen"], sla_hours=720, input_params=[inp("application_number"), inp("life_and_liberty", "boolean"), inp("needs_other_offices", "boolean"), inp("other_office_ids", "list"), inp("assigned_pio_user_id")]), step("screen", "service", "Screen application", 290, 200, ["urgency"], service={"task_id": task_id, "retries": 1}), step("urgency", "decision", "Life or liberty?", 560, 200, [ {"to": "urgent_review", "condition": when("life_and_liberty", "equals", True)}, {"to": "assign_seat", "condition": None}, ]), step("assign_seat", "task", "Clerk assigns seat", 830, 200, ["gather"], approver_roles=[clerk], sla_hours=24, remind_before_hours=4, escalation={"action": "notify", "role_ids": [nodal], "user_ids": [], "route_to": None}, target_type="form", target_id=forms["Seat Assignment"]), step("gather", "parallel", "Gather information", 1100, 200, [ {"to": "draft_reply", "condition": None}, {"to": "clearance", "condition": None}, {"to": "other_offices", "condition": when("needs_other_offices", "equals", True)}, ]), step("clearance", "approval", "Legal & finance clearance", 1370, 200, ["info_join", "rejected"], approver_roles=[approver], approval_mode="all", sla_hours=48), step("info_join", "join", "Information gathered", 1100, 340, ["pio_decision"], join_mode="all"), step("other_offices", "subprocess", "Ask other offices", 1370, 340, ["info_join"], subprocess={ "workflow_id": child_id, "for_each": "other_office_ids", "input_map": {"office_id": "item", "application_number": ""}, "output_map": {"other_office_replies": "info_summary"}, "wait": True, }), step("pio_decision", "approval", "PIO decision", 830, 340, ["appeal_window", "demand_fee", "rejected"], approver_roles=[pio], assignee_field="assigned_pio_user_id", sla_hours=120, remind_before_hours=24, escalation={"action": "reassign", "role_ids": [appellate], "user_ids": [], "route_to": None}, notifications=notify_pio, output_params=[out("decision_note")]), # The fee loop step("demand_fee", "task", "Demand additional fee", 830, 480, ["fee_race"], approver_roles=[clerk], target_type="form", target_id=forms["Additional Fee Demand"]), step("fee_race", "parallel", "Await payment", 560, 480, ["confirm_payment", "fee_deadline"]), step("confirm_payment", "task", "Payment received", 290, 420, ["fee_join"], approver_roles=[clerk], output_params=[out("payment_confirmed", "boolean")]), step("fee_deadline", "timer", "7-day payment deadline", 290, 560, ["fee_join"], timer={"mode": "duration", "hours": 168, "field": None}), step("fee_join", "join", "Paid or out of time", 20, 480, ["fee_paid"], join_mode="any"), step("fee_paid", "decision", "Fee paid?", 20, 640, [ {"to": "pio_decision", "condition": when("payment_confirmed", "equals", True)}, {"to": "closed_unpaid", "condition": None}, ]), step("closed_unpaid", "end", "Closed — fee not paid", 290, 700), # Reply and appeal step("appeal_window", "timer", "Reply sent (30-day appeal window)", 1100, 480, ["appeal_filed"], timer={"mode": "duration", "hours": 720, "field": None}), step("appeal_filed", "decision", "First appeal filed?", 1370, 480, [ {"to": "appeal_panel", "condition": when("appeal_filed", "equals", True)}, {"to": "closed", "condition": None}, ]), step("appeal_panel", "approval", "Appeal panel", 1370, 640, ["pio_decision", "closed"], approver_roles=[appellate], approval_mode="majority", sla_hours=720, escalation={"action": "notify", "role_ids": [owner], "user_ids": [], "route_to": None}), step("rejected", "end", "Rejected", 560, 340, terminate=True), step("closed", "end", "Closed", 1100, 640), ] async def problems(tenant_id, workflow_ids: list[str]) -> list[list[str]]: """What would stop each workflow from publishing (the engine's own check).""" from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.pool import NullPool from app.crud.workflow_engine import definition_problems from app.models.workflow import Workflow from tests.conftest import _superuser_engine # Its own connection: this runs in a different event loop from user_identity's. engine = create_async_engine(_superuser_engine.url, poolclass=NullPool) try: async with AsyncSession(engine) as db: out = [] for workflow_id in workflow_ids: wf = (await db.execute(select(Workflow).where(Workflow.workflow_id == workflow_id, Workflow.is_current.is_(True)))).scalar_one() out.append(await definition_problems(db, tenant_id, wf)) return out finally: await engine.dispose() def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--api", default="http://127.0.0.1:8000") parser.add_argument("--publish", action="store_true", help="publish the Task, the sub-workflow and the main workflow") args = parser.parse_args() tid, uid, vid = asyncio.run(user_identity(TENANT, DESIGNER)) api = Api(args.api, tid, uid, vid) roles = {r["name"]: r["id"] for r in api.get("/roles")} for needed in ("PIO", "Clerk", "Approver", "Appellate authority", "Nodal officer", "Process Owner"): if needed not in roles: raise SystemExit(f"role {needed!r} isn't in the RTI app tenant") forms = {f["name"]: f["form_id"] for f in api.get("/forms", limit=200)} for needed in ("RTI Reply", "Seat Assignment", "Additional Fee Demand"): by_name([{"name": n} for n in forms], needed) smtp = next((c["id"] for c in api.get("/connectors", limit=200) if c["type"] == "email_smtp" and c["status"] == "enabled"), None) task = upsert_task(api) child = upsert_workflow(api, CHILD_NAME, child_steps(roles, forms), {"business_days": True}) main_wf = upsert_workflow(api, MAIN_NAME, main_steps(roles, forms, task["id"], child["workflow_id"], smtp), {"business_days": True}) for wf, found in zip((child, main_wf), asyncio.run(problems(tid, [child["workflow_id"], main_wf["workflow_id"]]))): print(f"{wf['name']}: " + ("ready to publish" if not found else "problems: " + " · ".join(found))) if args.publish: if task["status"] == "draft": api.post(f"/tasks/{task['id']}/publish") for wf in (child, main_wf): if wf["status"] == "draft": api.post(f"/workflows/{wf['workflow_id']}/publish") print("published") print("main workflow:", main_wf["workflow_id"]) if __name__ == "__main__": main()