Short answer: Don’t rely on the ephemeral API for tests. Either start a local Prefect server and point your tests at it, or increase the ephemeral startup timeout. Assuming you’re on Prefect 3.x, here are reliable ways to avoid the timeout: Option A — Run a local server (recommended for testing) - Start a server in a separate terminal:

prefect server start

- Point your flow/tests to it:

export PREFECT_API_URL="<http://127.0.0.1:4200/api>"
pytest  # or python my_flow.py

Docs: Server concepts, Server CLI Option B — Keep ephemeral mode but increase its startup timeout If you want to stick with ephemeral mode during local testing:

export PREFECT_SERVER_EPHEMERAL_STARTUP_TIMEOUT_SECONDS=60
pytest  # or python my_flow.py

You can also put this in `prefect.toml`:

[server.ephemeral]
startup_timeout_seconds = 60

Option C — Use Prefect’s test harness (pytest) The harness sets up and tears down a clean test environment:

import pytest
from prefect.testing.utilities import prefect_test_harness

@prefect_test_harness
def test_my_flow():
    # run your flow here
    pass

This helps avoid flakiness around API state. Docs: Settings reference Option D — If DB startup is the bottleneck, give it more time (or use Postgres)

export PREFECT_API_DATABASE_CONNECTION_TIMEOUT=15
export PREFECT_API_DATABASE_TIMEOUT=30
# Optional: switch to Postgres for stability under load
export PREFECT_API_DATABASE_CONNECTION_URL="<postgresql+asyncpg://user:pass@localhost:5432/prefect>"

If you truly want to avoid ephemeral entirely - Disable it and use a real server:

export PREFECT_SERVER_ALLOW_EPHEMERAL_MODE=false
export PREFECT_API_URL="<http://127.0.0.1:4200/api>"

If you disable ephemeral but do not set an API URL, operations that require the API will fail. Extra debugging tips - Enable more logs:

export PREFECT_LOGGING_LEVEL=DEBUG

- Check for port conflicts and slow/locked local DB files. Quick pick - Fastest fix: set

PREFECT_SERVER_EPHEMERAL_STARTUP_TIMEOUT_SECONDS=60

- Most stable for tests: run

prefect server start

and set

PREFECT_API_URL

Useful links: - Prefect Server concepts - Server CLI - Settings reference If you can share how you’re running tests (pytest, notebooks, plain Python) and whether you want to avoid running a server entirely, I can tailor the exact setup/commands.

Great question. You have two good options to reuse a single API server across multiple test runs/debug sessions so you don’t get a new random port each time: Option 1 — Use the built-in session-scoped server fixtures (recommended) - Prefect provides a session-wide server you can reuse across all tests: -

hosted_api_server

(session-scoped) starts one server and yields its API URL -

use_hosted_api_server

(autouse) sets

PREFECT_API_URL

for all tests to that URL - Setup in tests/conftest.py:

# Just import to enable them
from prefect.testing.fixtures import hosted_api_server, use_hosted_api_server

- Now: - One server per pytest session - Stable API URL for the whole session - No per-test port changes - You can also inject the URL if you need it:

def test_uses_session_server(hosted_api_server: str):
    assert hosted_api_server.startswith("http://")

Option 2 — Make prefect_test_harness session-scoped and read the URL from settings - If you prefer

prefect_test_harness

, wrap it as a session-scoped autouse fixture so it only starts once, then read the URL from settings:

# tests/conftest.py
import pytest
from prefect.testing.utilities import prefect_test_harness
from prefect.settings import PREFECT_API_URL

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

@pytest.fixture(scope="session")
def session_api_url() -> str:
    return PREFECT_API_URL.value()

- Now all tests share the same ephemeral server and you can use

session_api_url

anywhere without chasing changing ports. Notes and tips - If you need a specific fixed port, you can use the hosted server approach and bind to a chosen port (via a custom fixture that instantiates

SubprocessASGIServer(port=...)

) and set

PREFECT_API_URL

using

temporary_settings

. But the built-in fixtures handle port allocation well and are simpler. - For faster runs and fewer surprises, Option 1 (hosted_api_server + use_hosted_api_server) is the most streamlined pattern. - If you’re debugging outside pytest (e.g., VSCode debug on a single test file), running one server once in a terminal and setting

PREFECT_API_URL

is also convenient:

prefect server start
export PREFECT_API_URL="<http://127.0.0.1:4200/api>"
pytest -k my_test  # or run your debugger

If you share how you run/debug (pytest CLI, VSCode pytest UI, notebooks), I can tailor the exact fixture or startup pattern for your workflow.

Perfect—here are two solid ways to combine pytest with the VSCode debugger for step-by-step debugging, without a new random API port on every debug run. Approach 1 — Run a local Prefect server once, point VSCode at it (most stable) 1) In a separate terminal, start a server on a known port:

prefect server start --port 4200

2) In VSCode, add a pytest debug config in .vscode/launch.json:

{
  "name": "pytest: debug current test",
  "type": "python",
  "request": "launch",
  "module": "pytest",
  "args": [
    "-k",
    "test_file::test_case",
    "-s",
    "-vv"
  ],
  "env": {
    "PREFECT_API_URL": "<http://127.0.0.1:4200/api>",
    "PREFECT_LOGGING_LEVEL": "DEBUG"
  },
  "envFile": "${workspaceFolder}/.env",
  "console": "integratedTerminal",
  "justMyCode": false
}

- Put breakpoints in your flow/task code and hit F5. - This avoids ephemeral servers entirely, so your debug runs always hit the same API URL/port. Approach 2 — Keep it in pytest, but fix the port via a session fixture If you want pytest to spin up the server for you (per session) but keep a stable port between runs, define a custom session-scoped fixture that starts Prefect’s subprocess server on a fixed port and sets `PREFECT_API_URL`: tests/conftest.py

import pytest
from prefect.server.api.server import SubprocessASGIServer
from prefect.settings import PREFECT_API_URL, temporary_settings

@pytest.fixture(scope="session", autouse=True)
def session_server():
    # pick a fixed port you like
    server = SubprocessASGIServer(port=4201)
    server.start()
    try:
        with temporary_settings({PREFECT_API_URL: server.api_url}):
            yield
    finally:
        server.stop()

Then your VSCode pytest launch config can be simple:

{
  "name": "pytest: debug current test (session server)",
  "type": "python",
  "request": "launch",
  "module": "pytest",
  "args": [
    "-k",
    "test_file::test_case",
    "-s",
    "-vv"
  ],
  "console": "integratedTerminal",
  "justMyCode": false
}

Notes: - Each debug run is a new Python process, so pytest will start/stop the server each session—but it will always use port 4201, so your breakpoints and any external tools can consistently target that URL. - If server startup can be slow on your machine, add to your .env:

PREFECT_SERVER_EPHEMERAL_STARTUP_TIMEOUT_SECONDS=60

VSCode tips for smooth step debugging with pytest - Use the integrated terminal and

-s

so logs/prints are visible. - Set

justMyCode: false

to step through library code if needed. - Prefer breakpoints in your flow and task functions (they run in-process during local tests). - If you see intermittent startup delays, bump:

PREFECT_SERVER_EPHEMERAL_STARTUP_TIMEOUT_SECONDS=60

Docs: - Server CLI - Settings reference If you tell me whether you prefer a persistent server (Approach 1) or an all-in-pytest setup (Approach 2), I can tailor the exact launch.json and conftest.py to your project layout and test selection pattern.

In Prefect 3 there isn’t a first-class “Skipped” state type. The recommended patterns are: - Don’t call the task when the condition is false (most Pythonic) - Or, return a Completed state with a custom name/message so the UI shows “Skipped” while remaining green (Completed) Examples 1) Don’t call the task if the condition isn’t met

from prefect import flow, task

@task
def heavy_step(x):
    return x * 2

@flow
def pipeline(run_heavy: bool):
    if run_heavy:
        result = heavy_step(21)
        return result
    # Skipped by not calling the task at all
    return None

2) Mark a task as “Skipped” in the UI (Completed with custom name/message)

from prefect import flow, task
from prefect.states import Completed

@task
def maybe_step(should_run: bool):
    if not should_run:
        # Shows as “Skipped” in the UI but remains a Completed state
        return Completed(name="Skipped", message="Condition not met")
    return "did work"

@flow
def pipeline():
    return maybe_step(False)

3) Make downstream decisions using return_state=True

from prefect import flow, task
from prefect.states import Completed

@task
def maybe_step(should_run: bool):
    if not should_run:
        return Completed(name="Skipped", message="Condition not met")
    return 42

@flow
def pipeline():
    state = maybe_step(False, return_state=True)
    if state.name == "Skipped":
        # gate downstream logic based on the state
        return "downstream skipped"
    value = state.result()
    return value

Notes - The UI will display your custom state name (“Skipped”) and message, but the state type is still Completed (green). - If your goal is to avoid re-running work after a retry, prefer either: - Conditional calls (don’t call the task), or - Caching/persisted results patterns so Prefect reuses prior results instead of recomputing. Docs: - States overview: States concept If you share your specific use case (e.g., skipping a step based on prior outputs, or marking a whole flow as skipped), I can tailor the pattern.

Best practice: create two SlackWebhook blocks (test and prod) and select the right one at runtime via environment, variables, or deployment parameters. Pick the pattern that fits your workflow: Option 1 — Two blocks + environment variable (simple and explicit) - Create blocks: -

slack-webhook-test

→ your test channel webhook URL -

slack-webhook-prod

→ your prod channel webhook URL - In code:

import os
from prefect import flow
from prefect.blocks.notifications import SlackWebhook

@flow
def my_flow():
    env = os.getenv("PREFECT_ENV", "test")  # set to "prod" in prod runs
    block_name = f"slack-webhook-{env}"
    slack = SlackWebhook.load(block_name)
    slack.notify(subject=f"[{env}] Flow finished", body="All done")

- In prod, set

