22 - Approval Page

from pydantic import BaseModel
from polos import workflow, WorkflowContext


class DeployRequest(BaseModel):
    service: str
    version: str
    environment: str


class DeployResult(BaseModel):
    service: str
    version: str
    environment: str
    status: str
    approved_by: str | None = None
    reason: str | None = None


@workflow(id="deploy_with_approval")
async def deploy_workflow(ctx: WorkflowContext, payload: DeployRequest) -> DeployResult:
    # Step 1: Run pre-deploy checks
    checks = await ctx.step.run(
        "pre_deploy_checks",
        lambda: {
            "tests_pass": True,
            "build_success": True,
            "service": payload.service,
            "version": payload.version,
        },
    )

    # Step 2: Suspend and wait for human approval via the web UI.
    # The _form schema tells the approval page what to render.
    resume_data = await ctx.step.suspend(
        "approve_deploy",
        data={
            "_form": {
                "title": f"Deploy {payload.service} v{payload.version}",
                "description": (
                    f"Approve deployment to {payload.environment}. "
                    "All pre-deploy checks passed."
                ),
                "fields": [
                    {
                        "name": "approved",
                        "type": "boolean",
                        "label": "Approve this deployment",
                        "default": False,
                    },
                    {
                        "name": "approver",
                        "type": "text",
                        "label": "Your name",
                        "required": True,
                    },
                    {
                        "name": "reason",
                        "type": "textarea",
                        "label": "Comments",
                        "description": "Optional reason or notes for this decision",
                    },
                ],
                "context": {
                    "service": payload.service,
                    "version": payload.version,
                    "environment": payload.environment,
                    "tests": "passing" if checks["tests_pass"] else "failing",
                    "build": "success" if checks["build_success"] else "failed",
                },
            },
        },
        timeout=86400,  # 24 hour timeout
    )

    # Step 3: Process the decision
    decision = resume_data.get("data", resume_data) if isinstance(resume_data, dict) else {}
    approved = bool(decision.get("approved"))
    approved_by = str(decision.get("approver", "unknown"))
    reason = str(decision["reason"]) if decision.get("reason") is not None else None

    if approved:
        await ctx.step.run("execute_deploy", lambda: {"deployed": True})

    return DeployResult(
        service=payload.service,
        version=payload.version,
        environment=payload.environment,
        status="deployed" if approved else "rejected",
        approved_by=approved_by,
        reason=reason,
    )

handle = await deploy_workflow.invoke(
    client,
    {"service": "api-gateway", "version": "2.4.0", "environment": "production"},
)

async for event in events.stream_workflow(client, handle.root_workflow_id, handle.id):
    if event.event_type and event.event_type.startswith("suspend_"):
        data = event.data if isinstance(event.data, dict) else {}
        approval_url = data.get("_approval_url")
        step_key = event.event_type[len("suspend_"):]

        ui_base_url = os.getenv("POLOS_UI_URL", "http://localhost:5173")
        display_url = approval_url.replace(api_url, ui_base_url) if approval_url else \
            f"{ui_base_url}/approve/{handle.id}/{step_key}"

        print(f"  Open this URL in your browser:\n  {display_url}")
        print("  Fill in the form and click Submit.")

git clone https://github.com/polos-dev/polos.git
cd polos/python-examples/22-approval-page
cp .env.example .env
uv sync
python worker.py      # Terminal 1
python main.py        # Terminal 2