Prefect Flows skill for Codex

Done well means: the flow is plain, testable Python; each task is a retryable, idempotent unit with a timeout; expensive steps cache on their inputs; configuration and secrets come from blocks or environment, and the deployment runs on a named work pool that someone can see.

Before writing code

1. Read AGENTS.md and dependency files. prefect version shows client and server versions; this skill targets Prefect 3.x. Prefect 2 code mostly runs, but agents and infrastructure blocks are gone: deployments run on work pools with workers.
2. Find where flows run: Prefect Cloud or a self-hosted server (PREFECT_API_URL), and which work pool types exist (prefect work-pool ls). Listing needs API access; in a sandbox with network off, ask the user for the output.
3. Identify side effects in the script (writes, emails, API calls). They decide retry and caching policy.

Structure

from datetime import timedelta

import httpx
from prefect import flow, task, get_run_logger
from prefect.cache_policies import INPUTS, TASK_SOURCE

@task(
    retries=3,
    retry_delay_seconds=[10, 60, 300],          # backoff per attempt
    retry_jitter_factor=0.5,
    timeout_seconds=600,
    cache_policy=INPUTS + TASK_SOURCE,          # reuse result when inputs and code are unchanged
    cache_expiration=timedelta(days=1),
)
def fetch_page(day: str, page: int) -> list[dict]:
    r = httpx.get("https://api.example.com/orders", params={"day": day, "page": page}, timeout=30)
    r.raise_for_status()
    return r.json()["items"]

@task(retries=2, retry_delay_seconds=30)
def load(day: str, rows: list[dict]) -> int:
    return replace_partition("raw.orders", day, rows)   # delete+insert keeps retries safe

@flow(name="orders-daily", log_prints=True, timeout_seconds=3600)
def orders_daily(day: str, pages: int = 10) -> int:
    logger = get_run_logger()
    futures = fetch_page.map(day, range(pages))     # runs concurrently
    rows = [r for f in futures for r in f.result()]
    n = load(day, rows)
    logger.info("loaded %s rows for %s", n, day)
    return n

if __name__ == "__main__":
    orders_daily("2025-09-01")

Rules:

• Flows orchestrate; tasks do I/O. Business logic lives in plain functions that tasks call, so it can be unit-tested without Prefect.
• Pass the logical date (day) as a parameter. Never compute “yesterday” inside a task: reruns of old dates become impossible.
• Resolve every future (.result() or wait(futures)); unresolved futures can end the flow before the work finishes.

Retries, caching and timeouts

Situation	Setting
Flaky network or rate limits	retries + list of retry_delay_seconds, plus retry_jitter_factor
Retry only on specific errors	retry_condition_fn= returning True for retryable exceptions
Expensive deterministic step	cache_policy=INPUTS (add TASK_SOURCE to invalidate on code change)
Step must always run	cache_policy=NO_CACHE
Hang protection	timeout_seconds on tasks and on the flow
Results must survive across runs / processes	persist_result=True and a result storage block
Outward-facing action (email, payment, publish)	retries=0, separate task, explicit user confirmation before enabling in a deployment

Caching needs stored results; with caching on, Prefect persists task results by default to local storage (~/.prefect/storage). On ephemeral workers configure remote result storage or caches will never hit.

Configuration and secrets

from prefect.blocks.system import Secret
from prefect.variables import Variable

api_key = Secret.load("orders-api-key").get()
batch = Variable.get("orders_batch_size", default=500)

Never hard-code secrets or put them in flow parameters (parameters are visible in the UI). Create blocks with prefect block create or the UI; reference by name.

Concurrency control

• Limit parallel tasks hitting one API: global concurrency limit with from prefect.concurrency.sync import concurrency then with concurrency("orders-api", occupy=1): ..., after creating it with prefect gcl create orders-api --limit 5.
• Tag-based limits on tasks: @task(tags=["warehouse"]) and prefect concurrency-limit create warehouse 4.
• Limit simultaneous flow runs per deployment with concurrency_limit on the deployment.

Deployments

Need	Use
Run on this machine on a schedule, simplest	orders_daily.serve(name="daily", cron="0 3 * * *") (long-running process)
Run on managed infrastructure (Docker, Kubernetes, ECS, serverless)	Work pool + worker, flow.deploy(...) or prefect deploy with prefect.yaml
Code pulled from git at run time	flow.from_source(source="https://github.com/org/repo.git", entrypoint="flows/orders.py:orders_daily")

# prefect.yaml
deployments:
  - name: orders-daily
    entrypoint: flows/orders.py:orders_daily
    work_pool:
      name: k8s-default
    parameters:
      pages: 10
    schedules:
      - cron: "0 3 * * *"
        timezone: UTC

prefect deploy --name orders-daily          # registers; needs API access
prefect worker start --pool k8s-default     # a worker must poll the pool
prefect deployment run 'orders-daily/orders-daily' --param day=2025-09-01

Registering or changing a production deployment and triggering runs on shared infrastructure require the user’s explicit go-ahead.

Testing

import pytest
from prefect.testing.utilities import prefect_test_harness

@pytest.fixture(autouse=True, scope="session")
def prefect_backend():
    with prefect_test_harness():
        yield

def test_orders_daily(monkeypatch):
    monkeypatch.setattr("flows.orders.replace_partition", lambda *a: len(a[2]))
    assert orders_daily.fn("2025-09-01", pages=1) >= 0   # .fn calls the function without Prefect

def test_flow_run():
    assert orders_daily("2025-09-01", pages=1) >= 0     # full Prefect run on a temp backend

Gotchas

• Scheduled runs stay “Late” or “Pending” forever when no worker polls the work pool, or the worker type does not match the pool type.
• Large task return values are serialized for results and caching; return paths or counts, not DataFrames with millions of rows.
• Cache keys computed from unhashable or huge inputs fail or are slow; pass identifiers.
• log_prints=True captures print; library logs need PREFECT_LOGGING_EXTRA_LOGGERS.
• Flow parameters are validated with type hints; a day: date parameter rejects a malformed string at submit time, which beats failing deep inside a task.

Verification and report

• Run the flow locally once with a real parameter and check the run in the UI or prefect flow-run ls --limit 5.
• Force a failure (raise in a task) to confirm retries and alerting behave as expected.
• Rerun the same day and confirm row counts in the target do not change.

**Flow:** orders-daily (Prefect 3.4), deployment on work pool k8s-default, 03:00 UTC
**Tasks:** fetch_page x10 (retries 3, cached on inputs 1 day), load (delete+insert)
**Limits:** global concurrency orders-api = 5
**Checks:** local run 2025-09-01 ok (14,220 rows); rerun idempotent; forced failure retried 3x
**Awaiting approval:** prefect deploy to production workspace