PREFECT_ENV=prod

(profile, .env, deployment env vars). Option 2 — Use Prefect Variables (clean config management) - Store the block names (or URLs) in Variables: -

slack_webhook_test = slack-webhook-test

-

slack_webhook_prod = slack-webhook-prod

- In code:

import os
from prefect import flow
from prefect.variables import Variable
from prefect.blocks.notifications import SlackWebhook

@flow
async def my_flow():
    env = os.getenv("PREFECT_ENV", "test")
    block_name = await Variable.get(f"slack_webhook_{env}", default="slack-webhook-test")
    slack = SlackWebhook.load(block_name)
    await slack.notify(subject=f"[{env}] Done", body="Success")

- Benefits: switch channels without code changes; safe for multi-env deployments. Option 3 — Pass the channel as a flow parameter (great for deployments)

from prefect import flow
from prefect.blocks.notifications import SlackWebhook

@flow
def my_flow(notification_block: str = "slack-webhook-test"):
    slack = SlackWebhook.load(notification_block)
    slack.notify(subject="Flow finished", body="All done")

- In your prod deployment, set parameter

notification_block="slack-webhook-prod"

. Option 4 — Use Automations for event-driven alerts (centralized) - Create two automations (test/prod) with SendNotification actions pointing to each block. - Scope them by workspace or by naming convention and enable per environment. - Helpful when you want UI-driven, event-based notifications (e.g., on flow failed/completed) rather than in-flow code. - Docs: Server concepts and Automations UI Additional tips - Never hardcode webhook URLs in code. Use SlackWebhook blocks (URL is SecretStr) or Variables. - Prefix subjects to make channels obviously separate, e.g., “[TEST] …” vs “🚨 [PROD] …”. - If you run both test and prod from the same codebase, prefer Option 2 (Variables) +

PREFECT_ENV

. - For CI, set

PREFECT_ENV

or flow parameters per pipeline/job. If you share how you deploy (local, workers, CI/CD) and whether you want in-flow or automation-based notifications, I can suggest the exact wiring and example YAML/CLI you need.

Two good patterns, depending on whether you want code-level hooks or centralized event rules: Pattern A — Reuse one code handler across multiple states (flows/tasks) - Write one common notifier and small adapters for flow/task hooks. - Pass the same adapter to multiple on_* hook slots. Example:

import os
from prefect import flow, task
from prefect.blocks.notifications import SlackWebhook

async def notify_common(kind: str, name: str, run_id: str, state):
    env = os.getenv("PREFECT_ENV", "test")
    slack = SlackWebhook.load(f"slack-webhook-{env}")

    # Map multiple states to a single handler behavior
    severity_by_type = {
        "COMPLETED": "success",
        "FAILED": "failure",
        "CRASHED": "failure",
        "CANCELLED": "warning",
        "RUNNING": "info",
    }
    state_type = getattr(getattr(state, "type", None), "name", str(getattr(state, "type", "UNKNOWN")))
    severity = severity_by_type.get(state_type, "info")

    subject = f"[{env.upper()}] {kind.title()} {severity}: {name}"
    body = f"State: {state.name}\nMessage: {state.message or ''}\nRun ID: {run_id}"
    await slack.notify(subject=subject, body=body)

# Thin adapters for flow and task hooks
async def flow_notify(flow, flow_run, state):
    await notify_common("flow", flow.name, str(flow_run.id), state)

async def task_notify(task, task_run, state):
    await notify_common("task", task.name, str(task_run.id), state)

@flow(
    on_running=[flow_notify],
    on_completion=[flow_notify],
    on_failure=[flow_notify],
    on_cancellation=[flow_notify],
    on_crashed=[flow_notify],
)
def my_flow():
    t()

@task(
    on_completion=[task_notify],
    on_failure=[task_notify],
)
def t():
    return "ok"

- You reuse the same handler for many states by listing it in multiple on_* parameters and branching inside by

state.type

or

state.name

. Pattern B — One automation that reacts to multiple states (centralized, no code changes) - Use a single Automation with an event trigger that expects multiple events (Completed, Failed, Crashed, Cancelled, etc.) and a single SendNotification action pointing to your SlackWebhook block. - This applies to all flows matching your trigger filters and keeps logic out of code. - In YAML (conceptual):

automations:
  - name: "Notify Slack for multiple states"
    enabled: true
    trigger:
      type: event
      posture: Reactive
      expect:
        - prefect.flow-run.Completed
        - prefect.flow-run.Failed
        - prefect.flow-run.Crashed
        - prefect.flow-run.Cancelled
      threshold: 1
    actions:
      - type: send-notification
        block_document_id: "<your-slack-block-id>"
        subject: "[{{ event.resource.environment|default('TEST') }}] {{ event.resource.state_name }}: {{ event.resource.name }}"
        body: |
          Flow: {{ event.resource.name }}
          State: {{ event.resource.state_name }}
          Message: {{ event.resource.state_message }}

When to pick which: - Few flows and you want per-flow customization: Pattern A (hooks in code). - Many flows or you want uniform behavior and zero code changes: Pattern B (automations). If you tell me whether you want flow-level hooks or central automations, I can provide a ready-to-run snippet (including how to get your Slack block ID for the automation).

Short answer: smaller isn’t always better. Very fine‑grained tasks increase parallelism, but they also increase orchestration overhead (task run creation, state updates, logs/events) and interprocess serialization. Prefer “right‑sized” tasks. Practical guidance - Aim for coarser tasks that each do a meaningful unit of work (seconds, not milliseconds). Ultra‑short tasks (<100–200 ms) usually cost more in overhead than they save in parallelism. - Batch small operations. Map over batches/chunks instead of individual records. - Pick the right task runner: - CPU‑bound:

ProcessPoolTaskRunner

- I/O‑bound:

ThreadPoolTaskRunner

- Keep results small. Don’t pass large dataframes/arrays through task results; write to storage and pass references. - Reduce metadata volume: - Keep

persist_result=False

unless you need persistence/caching - Limit logging noise (

log_prints=False

, fewer

<http://logger.info|logger.info>

calls, set

PREFECT_LOGGING_LEVEL=WARNING

) - Avoid excessive task retries on very short tasks Chunking pattern (good parallelism, limited overhead)

from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner

def chunk(seq, size):
    for i in range(0, len(seq), size):
        yield seq[i:i+size]

@task(persist_result=False, log_prints=False)
def process_batch(batch):
    # Do multiple small ops inside one task to amortize overhead
    return [expensive_compute(x) for x in batch]

@flow(task_runner=ProcessPoolTaskRunner(max_workers=8))
def pipeline(items, batch_size=1000):
    futures = [process_batch.submit(b) for b in chunk(items, batch_size)]
    results = [f.result() for f in futures]
    return [y for r in results for y in r]

When to split vs merge - Split into separate tasks if: - You want independent retries/failure isolation - You need concurrency on truly heavy, independent steps - You want distinct observability for that step - Merge into one task if: - Steps are tiny, tightly coupled, or must share process memory - Data handoff would serialize large objects - You’re seeing heavy API/log traffic from too many task runs References - Task runners: Task runners - Results and storage: Results - Logging settings: Settings reference If you can share your task durations and throughput targets (e.g., items/sec), I can suggest a batch size and runner config tailored to your workload.

Yes. You can keep Prefect orchestration coarse (few tasks) and do the parallelism “inside” a single task. Common patterns: - Async I/O inside one task (best for HTTP/DB calls)

from prefect import flow, task
import asyncio, aiohttp

@task
async def fetch_all(urls, limit=50):
    sem = asyncio.Semaphore(limit)
    async with aiohttp.ClientSession() as session:
        async def fetch(u):
            async with sem:
                async with session.get(u) as r:
                    return await r.text()
        return await asyncio.gather(*(fetch(u) for u in urls))

@flow
async def pipeline(urls):
    pages = await fetch_all(urls, limit=100)
    return len(pages)

- Threads inside one task (good for I/O-bound workloads)

from prefect import flow, task
from concurrent.futures import ThreadPoolExecutor

def io_work(x):
    # do network/disk I/O
    return x

@task
def run_threaded(items, max_workers=32):
    with ThreadPoolExecutor(max_workers=max_workers) as ex:
        return list(ex.map(io_work, items))

@flow
def pipeline(items):
    return run_threaded(items)

- Processes inside one task (CPU-bound)

from prefect import flow, task
from concurrent.futures import ProcessPoolExecutor

def cpu_work(x):
    # heavy CPU
    return x * x

def cpu_batch(batch):
    return [cpu_work(x) for x in batch]

@task
def run_multiprocess(items, max_workers=None, batch_size=1000):
    batches = [items[i:i+batch_size] for i in range(0, len(items), batch_size)]
    with ProcessPoolExecutor(max_workers=max_workers) as ex:
        results = ex.map(cpu_batch, batches)
    out = []
    for r in results:
        out.extend(r)
    return out

@flow
def pipeline(items):
    return run_multiprocess(items, max_workers=8, batch_size=5000)

- External parallel engines inside one task - Use Dask, Ray, Spark, or a database to parallelize, while Prefect treats it as a single task step (driver submits work; cluster executes). Notes and caveats - Prefect only “sees” one task run this way: one set of states/logs/retries. Inner work is opaque to Prefect. - Cancellation/retries happen at the task boundary; add your own handling inside if needed. - Avoid nested process pools: if your flow uses

ProcessPoolTaskRunner

, don’t also spawn a process pool inside tasks (oversubscription). Prefer one tier: either internal processes with a default/sync task runner, or multiple Prefect tasks with a process pool runner. - Keep inter-task data small; large payloads should be written to storage and referenced. This approach gives you parallelism without many tiny tasks, minimizing server metadata while still using all cores/network efficiently. If you share your workload type (CPU vs I/O), typical item count, and average work per item, I can recommend exact pool sizes and batch sizes.

You can get the current run’s URLs from Prefect’s runtime variables. - Flow run URLs:

