Absolutely—here’s a tiny “state tester” flow you can deploy and use to trigger specific Prefect states on demand. It’s designed for Prefect 3.x. What it does: - mode=success → Completed - mode=fail → Failed (raises exception) - mode=timeout → Failed with name “TimedOut” - mode=crash → Crashed (raises SystemExit) - mode=hang → use it to test Cancelling/Canceled Put this in a file like `flows/state_tester.py`:

from prefect import flow, task
import time
import sys

@task
def succeed():
    return "all good"

@task
def fail():
    raise ValueError("intentional failure")

# This will enter a TimedOut -> Failed state
@task(timeout_seconds=2)
def timeout_task():
    time.sleep(10)

# This produces a Crashed state in Prefect (safer than os._exit)
@task
def crash():
    raise SystemExit("intentional crash via SystemExit")

@task
def hang(seconds: int = 300):
    time.sleep(seconds)

@flow(name="state-tester")
def state_tester(mode: str = "success"):
    if mode == "success":
        return succeed()
    elif mode == "fail":
        return fail()
    elif mode == "timeout":
        return timeout_task()
    elif mode == "crash":
        return crash()
    elif mode == "hang":
        return hang()
    else:
        raise ValueError(f"Unknown mode: {mode}")

Notes: - Use

SystemExit

(as above) to reliably get a Crashed state without killing your worker. Avoid

os._exit()

for testing—it hard-exits the process. - A “timeout” task becomes a

Failed

state with name “TimedOut”. Quick deploy and run 1) Create a work pool and start a worker (local process example):

prefect work-pool create "local-pool" --type process
prefect worker start -p local-pool

2) Deploy the flow (from your project root): - If you’re using a Prefect project (via

prefect init

):

prefect deploy flows/state_tester.py:state_tester -n state-tester -p local-pool

- Or programmatically (no project required):

python -c "from prefect.flows import Flow; Flow.from_source('.', 'flows/state_tester.py:state_tester').deploy(name='state-tester', work_pool_name='local-pool')"

3) Trigger specific states with parameters: - Fail:

prefect deployment run "state-tester/state-tester" -p mode=fail --watch

- Timeout:

prefect deployment run "state-tester/state-tester" -p mode=timeout --watch

- Crash:

prefect deployment run "state-tester/state-tester" -p mode=crash --watch

- Success:

prefect deployment run "state-tester/state-tester" -p mode=success --watch

- Cancellation: start a “hang” run, then cancel it

prefect deployment run "state-tester/state-tester" -p mode=hang
# note the printed flow run ID, then:
prefect flow-run cancel <YOUR_FLOW_RUN_ID>

Reference - States overview: Prefect states If you want me to tailor this for Docker/Kubernetes pools or add retries/backoff examples, just say the word.

Love it—here’s an updated “state tester” flow with optional countdown waiting and periodic countdown logs before shifting to the requested state. What’s new: -

wait_until

accepts an ISO timestamp (supports “Z” for UTC), e.g.,

2025-11-20T18:00:00Z

- Or specify a duration via

wait_seconds

/

wait_minutes

/

wait_hours

-

tick_seconds

