Minimal Prefect pause-for-input inspection example

pause_input_inspection.py

"""
Minimal example: inspect values submitted to a Prefect pause-for-input form.

Run the flow, resume it from the UI with a value, then run:

    python pause_input_inspection.py <flow-run-id>
"""

from __future__ import annotations

import asyncio
import json
import sys
from uuid import UUID

from prefect import flow
from prefect.client.orchestration import get_client
from prefect.flow_runs import pause_flow_run
from prefect.input import RunInput


class Approval(RunInput):
    approved: bool
    note: str | None = None


@flow
def needs_approval() -> None:
    approval = pause_flow_run(wait_for_input=Approval)
    print(f"Approved: {approval.approved}; note: {approval.note!r}")


async def inspect_pause_inputs(flow_run_id: UUID) -> None:
    async with get_client() as client:
        states = await client.read_flow_run_states(flow_run_id)

        for state in states:
            keyset = state.state_details.run_input_keyset
            if not keyset:
                continue

            print(f"\nPaused state: {state.name} ({state.id})")

            for label in ("schema", "description", "response"):
                key = keyset.get(label)
                if not key:
                    continue

                try:
                    raw = await client.read_flow_run_input(flow_run_id, key)
                except Exception:
                    raw = None

                if raw is None:
                    print(f"{label}: <not stored>")
                    continue

                try:
                    value = json.loads(raw)
                except json.JSONDecodeError:
                    value = raw

                print(f"{label}: {json.dumps(value, indent=2)}")


if __name__ == "__main__":
    if len(sys.argv) == 1:
        needs_approval()
    else:
        asyncio.run(inspect_pause_inputs(UUID(sys.argv[1])))