from prefect.runtime import flow_run

# Link to this flow run in the UI (Cloud or Server)
flow_ui = flow_run.ui_url()
# API URL for this flow run resource
flow_api = str(flow_run.api_url())

- Task run URLs:

from prefect.runtime import task_run

# Link to this task run in the UI
task_ui = task_run.ui_url()
# API URL for this task run resource
task_api = str(task_run.api_url())

Notes - These calls only work while running inside a Prefect flow/task; otherwise you’ll get a MissingContextError. - If you just want the base server URLs (not run-specific), read settings:

from prefect.settings import PREFECT_API_URL, PREFECT_UI_URL

api_base = PREFECT_API_URL.value()
ui_base = PREFECT_UI_URL.value()

Docs: - Runtime variables: Runtime variables - Settings: Settings reference

Short answer: Yes, but how you enforce it depends on your execution backend. Prefect 3 itself doesn’t have a global “memory cap” setting; you set limits at the infrastructure level. Pick your backend: - Docker work pool (easy, hard limits) - Set memory on the job via work pool base job template or per-deployment job variables: - mem_limit: “512m”, “1g”, etc. (hard cap) - memswap_limit: total memory + swap

# In your Docker work pool base job template or job_variables
{
  "job_configuration": {
    "image": "prefecthq/prefect:3-latest",
    "mem_limit": "512m",
    "memswap_limit": "1g",
    "container_create_kwargs": {
      "cpu_shares": 1024,         # optional CPU controls
      "cpuset_cpus": "0,1"
    }
  }
}

- If a container exceeds its memory limit, Docker OOM-kills it (exit code 137). Prefect will mark the run failed/crashed accordingly. - Kubernetes work pool (best for production) - Add limits in the job manifest’s container resources:

spec:
  template:
    spec:
      containers:
      - name: prefect-job
        resources:
          requests:
            memory: "256Mi"
            cpu: "100m"
          limits:
            memory: "512Mi"
            cpu: "500m"

- The kubelet enforces the memory limit; OOMKilled if exceeded. You can template these in your work pool’s base_job_template and override via deployment job_variables. - Local process work pool (no native memory limit) - There’s no built-in memory cap for plain processes. - Options: - Run flows inside Docker and use the Docker limits (recommended). - Use OS-level limits (e.g., ulimit on Linux) before starting the worker: - ulimit -v <bytes> to cap virtual memory - Custom wrapper that sets resource limits via Python’s resource module (Unix-only) before launching the run. - Caveat: OS-level RSS limits are advisory on Linux; RLIMIT_AS (virtual memory) is the only strict cap. Monitoring and behavior - Docker: OOM kill → exit 137. Check container status or Prefect run logs; Prefect will surface the failure/crash. - Kubernetes: Pod terminated with reason OOMKilled; Prefect run reflects infrastructure failure. - Consider adding retries and alerts (e.g., Slack) for OOM events. Recommendation - If you need reliable per-run caps, use Docker or Kubernetes work pools and set limits in the job template. That gives you enforceable kills when memory is exceeded. - For local-only development, prefer Docker worker locally to get the same constraints you’ll have in prod. If you share your current work pool type (process, docker, kubernetes) I can provide an exact base_job_template/job_variables snippet you can drop in.

Got it — with the local (process) worker there isn’t a built-in Prefect setting to cap RAM per flow run. You’ll need to enforce limits at the OS/container layer. Here are practical options: Option A — Limit the whole worker (Linux, recommended) - Start the worker in a memory-limited cgroup so every flow-run it spawns is constrained. - With systemd:

# Limits the worker and all children (flow runs)
systemd-run --scope -p MemoryMax=1G -p MemorySwapMax=0 \
  prefect worker start --type process --pool my-process-pool

- This cleanly OOM-kills flow runs that exceed the cap; Prefect will mark the run failed/crashed. Option B — Per-run limit via ulimit wrapper (Unix) - Wrap the flow-run execution with a script that sets RLIMIT_AS (virtual memory) before exec’ing Prefect’s run command. - Wrapper:

# run_with_mem_limit.sh
#!/usr/bin/env bash
# Usage: run_with_mem_limit.sh <flow_run_id> <mem_mb>
ulimit -v $(( $2 * 1024 ))   # virtual memory in KB
exec prefect flow-run execute "$1"

- Then override the Process work pool “command” to call the wrapper (so each run is capped). In your process work pool’s base job template (or deployment job_variables), set command to something like:

/path/to/run_with_mem_limit.sh {{ flow_run_id }} 1024

Note: the process worker passes the flow run id to the command; if your template variable name differs, I can help wire it up. Option C — Python resource limits wrapper (Unix) - Same idea as B, but using a small Python entrypoint that calls

resource.setrlimit(resource.RLIMIT_AS, ...)

then execs

prefect flow-run execute <id>

. Useful if you prefer Python over shell. Option D — If you can use containers locally - Switch to a Docker work pool (even on your dev box) and set:

"mem_limit": "1g",
"memswap_limit": "1g"

- This is the most robust per-run cap; Docker will OOM-kill exceeding runs (exit 137). Caveats - macOS/Windows don’t have Linux cgroups; Option A requires Linux. On macOS, the most reliable path is Docker. - RLIMIT_RSS is advisory on Linux; RLIMIT_AS (virtual memory) is the strict cap for ulimit/resource-based approaches. - Avoid nested process pools (e.g., using ProcessPoolTaskRunner inside a process-limited worker) to prevent CPU oversubscription. If you tell me your OS (Linux/macOS/Windows) and whether you prefer per-run or whole-worker limits, I’ll give you an exact command/template you can copy/paste.

Great question! Here’s the quick breakdown (assuming Prefect 3.x): - Work pool - A named, typed “bridge” in the Prefect API that represents where and how your flows should run (e.g., process, Docker, Kubernetes, ECS). - Stores configuration like base job templates, concurrency limits, and work queues. - Deployments target a work pool. - Worker - A long-running process that polls a work pool’s queues for scheduled flow runs and launches them on the target infrastructure. - Must match the pool’s type (a Kubernetes worker polls a Kubernetes pool). - You can run multiple workers against the same pool for scale/HA. - Work queues - Still exist in Prefect 3.x and live under a work pool. - Useful for prioritization and per-queue concurrency. - A worker can poll all queues in the pool or specific queues you choose. Is it good to cron a restart of the worker? - Generally, no. Prefer a supervisor that auto-restarts on failure and uses health checks: - Linux: systemd with Restart=always - Docker: --restart=always (and a container HEALTHCHECK) - Kubernetes: a Deployment with liveness/readiness probes - If you do want periodic restarts (e.g., to mitigate a temporary memory leak), it’s okay—but make sure you: - Run multiple workers for HA so polling doesn’t pause during restart - Send a graceful signal (SIGTERM) so the worker stops polling and finishes in-flight submissions cleanly - Use the worker’s healthcheck server to decide when to restart Recommended patterns - Start workers with a health endpoint:

prefect worker start --pool <your-pool> --with-healthcheck

- Use a supervisor instead of cron: - systemd unit: set

Restart=always

and

RestartSec=10

- Docker:

--restart=always

and a HEALTHCHECK hitting

<http://localhost:8080/health>

- Kubernetes: Deployment with liveness/readiness probes to the same health endpoint - Optional tuning: -

--limit

to cap concurrent runs per worker - Work queues for priority routing - Multiple workers per pool for redundancy Docs - Workers: Workers (concepts) - Work pools: Work pools (concepts) If you can share where your worker runs (bare VM, Docker, or Kubernetes), I can give you a minimal, copy-paste setup that matches your environment.

Short answer: Yes, 5,000 task futures can be acceptable in a single flow, but not as 5,000 truly-concurrent tasks on one worker with the default runner. It’s fine if you batch them and/or limit concurrency. What to consider - No hard limit: Prefect 3.x doesn’t enforce a hard cap on futures. Practical limits come from your task runner, machine memory/CPU, and how many you run concurrently. - Concurrency matters: - ThreadPoolTaskRunner: good for I/O-bound work, but don’t set max_workers in the thousands. Start with 32–256 depending on your machine and workload. - ProcessPoolTaskRunner: for CPU-bound work; keep max_workers ≈ CPU cores, not thousands. - For thousands of concurrent tasks: use a distributed runner (Dask or Ray) or multiple flow runs. - Memory and UI: Holding thousands of unresolved futures and their results in one process can use a lot of memory, and a single flow with 5k–10k+ tasks can feel heavy in the UI. Recommended patterns - Batch/Chunk submissions:

from prefect import flow, task
  from prefect.futures import wait

  @task
  def do_work(x): 
      ...

  @flow
  def big_flow(items):
      batch = 500
      for i in range(0, len(items), batch):
          futs = [do_work.submit(x) for x in items[i:i+batch]]
          wait(futs)  # or [f.result() for f in futs] to release memory as you go

- Use concurrency controls (tags or named concurrency limits) to throttle shared resources. - For large parallelism, use a distributed task runner:

from prefect_dask.task_runners import DaskTaskRunner
  from prefect import flow

  @flow(task_runner=DaskTaskRunner())
  def big_flow(...):
      ...

Rule of thumb - 5,000 total tasks: OK if batched and resolved progressively. - 5,000 concurrent tasks on one machine: Not recommended. Use distributed runners or multiple workers/pools. Helpful docs - Task runners: Task runners (concepts) - Concurrency: Concurrency (concepts) If you share whether your tasks are CPU or I/O bound and which runner you’re using, I can suggest batch sizes and runner settings tailored to your setup.

Yes—Prefect 3.x gives you several ways to control or throttle logs that tasks/flows send to the server via the client. For thousands of tasks, you can reduce load significantly: Most impactful switches - Turn off to-API logs entirely (keeps events/states unaffected):

export PREFECT_LOGGING_TO_API_ENABLED=False

- Or raise the level so only warnings/errors go to the API:

export PREFECT_LOGGING_LEVEL=WARNING