controls the countdown log interval - Countdown is logged at regular intervals, then the mode task runs Drop this into `flows/state_tester.py`: ``` from prefect import flow, task, get_run_logger from datetime import datetime, timezone import time def _parse_wait_seconds(wait_until: str | None, wait_seconds: int, wait_minutes: int, wait_hours: int) -> int: # Priority: wait_until > cumulative wait_* duration if wait_until: ts = wait_until.strip() if ts.endswith("Z"): ts = ts[:-1] + "+00:00" try: target = datetime.fromisoformat(ts) except ValueError: # Best-effort: treat as naive UTC if parsing fails target = datetime.fromisoformat(ts.replace(" ", "T")) if target.tzinfo is None: target = target.replace(tzinfo=timezone.utc) now = datetime.now(timezone.utc) remaining = (target - now).total_seconds() return max(0, int(remaining)) # Accumulate duration fields total = 0 if wait_seconds: total += int(wait_seconds) if wait_minutes: total += int(wait_minutes) * 60 if wait_hours: total += int(wait_hours) * 3600 return max(0, int(total)) def _fmt_time(seconds: int) -> str: seconds = int(max(0, seconds)) m, s = divmod(seconds, 60) h, m = divmod(m, 60) if h > 0: return f"{h}h {m}m {s}s" if m > 0: return f"{m}m {s}s" return f"{s}s" @task def wait_with_countdown( wait_until: str | None = None, wait_seconds: int = 0, wait_minutes: int = 0, wait_hours: int = 0, tick_seconds: int = 5, ): logger = get_run_logger() remaining = _parse_wait_seconds(wait_until, wait_seconds, wait_minutes, wait_hours) if remaining <= 0: logger.info("No waiting requested; proceeding immediately.") return if wait_until: logger.info(f"Countdown until {wait_until} (UTC if 'Z' or no tz).") else: logger.info( f"Countdown for {_fmt_time(remaining)} " f"(from hours/minutes/seconds inputs)." ) tick = max(1, int(tick_seconds)) while remaining > 0: logger.info(f"T-minus {_fmt_time(remaining)}") sleep_for = min(tick, remaining) time.sleep(sleep_for) remaining -= sleep_for logger.info("Countdown complete; proceeding to requested state.") @task def succeed(): return "all good" @task def fail(): raise ValueError("intentional failure") # Will enter a TimedOut -> Failed state @task(timeout_seconds=2) def timeout_task(): time.sleep(10) # Produces a Crashed state (safer than os._exit) @task def crash(): raise SystemExit("intentional crash via SystemExit") @task def hang(seconds: int = 300, tick_seconds: int = 5): logger = get_run_logger() logger.info(f"Starting hang for {_fmt_time(seconds)} (set higher and cancel to test cancellation).") remaining = int(seconds) tick = max(1, int(tick_seconds)) while remaining > 0: logger.info(f"Hanging... {_fmt_time(remaining)} remaining") sleep_for = min(tick, remaining) time.sleep(sleep_for) remaining -= sleep_for logger.info("Hang completed (was not cancelled).") @flow(name="state-tester") def state_tester( mode: str = "success", # Waiting controls wait_until: str | None = None, wait_seconds: int = 0, wait_minutes: int = 0, wait_hours: int = 0, tick_seconds: int = 5, # Hang controls hang_seconds: int = 300, ): # 1) Optional countdown before triggering the target state wait_with_countdown( wait_until=wait_until,

wait_seconds=wait_seconds, wait_minutes=wait_minutes, wait_hours=wait_hours, tick_seconds=tick_seconds, ) # 2) Trigger the chosen state if mode == "success": return succeed() elif mode == "fail": return fail() elif mode == "timeout": return timeout_task() elif mode == "crash": return crash() elif mode == "hang": return hang(seconds=hang_seconds, tick_seconds=tick_seconds) else: raise ValueError(f"Unknown mode: {mode}")

Example runs (after deployment):
- Crash after a 1-minute countdown, logging every 10s

prefect deployment run "state-tester/state-tester" -p mode=crash -p wait_minutes=1 -p tick_seconds=10 --watch

- Timeout at an exact timestamp (UTC), logging every 5s

prefect deployment run "state-tester/state-tester" -p mode=timeout -p wait_until="2025-11-20T180000Z" -p tick_seconds=5 --watch

- Hang for 1 hour, then cancel it manually

prefect deployment run "state-tester/state-tester" -p mode=hang -p hang_seconds=3600 -p tick_seconds=15 # then: prefect flow-run cancel <YOUR_FLOW_RUN_ID> ``` Want me to package this into a ready-to-run Prefect project (with

prefect deploy

) or tailor it for Docker/Kubernetes work pools?