Per-log, in code - Skip sending specific log records to the API while still logging locally:

from prefect.logging import get_run_logger

  @task
  def work(x):
      logger = get_run_logger()
      <http://logger.info|logger.info>("important")  # goes to API
      logger.debug("spam", extra={"send_to_api": False})  # local only

Batching and size tuning - Prefect batches logs before sending; you can tune responsiveness vs overhead:

export PREFECT_LOGGING_TO_API_BATCH_SIZE=5000000     # bytes
  export PREFECT_LOGGING_TO_API_BATCH_INTERVAL=2.0     # seconds
  export PREFECT_LOGGING_TO_API_MAX_LOG_SIZE=500000    # chars (Cloud caps at ~25KB/log)

Suppress in hot loops - Temporarily disable run logger in tight sections:

from prefect.logging import disable_run_logger

  with disable_run_logger():
      # high-throughput section without API logs
      ...

Target which loggers go to API/console - Customize ~/.prefect/logging.yml (or point with PREFECT_LOGGING_SETTINGS_PATH) to: - Send only WARNING+ to API, DEBUG/INFO to a local file - Include/exclude extra third-party loggers (

PREFECT_LOGGING_EXTRA_LOGGERS=dask,...

) - Example idea: have

prefect.task_runs

-> API (WARNING), and also a rotating file handler for full DEBUG locally. Notes - Events are separate from logs: disabling API logs does not stop events/state transitions in the UI. - For very large task counts, prefer raising levels or disabling API logs, and keep rich logs locally (e.g., JSON to file) if you need them for audits. If you share your current volume and whether you need task-level INFO logs in the UI, I can suggest a concrete config (env vars + logging.yml) tailored to your workload.

Yes. You can keep Prefect’s retry behavior, but on the final attempt return a value instead of failing by catching the exception and checking the current attempt number. Pattern 1: Handle inside the task (recommended for simple cases)

from prefect import task, flow, get_run_logger
from prefect.runtime import task_run

RETRIES = 3  # total tries = 1 initial + RETRIES

@task(retries=RETRIES, retry_delay_seconds=10)
def might_fail(x):
    logger = get_run_logger()
    attempt = task_run.run_count()  # 1-based: 1, 2, ..., RETRIES+1

    try:
        # your risky logic
        result = do_something(x)
        return {"status": "ok", "data": result}
    except Exception as e:
        if attempt <= RETRIES:
            logger.warning(f"Attempt {attempt} failed; retrying: {e}")
            raise  # trigger Prefect retry
        else:
            logger.error(f"Final attempt failed; returning error status: {e}")
            return {"status": "error", "message": str(e)}

@flow
def main(x):
    return might_fail(x)

Explanation: - For attempts 1..RETRIES, re-raise to let Prefect retry. - On the final attempt (RETRIES+1), return your error payload. The task will be Completed (not Failed), so make sure downstream logic checks your returned status. Pattern 2: Let the task fail, then handle downstream with allow_failure - This keeps the task’s true failure state, but your downstream task can translate that to a structured status.

from prefect import task, flow, allow_failure

@task
def risky(x):
    return do_something(x)  # may raise

@task
def to_status(state):
    # `state` is a State object instead of raising UpstreamTaskError
    if state.is_failed():
        # get the exception text without raising
        err = state.result(raise_on_failure=False)
        return {"status": "error", "message": str(err)}
    else:
        return {"status": "ok", "data": state.result()}

@flow
def main(x):
    with allow_failure():
        s = risky.submit(x)  # returns the task's State to downstream
    return to_status(s)

Notes - Pattern 1 makes the task appear successful on the final attempt; Pattern 2 preserves the failed state (useful for alerting) and translates it later. - If you need exponential backoff, you can implement it yourself or use Prefect helpers. - Be explicit with logging so it’s clear a “Completed” task may still be returning an error status. Docs - Tasks and retries: Tasks (retries) - Runtime (getting run counts, etc.): Runtime context

Got it—if you’re looking for “return_state” on a future, here’s how it works in Prefect 3.x: - There is no return_state argument on PrefectFuture.result(). The return_state option exists when you call a task/flow function directly (not via .submit):

state = my_task(return_state=True)   # returns a State
  result = my_task(return_state=False) # default, returns the value

Working with a future from .submit(...) - Get the current State (non-blocking):

fut = my_task.submit(...)
  st = fut.state   # property, returns a State immediately (may not be final yet)

- Get the result (and control failure raising):

value_or_exc = fut.result(raise_on_failure=False)
  if isinstance(value_or_exc, Exception):
      # task failed; you have the exception object
      ...

- Wait for completion, then inspect both:

fut.wait()               # optional; result() also waits
  final_state = fut.state
  value = fut.result()     # raises if failed unless raise_on_failure=False

Extracting exceptions from State - You can pull the exception from a failed State without raising:

from prefect.states import get_state_exception

  st = fut.state
  if st.is_failed():
      exc = get_state_exception(st)

Passing failed futures downstream without raising - Use allow_failure typing to accept failed upstream futures and handle them:

from prefect.utilities.annotations import allow_failure
  from prefect import task, flow

  @task
  def upstream():
      ...

  @task
  def handle(up: allow_failure[int]):
      if isinstance(up, Exception):
          return {"status": "error", "message": str(up)}
      return {"status": "ok", "data": up}

  @flow
  def pipeline():
      f = upstream.submit()
      return handle.submit(f).result()

If you share what you want to do with the State (e.g., treat failures as structured results, implement custom retries, or route based on State types), I can sketch the exact snippet.

Short answer: No. Global concurrency limits do not change the task runner’s max_workers. They’re independent controls that combine at runtime. How they interact - Task runner max_workers: local parallelism in the flow process (threads/processes). It caps how many tasks your runner can execute at once. - Global concurrency (Prefect’s server-managed limits via concurrency names/tags): a cross-run/org-wide gate. Tasks that try to enter a guarded section will wait for a lease; if none available, they go into AwaitingConcurrencySlot until a slot frees up. Effective concurrency = min(task runner max_workers, available global concurrency slots for the guarded resource) Example

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.concurrency.sync import concurrency

@task
def call_db(i):
    # At most 10 tasks across all runs can be inside this block concurrently
    with concurrency("db", occupy=1):
        return do_something(i)

@flow(task_runner=ThreadPoolTaskRunner(max_workers=50))
def pipeline(items):
    return [call_db.submit(i) for i in items]

- Up to 50 tasks can be scheduled locally. - Only 10 may run the guarded section at a time globally; others will wait. Notes - If you don’t use a concurrency guard (or tag-based limits), only max_workers applies. - You can tune the guard’s behavior: -

timeout_seconds

to fail if a slot isn’t acquired in time -

max_retries

and

strict

for stricter acquisition semantics Docs - Concurrency: Concurrency - Task runners: Task runners

Great question. In Prefect’s concurrency model: - Guarded resource: the thing you are limiting access to. It can be anything “logical” you choose to protect—e.g., an external API, a database, a GPU, a shared file, or even a named quota like "api-x". - Guarded section: the code region that must only run when a slot for that resource has been acquired. You mark it with the concurrency context manager so Prefect can enforce the limit globally. How it looks in code

from prefect import task, flow
from prefect.concurrency.sync import concurrency

@task
def call_api(i):
    # This block is the guarded section
    # It uses the guarded resource named "api-x"
    with concurrency("api-x", occupy=1):
        return do_api_call(i)

@flow
def pipeline(items):
    return [call_api.submit(i) for i in items]

- You then set a global limit for the resource “api-x” (e.g., 10 concurrent uses) in the UI or via CLI:

prefect concurrency-limit create api-x --limit 10

- Tasks entering the with block will acquire a slot. If none are available, they’ll wait (state: AwaitingConcurrencySlot) until a slot frees up, then proceed. Exiting the with block releases the slot. Extras you can control - Multiple names:

with concurrency(["db", "redis"], occupy=1): ...

acquires slots on both resources. - Occupy more than one slot:

with concurrency("api-x", occupy=3): ...

to represent heavier usage. - Timeouts:

timeout_seconds=

to fail if a slot can’t be acquired in time. - max_retries/strict: tune acquisition behavior. Alternative: tag-based limits (no code changes in the task body) - Tag tasks, then limit that tag:

@task(tags=["api-x"])
  def call_api(...): ...
  prefect concurrency-limit create api-x --limit 10

- Prefect limits all tasks carrying that tag across all flows. Reminder on interaction with task runners - Concurrency limits don’t change your task runner’s max_workers. Your effective parallelism is the minimum of: - task runner capacity (threads/processes), and - available concurrency slots for the guarded resource. Docs - Concurrency concepts: Concurrency

Yes—grab the exception from the future and add your own record. Two easy patterns: A) Use result(raise_on_failure=False)

from prefect.futures import wait

done, not_done = wait(check_stock_futures)

daily_stock_results = []
for fut in done:
    res_or_exc = fut.result(raise_on_failure=False)
    if isinstance(res_or_exc, Exception):
        daily_stock_results.extend([{"msg": str(res_or_exc), "status": "error"}])
    else:
        # assuming the task returns a list on success
        daily_stock_results.extend(res_or_exc)

B) Use the State and extract the exception

from prefect.futures import wait
from prefect.states import get_state_exception

done, not_done = wait(check_stock_futures)

daily_stock_results = []
for fut in done:
    if fut.state.is_completed():
        daily_stock_results.extend(fut.result())
    else:
        exc = get_state_exception(fut.state)
        daily_stock_results.extend([{"msg": str(exc), "status": "error"}])

Tip: if you need to know which stock failed, keep a mapping from future to its params:

futures = {}
for params in stock_params_list:
    f = check_stock_task.submit(params)
    futures[f] = params

done, _ = wait(list(futures))
for f in done:
    params = futures[f]
    ...

Refs: Futures, States

Easiest way: keep a mapping from each future to the params you used when you submitted it, then add those params into your results. Example that handles both success and failure and includes params in each record:

from prefect.futures import wait

# Submit and remember which params go with which future
future_to_params = {}
for params in stock_params_list:
    f = check_stock_task.submit(params)
    future_to_params[f] = params

# Wait for completion
done, _ = wait(list(future_to_params))

daily_stock_results = []
for f in done:
    params = future_to_params[f]
    res_or_exc = f.result(raise_on_failure=False)

    if isinstance(res_or_exc, Exception):
        # failed: add error record with params
        daily_stock_results.append({
            "status": "error",
            "msg": str(res_or_exc),
            "params": params,   # or pull out a specific field e.g. params["ticker"]
        })
    else:
        # succeeded: task returned a list; annotate each item with params
        for item in res_or_exc:
            daily_stock_results.append({
                **item,          # keep the task's fields
                "params": params # add the params
            })

Notes - Using a dict preserves the association even though

wait()

returns an unordered set of futures. - If your task returns a single dict instead of a list, just append one record instead of iterating. - If you only need one field from params (e.g., ticker), store that instead of the whole dict:

"ticker": params["ticker"]

Optional: make runs easier to identify in the UI by naming the task with the param (doesn’t change results, but helps debugging).

@task(task_run_name=lambda params: f"check-stock-{params['ticker']}")
def check_stock_task(params): ...

Short answer: Yes, but only inside the task run. Use Prefect’s runtime context to read the task’s bound parameters, then include them in what the task returns so your flow can aggregate them. Inside the task

from prefect import task
from prefect.runtime import task_run
from prefect import get_run_logger

@task
def check_stock_task(params):
    logger = get_run_logger()
    # Bound parameters for this task run
    bound = task_run.parameters()   # dict of {arg_name: value}
    ticker = bound["params"]["ticker"]  # or whatever key you use

    try:
        data = do_check(params)
        # Return with params/ticker included
        return [{"ticker": ticker, "status": "ok", "data": data}]
    except Exception as e:
        logger.warning(f"check failed for {ticker}: {e}")
        # Return an error record instead of failing
        return [{"ticker": ticker, "status": "error", "msg": str(e)}]

Then your flow just aggregates:

futs = [check_stock_task.submit(p) for p in stock_params_list]
done, _ = wait(futs)

daily_stock_results = []
for f in done:
    daily_stock_results.extend(f.result(raise_on_failure=False))

Notes - Runtime context like

prefect.runtime.task_run.parameters()

is only available inside the running task/flow. You can’t use it in the parent flow to introspect a child task’s inputs after the fact. - If you don’t want to modify the task, keep a future→params mapping in the flow (as shown earlier), or fetch params from the API using the future’s

task_run_id

(more overhead). Optional: fetch params from the API in the flow (if you really need it)

from prefect import get_client

@flow
def collect(fut):
    async def get_params():
        async with get_client() as client:
            tr = await client.read_task_run(fut.task_run_id)
            return tr.parameters
    params = anyio.run(get_params)  # or make your flow async

Docs: Runtime context

Short version: - ProcessPoolTaskRunner = local parallelism with Python processes on one machine. Great for CPU-bound work on a single host; simple to set up. - DaskTaskRunner = distributed execution via Dask. Scales from your laptop to a Dask cluster across many machines; best for very large fan-out or heterogeneous resources. Key differences - Scope/scale - ProcessPoolTaskRunner: Parallelism limited to cores on the current machine (e.g., 4–64). - DaskTaskRunner: Can use a local Dask cluster or connect to a remote one; scale to many machines and 1000s of tasks. - Setup - ProcessPool: No extra services. Just set max_workers. - Dask: Requires dask/distributed. You can: - Let Prefect spin up a local Dask cluster automatically, or - Connect to an existing Dask scheduler (recommended for production). - Workload fit - ProcessPool: CPU-bound tasks, single-node throughput; avoids the GIL with true process parallelism. - Dask: Large fan-out, mixed CPU/IO, task graphs with dependencies, need autoscaling or resource-aware scheduling. - Data movement/serialization - ProcessPool: Pickle boundary between processes; large objects incur copy cost. - Dask: Object store / distributed scheduler; data locality and spill-to-disk available, but still serialized when moving between workers. - Fault tolerance & observability - ProcessPool: If the flow process exits, you lose the pool; limited fault isolation. - Dask: Scheduler/worker separation; can survive individual worker restarts, rich dashboard. - Tuning knobs - ProcessPool: max_workers ≈ CPU cores; keep small. - Dask: Choose local vs remote cluster, worker resources, adaptivity, retries at Dask level if desired. How to use - ProcessPoolTaskRunner (single machine)

from prefect import flow, task
  from prefect.task_runners import ProcessPoolTaskRunner

  @task
  def cpu_bound(x):
      return do_heavy_compute(x)

  @flow(task_runner=ProcessPoolTaskRunner(max_workers=8))
  def main(items):
      futures = [cpu_bound.submit(i) for i in items]
      return [f.result() for f in futures]

- DaskTaskRunner (local or remote cluster)

# pip install "prefect[dask]" or pip install prefect-dask
  from prefect import flow, task
  from prefect_dask.task_runners import DaskTaskRunner

  @task
  def work(x):
      return do_task(x)

  # Ephemeral local Dask cluster:
  @flow(task_runner=DaskTaskRunner())
  def local_main(items):
      return [work.submit(i) for i in items]

  # Connect to an existing Dask scheduler:
  @flow(task_runner=DaskTaskRunner(address="<tcp://scheduler-host:8786>"))
  def cluster_main(items):
      return [work.submit(i) for i in items]

Notes - Prefect global concurrency limits (guards/tags) are independent and will still gate task execution with either runner. - For very many tiny tasks, Dask’s scheduler overhead can dominate—batch/chunk where possible. - For heavy CPU tasks on one box, ProcessPool is simpler and often faster due to lower orchestration overhead. More info - Task runners overview: https://docs.prefect.io/latest/concepts/task-runners/ - Dask integration (repo): https://github.com/PrefectHQ/prefect-dask If you share your workload (CPU vs IO, single host vs cluster, task sizes), I can recommend which runner to use and specific settings.

Short answer: “Concurrency” in Prefect (the guarded section/limit) isn’t a task runner—it just gates entry to a code block. When you don’t add Dask/Process/Thread runners, your tasks execute locally with minimal scheduling/serialization overhead. Adding Dask/Process/Thread introduces extra layers that pay off for heavier workloads, but can slow down tiny/fast tasks. Why it feels faster - No extra scheduler: With only a concurrency guard, there’s no Dask scheduler handshake, no worker spin-up, no network RPCs. - No cross-process serialization: Process pools and Dask require pickling your task, args, and results; that dominates micro-tasks. - Less context switching: Threads/processes add scheduling overhead; for small tasks that overhead > compute time. - Immediate local execution: Submitting and resolving futures in-process is just cheaper for quick work. What runners add (and why they’re slower for tiny tasks) - ThreadPoolTaskRunner: thread scheduling + contextvars propagation overhead. - ProcessPoolTaskRunner: process IPC + pickling/unpickling + context switch. - DaskTaskRunner: scheduler/worker communication, cluster startup (if ephemeral), serialization, network latency. When to use each - Keep it simple (no Dask/Process) + concurrency guard when: - Tasks are very fast (sub-10–100 ms) - You’re just throttling access to an external resource (API/DB) - ThreadPoolTaskRunner for I/O-bound tasks that are moderately heavy (100ms+), on one machine. - ProcessPoolTaskRunner for CPU-bound tasks with heavier compute (seconds+), on one machine. - DaskTaskRunner for large fan-out or multi-node scaling; amortize overhead with bigger tasks and a persistent cluster (pass address=... instead of ephemeral). Tips to close the gap - Batch/chunk micro-tasks so each task does more work. - If using Dask, connect to an existing cluster (address=...) to avoid per-run startup cost. - Avoid passing large objects between processes/workers; use result storage or smaller payloads. - Tune logging/result persistence if you’re ultra-latency-sensitive (fewer API logs, smaller batches). References - Concurrency limits: Concurrency - Task runners overview: Task runners - Dask integration: prefect-dask

Short answer: nothing bad. Your flow will still wait for those submitted tasks to finish even if you don’t call

wait()

. Details: - Prefect 3 flows automatically wait for all submitted task futures before the flow completes. If futures are part of the flow’s return value, Prefect resolves them before setting the final state. If they aren’t returned, the task runner context still ensures outstanding tasks reach terminal states before the flow exits. - You only need

wait()

if you want explicit control (e.g., a timeout) or to branch on which tasks finished vs. are still running. It returns two sets:

done

and

not_done

. When to use `wait()`: - You want a timeout and to continue the flow even if some tasks are still running. - You want to inspect which futures are complete vs. pending. - You’re not going to call

.result()

but still need synchronization/branching behavior. Relevant source (Prefect 3.x): - Flow resolves futures in its return value: flow_engine.handle_success - Future resolution logic: resolve_futures_to_states -

wait()

helper and its behavior: prefect.futures.wait

Assuming Prefect 3.x. You control deployment parameters via the

parameters

argument on

Flow.deploy(...)

(or

Flow.from_source(...).deploy(...)

). They’re merged at runtime with the flow function’s defaults. Here are the three common patterns you asked about: 1) Use all flow defaults (don’t pass parameters) - Python

from prefect import flow

@flow
def greet(name: str = "World", count: int = 1):
    print(f"Hello {name} x{count}")

greet.deploy(
    name="greet-defaults",
    work_pool_name="local"  # no parameters passed
)

Runtime params: name="World", count=1 (from flow signature) - CLI

prefect deploy ./flows.py:greet --name greet-defaults --pool local

2) Partial override (some provided, others from defaults) - Python

greet.deploy(
    name="greet-partial",
    work_pool_name="local",
    parameters={"name": "Alice"}  # count uses default=1
)

Runtime params: name="Alice" (deployment), count=1 (flow default) - CLI

prefect deploy ./flows.py:greet --name greet-partial --pool local --param name=Alice

3) Full override (provide all parameters explicitly) - Python

greet.deploy(
    name="greet-full",
    work_pool_name="local",
    parameters={"name": "Bob", "count": 3}
)

- CLI

prefect deploy ./flows.py:greet --name greet-full --pool local --params '{"name":"Bob", "count":3}'

Notes and gotchas - Where defaults come from: Prefect inspects your flow function signature to build the parameter schema and defaults. Missing values at runtime fall back to the flow’s defaults. - Validation: By default

enforce_parameter_schema=True

. The server validates types and fields against the flow’s schema but allows missing required parameters at deployment creation (they can be supplied later at run time). If a required param has no default and you don’t supply it at run time, the run will fail. - Priority: Flow defaults < parameters in prefect.yaml < CLI

--param/--params

at deploy time. - From source: You can do the same with source-based deployments:

flow.from_source(
    source="<https://github.com/org/repo.git>",
    entrypoint="flows.py:greet",
).deploy(
    name="greet-defaults",
    work_pool_name="local",
    # parameters omitted for defaults
)

- Disabling schema enforcement (if needed):

greet.deploy(..., enforce_parameter_schema=False)
# or via CLI:
# prefect deploy ... --no-enforce-parameter-schema

References - Flow.deploy signature: prefect/flows.py - Runner deployment + parameter schema capture: prefect/deployments/runner.py - Parameter schema generation from function defaults: prefect/utilities/callables.py - Server-side deployment parameter validation: prefect/server/api/deployments.py - CLI parameter parsing (

--param

,

--params

): prefect/cli/deploy/_config.py

Here are the most common reasons logs don’t show up in the Prefect UI (Prefect 3.x) and how to fix them. If you can share a small code snippet or where your logs are produced (flow/task, threads, subprocesses, third‑party libs), I can tailor this further. Quick checklist - Use

get_run_logger()

inside flows/tasks rather than

logging.getLogger(...)

- Enable print capture if you rely on prints:

@flow(log_prints=True)

/

@task(log_prints=True)

- Watch out for threads/subprocesses: Prefect context doesn’t cross boundaries - Ensure log levels/handlers are configured so logs are sent to the API - Large logs can be truncated; logs are batched (may appear with a small delay) Common scenarios and fixes 1) Using standard logging or print - Symptom: You see output locally but not in the UI. - Fix: - Use

get_run_logger()

inside flow/task:

from prefect import flow, task
    from prefect.logging import get_run_logger

    @task
    def t():
        get_run_logger().info("This shows in the UI")

    @flow
    def f():
        get_run_logger().info("Flow log in UI")
        t()

- To capture print output, enable:

@flow(log_prints=True)
    @task(log_prints=True)

- Global toggle:

export PREFECT_LOGGING_LOG_PRINTS=True

2) Log level filtering - Symptom: INFO appears, DEBUG doesn’t (or vice versa). - Notes: - Run loggers (

prefect.flow_runs

,

prefect.task_runs

) do not filter by level, so

get_run_logger().debug(...)

should reach the API regardless of

PREFECT_LOGGING_LEVEL

. - Other Prefect/internal/infra loggers respect

PREFECT_LOGGING_LEVEL

. - Fix: - If you need infra/extra logs:

export PREFECT_LOGGING_LEVEL=DEBUG

3) Third‑party library logs (requests, sqlalchemy, dask, etc.) - Symptom: Library logs not in UI. - Fix: Add them to extra loggers so they use Prefect’s API handler:

export PREFECT_LOGGING_EXTRA_LOGGERS=requests,sqlalchemy,dask

Then logs from those libraries can be sent to the API. 4) Threads and subprocesses - Symptom: Logs from background threads or subprocesses don’t show. - Cause: Prefect run context isn’t automatically available in new threads or processes. - Fix: - Threads: capture context and set it inside the thread, then use

get_run_logger()

. Or log in the parent before/after the thread and pass messages via queue. - Subprocesses: capture stdout/stderr and log them with

get_run_logger()

in the parent process:

result = subprocess.run([...], capture_output=True, text=True)
    <http://logger.info|logger.info>(result.stdout)
    if result.stderr:
        logger.error(result.stderr)

5) Batching and truncation - Symptom: Logs appear with delay or big logs are cut off. - Behavior: Logs are batched before sending; very large messages are truncated. - Fixes: - Reduce batch interval:

export PREFECT_LOGGING_TO_API_BATCH_INTERVAL=0.2

- Split large messages into chunks; avoid single huge strings - Optional flush after critical logs (advanced): await

APILogHandler.aflush()

in async contexts 6) Worker/infrastructure logs vs run logs - Symptom: You expect worker or infra logs to appear in the run logs. - Notes: - Run logs (from

get_run_logger()

) appear in the flow/task run views. - Worker/infrastructure logs are separate; in open-source, worker logs aren’t shown in the UI like they are in Cloud. - If you need visibility in the run view, log with

get_run_logger()

from within the flow/task. 7) Schema or handler misconfiguration - Rare, but check: - API logging is enabled:

PREFECT_LOGGING_TO_API_ENABLED=True

(default) - Using

get_run_logger()

actually attaches an API handler. You can quickly inspect handlers by printing

get_run_logger().logger.handlers

during a run. If none of the above explain it, please share: - How you’re logging (code snippet) - Whether logs are from inside a flow/task, a mapped task, a thread, or a subprocess - Any relevant env vars (

prefect config view | grep PREFECT_LOGGING

)

- Whether you’re on Prefect Cloud or open-source server Helpful references - Logging config and handlers: prefect/logging - Log batching and limits: prefect/logging/handlers.py - Flow/task run loggers: prefect/logging/loggers.py

Short answer: As a helper function, this is fine. But don’t create and cache a “global” logger at import time. Always resolve the logger inside the flow/task run so it binds to the current run context; otherwise logs won’t reach the UI. Key points and tweaks - Where you call it matters: - Good: call

get_logger(...)

inside each `@flow`/`@task` body right before logging. - Risky: calling it at module import (e.g.,

logger = get_logger(__name__)

at top-level) will fall back to stdlib logging because there’s no Prefect run context yet; those messages won’t go to the UI. - Exception to catch: - Prefer catching

prefect.exceptions.MissingContextError

when calling

get_run_logger()

instead of a broad

RuntimeError

. - Double-wrapping: -

get_run_logger()

returns a PrefectLogAdapter. Wrapping that in

logging.LoggerAdapter

“works,” but it’s not necessary and can be confusing. If you want to tag messages, consider adding the info to the message text. Arbitrary extras like

source_module

are not displayed in the UI. - Don’t cache between runs or across tasks: - A single cached logger can be bound to the wrong run or lose context (especially across threads/processes). Always fetch a logger per call in the active context. A leaner helper pattern

import logging

def get_prefect_aware_logger(name: str | None = None):
    try:
        from prefect.logging import get_run_logger
        return get_run_logger()  # Use inside flows/tasks
    except Exception:
        # Outside of a run, fallback is fine for local dev
        return logging.getLogger(name or __name__)

Usage

from prefect import flow, task

@task
def t():
    logger = get_prefect_aware_logger(__name__)
    <http://logger.info|logger.info>("hello from task")

@flow
def f():
    logger = get_prefect_aware_logger(__name__)
    <http://logger.info|logger.info>("hello from flow")
    t()

If you really want a “global” symbol, make it lazy - You can create a lightweight proxy that resolves

get_run_logger()

on each call, but simplest and most reliable is to fetch the logger where you log. Related tips - To capture print output in the UI, set

@flow(log_prints=True)

/

@task(log_prints=True)

. - Third-party library logs won’t appear unless you route them:

PREFECT_LOGGING_EXTRA_LOGGERS=requests,sqlalchemy,dask

. - Logs from threads/subprocesses won’t have Prefect context by default; capture output and log from the parent with

get_run_logger()

. Refs - get_run_logger implementation: prefect/logging/loggers.py - Handlers and API batching: prefect/logging/handlers.py

In Prefect 3.x there is no direct equivalent of Prefect 1.x’s flatten utility. The 3.x model is more Pythonic/imperative, so you handle flattening with normal Python, and you control mapping behavior with

unmapped

. What to use instead - Flatten with Python before mapping:

from prefect import flow, task

  @task
  def process_item(x: int) -> int:
      return x * 2

  @flow
  def my_flow():
      nested = [[1, 2], [3, 4], [5]]
      flat = [i for sub in nested for i in sub]  # or itertools.chain.from_iterable(nested)
      futs = process_item.map(flat)
      return futs.result()

- Or keep structure and use nested mapping:

from prefect import flow, task

  @task
  def process_item(x: int) -> int:
      return x * 2

  @task
  def process_batch(batch: list[int]) -> list[int]:
      futs = process_item.map(batch)
      return futs.result()

  @flow
  def my_flow():
      nested = [[1, 2], [3, 4], [5]]
      futs = process_batch.map(nested)
      return futs.result()  # [[2,4], [6,8], [10]]

Mapped vs static params (Prefect 3’s unmapped) - Use

unmapped

to keep a parameter static across a mapped call:

from prefect import flow, task, unmapped

  @task
  def add(x: int, n: int) -> int:
      return x + n

  @flow
  def my_flow():
      xs = [1, 2, 3]
      futs = add.map(xs, n=unmapped(10))  # n is static
      return futs.result()  # [11, 12, 13]

Summary - Prefect 1.x flatten → In 3.x, use Python to flatten lists or nest mapping as needed. - Prefect 1.x unmapped → Prefect 3.x still uses

unmapped

to control mapping vs static params.

Short answer: not automatically. In Prefect 3.x, a flow only fails if it tries to use a failed future. If a future fails in the background and you never resolve/use it, the flow can still complete successfully. Here’s the behavior at a glance: - Flow fails when: - You call

.result()

on a failed future (raises the original exception) - You pass a future into another task as a parameter (failure propagates via resolution) - You return a future (or data structure containing it) from the flow and it failed - Flow does not fail when: - You submit a task and never touch its future - You call

.wait()

alone (no exception is raised just by waiting) - You handle failures explicitly (see below) Ways to handle failures without failing the flow: - Catch when getting results

try:
      val = fut.result()
  except Exception as e:
      logger.warning(f"Handled task failure: {e}")
      val = None

- Don’t raise on failure

val_or_exc = fut.result(raise_on_failure=False)
  if isinstance(val_or_exc, Exception):
      # handle it

- Allow a failed upstream into a downstream task without failing it

from prefect.utilities.annotations import allow_failure

  down = downstream_task.submit(allow_failure(upstream_future))

- Inspect state after waiting

fut.wait()
  if fut.state.is_failed():
      # handle failure without raising

If you can share how you’re launching and using the futures (e.g., calling

.result()

, passing them to other tasks, or returning them), I can tailor the pattern.

Using wait(futures) will not fail the flow by itself. - What it does: Blocks until the futures reach a terminal state (or until the timeout) and returns a DoneAndNotDoneFutures object. - What it does not do: It does not raise exceptions for failed tasks and does not change the flow state. The flow only fails if you subsequently resolve a failed future (e.g., call .result() with default behavior, pass it to another task, or return it from the flow). Typical pattern:

from prefect import flow, task
from prefect.futures import wait

@task
def might_fail(x):
    if x % 2:
        raise ValueError("boom")
    return x

@flow
def my_flow():
    fs = [might_fail.submit(i) for i in range(5)]
    done_and_not = wait(fs)  # optional: wait(fs, timeout=30)

    # Inspect states without failing the flow
    failed = [f for f in done_and_not.done if f.state.is_failed()]
    succeeded = [f for f in done_and_not.done if f.state.is_completed()]

    # Safely collect results without raising
    results = []
    for f in succeeded:
        results.append(f.result())  # safe because is_completed()
    for f in failed:
        exc = f.result(raise_on_failure=False)  # returns the exception
        # handle/log the exception without failing the flow

    # Important: don't return failed futures or pass them to downstream tasks
    # unless you intend to propagate failure.
    return results

Notes: - wait(...) supports a timeout; unfinished tasks remain in not_done and keep running. - Calling the instance method f.wait() also does not raise; it just waits. Exceptions are only raised when you resolve results (e.g., f.result() with default raise_on_failure=True) or when Prefect resolves a future passed as input/return.

Mostly yes — but there are a couple cleanup tweaks depending on which path you pick. Key points: - map vs submit: Both are fine.

map

returns a

PrefectFutureList

with a convenient `.result()`; a list comprehension returns a plain list of futures. - futures.result(): Resolves all items and raises if any task failed (fail-fast). After this, you don’t need

wait_for

because you already have concrete results. - wait(futures): Just blocks until terminal states; it does not raise and does not materialize results. You’d still need to pull results (and handle failures) if you want them. Recommended patterns: Option A: Fail-fast, collect results, then print

# fan-out
futures = print_bdate.map(bdates)

# block and materialize values (raises if any failed)
results = futures.result()

# downstream — no wait_for needed because results are already materialized
print_result.submit(results)

Option B: Block without raising, handle failures, then print

from prefect.futures import wait

# fan-out
futures = print_bdate.map(bdates)

# block without raising
done, not_done = wait(list(futures))  # list() is safe; PrefectFutureList is iterable

# collect successes (and optionally log failures)
ok = []
for f in done:
    if f.state.is_completed():
        ok.append(f.result())  # safe because is_completed()
    else:
        _ = f.result(raise_on_failure=False)  # returns the exception; handle/log if needed

# downstream — results are materialized, so no wait_for needed
print_result.submit(ok)

Option C: Don’t materialize; just enforce ordering - If you only want to ensure

print_result

runs after the fan-out but don’t need the outputs:

futures = print_bdate.map(bdates)
print_result.submit(wait_for=futures)  # sequencing only

- If

print_result

needs the values and you pass the futures as parameters, you don’t need `wait_for`; Prefect will resolve dependencies automatically when it resolves the parameters. Note: any failed future will propagate and fail the downstream task unless you use

allow_failure

or handle exceptions. Small adjustments to your snippet: - If you use

results = futures.result()

, drop

wait_for=[futures]

on

print_result

— it’s redundant because you already resolved the values. - If you use

done, not_done = wait(futures)

, make sure you then build

results

from

done

(and handle failures) before calling

print_result(results)

, or use

wait_for=futures

when you’re not passing the outputs but still want ordering. If you share what behavior you want on failures (fail-fast vs tolerate some failures), I can tailor the final pattern.

In Prefect 3.x, a flow only fails if the flow function raises. When you just call

wait(download_futures)

, it won’t raise on failures—it just returns which futures are done. To fail the flow when any of the submitted tasks fail, you need to “touch” the futures so exceptions propagate. Two simple patterns: 1) Resolve all futures (raises if any failed)

from prefect.futures import resolve_futures_to_results

# ... after creating download_futures
resolve_futures_to_results(download_futures)  # raises on any failed task -> flow fails

2) Wait, then force result retrieval (also raises on failures, but after everything finishes)

from prefect.futures import wait

dn = wait(download_futures)
for f in dn.done:
    f.result()  # raises if that task failed -> flow fails

Drop-in change for your flow:

@flow(name="cot_reports_cftc_download", timeout_seconds=1800,
      on_completion=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_CHANNEL_SUCCESS])],
      on_failure=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_CHANNEL_FAILED])],
      on_crashed=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_CHANNEL_FAILED])],
      on_cancellation=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_CHANNEL_FAILED])],
)
def cot_reports_cftc_download_flow(
    business_date_str: str = "",
    download_path: str = "/data/raw_data/external/cot_reports/CFTC",
    file_url_mapping: List[List[str]] = DEFAULT_FILE_URL_MAPPING,
):
    bdates = get_business_dates(business_date_str, default_prev_bdate=True)

    download_futures = [
        download_file_from_url.submit(bdate, file_url, download_path)
        for bdate in bdates
        for file_url in file_url_mapping
    ]

    # Option A: propagate any failure (and return results if you like)
    from prefect.futures import resolve_futures_to_results
    resolve_futures_to_results(download_futures)

Option B with

wait

is equivalent if you prefer to ensure all tasks finish before raising:

from prefect.futures import wait

dn = wait(download_futures)
for f in dn.done:
    f.result()

Short answer: Yes—with your code as written, all the futures will finish before any exception is raised. Details: -

wait(download_futures)

(with no timeout) blocks until all submitted futures complete. By the time it returns,

dn.not_done

is empty. - Then, when you iterate

dn.done

and call

f.result()

, you’re just propagating any failures that already occurred; other futures have already finished. If you want to fail the flow only after everything completes but also report all errors, collect them first, then raise:

from prefect.futures import wait

dn = wait(download_futures)  # blocks until all complete
errors = []
for f in dn.done:
    try:
        f.result()
    except Exception as e:
        errors.append(e)

if errors:
    # Optionally log individual errors here
    raise RuntimeError(f"{len(errors)} download(s) failed")

Note: If you omit

wait(...)

and call

f.result()

directly in a loop, the first failure will raise immediately and the flow will fail early; remaining tasks may be cancelled.

Short answer: not really. As written, each

.result()

blocks the loop, so your tables will run serially. You submit a task, immediately wait for it, then submit the next, etc.—that prevents concurrency across tables. In Prefect 3.x, submit first, then wait later. Also, pass futures between tasks (or use

wait_for

) instead of resolving to concrete values in the loop. Here’s a concurrent rewrite that keeps per-table dependencies but lets tables run in parallel:

from prefect import flow
from prefect.futures import wait
from prefect.task_runners import ThreadPoolTaskRunner

@flow(task_runner=ThreadPoolTaskRunner(max_workers=8))  # tune workers to your workload
def gds_local_storage_flow(
    business_date_str: str = "",
    storage_path: str = "/data_local/filesync/suntime_consensus_data"
):
    logger = get_logger(__name__)
    process_date_str = date_range_process(business_date_str)

    all_tables = TABLES_WITH_CONYEAR + TABLES_NO_CONYEAR
    terminal_futures = []

    for table in all_tables:
        <http://logger.info|logger.info>(f"Querying table: {table}")

        # 1) Submit query (non-blocking)
        q_future = query_from_gds.submit(process_date_str, table)

        # 2) Submit processing, passing the future (do NOT call .result() here)
        p_future = process_and_save_frame.submit(
            q_future,
            storage_path,
            table,
            has_con_year=(table in TABLES_WITH_CONYEAR),
        )

        # 3) Submit final check, depending on processing completion
        latest_future = is_latest_trade_date_generated.submit(
            raw_date_str=business_date_str,
            storage_path=storage_path,
            process_date_str=process_date_str,
            table_name=table,
            wait_for=[p_future],
        )

        terminal_futures.append(latest_future)

    # Wait for all tables to finish
    wait(terminal_futures)
    <http://logger.info|logger.info>("All tables processed")

Notes: - Passing a

PrefectFuture

as an argument automatically creates the dependency and resolves to the value inside the downstream task. No need to call

.result()

in the loop. - Use

prefect.futures.wait

or iterate over

as_completed

if you want streaming progress. - The default runner is thread-based; if these tasks are CPU-bound, consider a process pool:

from prefect.task_runners import ProcessPoolTaskRunner

  @flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
  def gds_local_storage_flow(...):
      ...

If you share whether your tasks are I/O-bound (e.g., network, disk) or CPU-bound (e.g., heavy DataFrame compute), I can suggest the best runner and

max_workers

settings.

It depends on how you handle the futures. - If you call

.result()

on a task future (e.g.,

query_from_gds.submit(...).result()

), any failure in that task will raise and fail the flow immediately. - If you pass a future into a downstream task (like you should for concurrency), Prefect will resolve that parameter before running the downstream. If the upstream task failed, the downstream will raise

UpstreamTaskError

, and your flow will fail unless you explicitly allow failure. - If you just submit tasks and never touch their futures (don’t call

.result()

, don’t pass them downstream, don’t return them from the flow), their failures will not fail the flow. - Using

wait(futures)

or

as_completed(futures)

alone does not fail the flow; they only block/yield. The flow’s final state is determined by what you return or by exceptions raised while executing the flow body. Given your pattern: - In the serial version where you do

future.result()

for

query_from_gds

and later for

process_and_save_frame

, any failure in those tasks will fail the flow at that point. - In the concurrent rewrite I shared (submit, pass futures downstream, and

wait

at the end), any failure in

query_from_gds

will cause the downstream

process_and_save_frame

(or

is_latest_trade_date_generated

) to error with

UpstreamTaskError

, and the flow will fail unless you handle/allow it. If you want to tolerate some task failures without failing the entire flow, you have a few options: - Wrap

.result()

calls in try/except and handle/log the error. - Use

allow_failure(future)

when passing a future to a downstream task to prevent failure propagation for that dependency. - Aggregate results and errors explicitly, and return only a success indicator from the flow. If you share your desired failure semantics (fail on any table vs. continue and report which tables failed), I can show a concrete pattern.

Short answer: No.

wait(terminal_futures)

only blocks until they finish; it does not raise or fail the flow on its own. If you want the flow to fail when any terminal future fails, use one of these patterns: - Force failure on first error:

from prefect.futures import as_completed

for fut in as_completed(terminal_futures):
    fut.result()  # raises on failure, failing the flow

- Collect all failures, then fail:

from prefect.futures import wait

done, _ = wait(terminal_futures)
failed = [f for f in done if f.state.is_failed()]
if failed:
    raise RuntimeError(f"{len(failed)} terminal tasks failed")

- Return the futures from the flow:

return terminal_futures

Prefect will aggregate returned states; if any failed, the flow is marked failed. If you want to keep the flow successful while recording which tables failed, capture futures per table and log/return the failure list instead of raising:

table_to_future = {...}  # map table -> future
done, _ = wait(list(table_to_future.values()))
failed_tables = [t for t, f in table_to_future.items() if f.state.is_failed()]
return {"failed_tables": failed_tables}

Short answer: in Prefect 3, your pattern is exactly how to do it. Use

Task.map()

with both upstream iterables; Prefect zips them by index and automatically preserves the per‑item dependencies. You don’t need any extra orchestration. A cleaned-up version of your flow (note the fixed variable name and zipped mapping):

from prefect import flow, task, unmapped

@task
def get_business_dates(...): ...
@task
def fetch_china_stock_calendar(): ...
@task
def filter_trade_date(bdates_dt, trade_calendar_list): ...
@task
def get_stock_list(trade_date): ...
@task
def download_annual_dividend_data(trade_date, stock_list): ...
@task
def save_annual_dividend_data(dividend_df, trade_date): ...

@flow(
    name="financial_data_daily_download",
    on_completion=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_SUCCESS_CHANNEL])],
    on_failure=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_FAILED_CHANNEL])],
    on_cancellation=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_FAILED_CHANNEL])],
    on_crashed=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_FAILED_CHANNEL])],
)
def financial_data_daily_download(business_dates: str = ""):
    bdates_dt = get_business_dates(
        business_date_str=business_dates,
        timezone="Asia/Shanghai",
        default_prev_bdate=True,
    )
    trade_calendar_list = fetch_china_stock_calendar()
    process_trade_dates = filter_trade_date(bdates_dt, trade_calendar_list)

    # map over dates -> returns one future per date
    stock_lists = get_stock_list.map(process_trade_dates)

    # map over both dates and their corresponding stock list (zipped by index)
    dividend_dfs = download_annual_dividend_data.map(
        process_trade_dates, stock_lists
    )

    # per-item save, also zipped by index
    save_annual_dividend_data.map(dividend_dfs, process_trade_dates)

Notes and tips: - When you pass multiple iterables to

.map()

, Prefect zips them positionally; each

download_annual_dividend_data[i]

depends on both

process_trade_dates[i]

and

stock_lists[i]

. - All mapped iterables must be the same length. If one argument should be identical for all mapped calls, wrap it with

unmapped(...)

. - You generally don’t need to call

.wait()

between these steps; using mapped futures as inputs automatically creates correct dependencies. - Prefer

.map()

over loops with

.submit()

for this use case; it’s cleaner and records dependencies correctly. Question for you: do you intend to process per (date, stock) pair (i.e., explode each

stock_list

into individual stocks), or per date with the whole list at once? If you want per (date, stock), we can show a pattern to flatten to (date, stock) pairs and map over those.

Short answer: no,

.map()

itself doesn’t pile on memory just because each item waits for its paired input. Each mapped child run i only starts when its own two inputs i are ready; Prefect holds lightweight futures, not the actual values, until the downstream task executes. What can increase memory or slowdowns: - Large in-memory results: if an upstream task returns a big object and you don’t persist it, that object lives in the flow process until the downstream consumes it. - Very large fan-out: mapping over tens/hundreds of thousands of items creates many task runs/futures and can increase scheduler/driver memory. - Calling

.result()

too early: resolving all results at once materializes them in memory. Ways to keep memory under control - Persist results for large payloads - Write big data to storage inside the task (S3/GCS/local) and pass URIs/keys downstream. - Or set

persist_result=True

on tasks and configure result storage, so Prefect passes references instead of big in-memory objects.

@task(persist_result=True)
  def download_annual_dividend_data(date, stock_list):
      ...
      return big_df  # will be stored in result storage, not held in RAM

- Limit parallelism - Constrain task concurrency with tags:

@task(tags=["dividend"])
    def download_annual_dividend_data(...): ...

    # once, from your shell
    prefect concurrency-limit create dividend 5

- Or cap workers in your flow’s task runner:

from prefect import flow
    from prefect.task_runners import ThreadPoolTaskRunner

    @flow(task_runner=ThreadPoolTaskRunner(max_workers=8))
    def financial_data_daily_download(...):
        ...

- Chunk large maps - If

process_trade_dates

can be very long, process in batches to avoid creating a huge number of futures at once:

for batch in batched(process_trade_dates, 200):
      stock_lists = get_stock_list.map(batch)
      dividend_dfs = download_annual_dividend_data.map(batch, stock_lists)
      save_annual_dividend_data.map(dividend_dfs, batch)

- Avoid resolving results early - Don’t call

.result()

on large mapped collections until you actually need the data, and prefer passing futures between tasks. Net: your

.map(date, stock_list)

pattern is fine and won’t inherently bloat memory. If your payloads are large or your fan-out is big, use result persistence, concurrency limits, and batching to keep memory/throughput balanced.

Great question — you can serialize only the downloads while keeping everything else concurrent by using Prefect 3.x concurrency limits. Assuming you’re on Prefect 3.x, here’s the simplest pattern: 1) Tag the download task and set a tag-based concurrency limit of 1 - Prefect will ensure only one task with that tag runs at a time globally. - Other tasks (like save) remain fully concurrent because they’re not tagged. Code changes:

from typing import List
from prefect import flow, task

@task(tags=["annual-dividend-download"])
def download_annual_dividend_data(trade_date, stock_list):
    # your download logic here
    ...

@task
def save_annual_dividend_data(dividend_df, trade_date, local_path, upload_path):
    # your save logic here
    ...

@task
def get_stock_list(trade_date):
    # assuming this can be a task too
    ...

@flow(name="annual_dividend_daily_download")
def annual_dividend_daily_download_flow(
    business_dates: str = "",
    local_path: str = "/some/local",
    upload_path: str = "/data_local/filesync",
) -> List:
    bdates_dt = get_business_dates(
        business_date_str=business_dates,
        timezone="Asia/Shanghai",
        default_prev_bdate=True,
    )
    trade_calendar_list = fetch_china_stock_calendar()
    process_trade_dates = filter_trade_date(bdates_dt, trade_calendar_list)

    save_futures = []
    for trade_date in process_trade_dates:
        stock_list_future = get_stock_list.submit(trade_date)

        # Schedule all downloads concurrently, but Prefect will run only one at a time
        dl_future = download_annual_dividend_data.submit(trade_date, stock_list_future)

        # Schedule the save right away; it will wait on its own download
        save_futures.append(
            save_annual_dividend_data.submit(dl_future, trade_date, local_path, upload_path)
        )

    # Optionally wait for all saves
    return [f.result() for f in save_futures]

Then create the concurrency limit once:

prefect concurrency-limit create "annual-dividend-download" 1
# Helpful commands:
prefect concurrency-limit ls
prefect concurrency-limit inspect "annual-dividend-download"

Why this works - Only tasks tagged

annual-dividend-download

are serialized (one-at-a-time). - All other tasks (e.g.,

save_annual_dividend_data

) can run concurrently. - If one download fails, its corresponding save will be skipped (as it depends on that result), but other downloads/saves continue unaffected. - You can add retries to downloads if needed:

@task(tags=["annual-dividend-download"], retries=3, retry_delay_seconds=30)
def download_annual_dividend_data(...):
    ...

Alternative (advanced): global concurrency context If you prefer not to use tags, you can guard the critical section with a global concurrency limit and a context manager:

from prefect.concurrency.sync import concurrency

@task
def download_annual_dividend_data(trade_date, stock_list):
    with concurrency(names="dividend-processing", occupy=1):
        # critical section
        ...

Create the global limit:

prefect global-concurrency-limit create "dividend-processing" --limit 1

Docs and references - Concurrency overview: Concurrency limits in Prefect - CLI: CLI reference - Tag-based:

prefect concurrency-limit ...

- Global:

prefect global-concurrency-limit ...

If you’re on Prefect 2.x, let me know and I’ll adjust the guidance.