Great question! In Prefect 3.x you’ve got a few options to delete multiple deployments in one go. CLI options - Delete all deployments:

prefect deployment delete --all

- Delete a specific list of deployments by name (loop):

for d in "flow_a/dep1" "flow_b/dep2" "flow_c/dep3"; do
  prefect deployment delete "$d"
done

- Delete by IDs (if you have them):

for id in "uuid-1" "uuid-2" "uuid-3"; do
  prefect deployment delete --id "$id"
done

Programmatic (Python) with filters - You can filter deployments (by name pattern, tags, etc.) and delete them with the client. Async example:

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import DeploymentFilter, DeploymentFilterName, DeploymentFilterTags

async def bulk_delete_by_name_like(name_substring: str):
    async with get_client() as client:
        deployments = await client.read_deployments(
            deployment_filter=DeploymentFilter(
                name=DeploymentFilterName(like_=name_substring)
            )
        )
        for dep in deployments:
            await client.delete_deployment(dep.id)
            print(f"Deleted: {dep.name}")

asyncio.run(bulk_delete_by_name_like("old-"))

Sync example:

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import DeploymentFilter, DeploymentFilterTags

with get_client(sync_client=True) as client:
    deployments = client.read_deployments(
        deployment_filter=DeploymentFilter(
            tags={"all_": ["prod", "deprecated"]}
        )
    )
    for dep in deployments:
        client.delete_deployment(dep.id)
        print(f"Deleted: {dep.name}")

Notes - Deletions are destructive—be sure of your filters before running. - The CLI does not take multiple names in a single command, but looping works well. - If you’d like help crafting a filter (e.g., by flow name, tag, or pattern), tell me what you want to target and I’ll tailor the command/snippet.

Yes—use the Python client’s

read_deployments

with a name filter that accepts multiple values. In Prefect 3.x, you can fetch them in a single request by passing

DeploymentFilterName(any_=...)

. Note: the deployment “name” here is the full name in the form

FLOW_NAME/DEPLOYMENT_NAME

(e.g.,

my_flow/prod

). Do you have them in that form? Async example:

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import DeploymentFilter, DeploymentFilterName

names = ["flow_a/dep1", "flow_b/dep2", "flow_c/dep3"]

async def fetch_by_names(names):
    async with get_client() as client:
        deployments = await client.read_deployments(
            deployment_filter=DeploymentFilter(
                name=DeploymentFilterName(any_=names)
            )
        )
        # Optional: map back by name
        return {d.name: d for d in deployments}

Sync example:

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import DeploymentFilter, DeploymentFilterName

names = ["flow_a/dep1", "flow_b/dep2", "flow_c/dep3"]

with get_client(sync_client=True) as client:
    deployments = client.read_deployments(
        deployment_filter=DeploymentFilter(
            name=DeploymentFilterName(any_=names)
        )
    )
    by_name = {d.name: d for d in deployments}

If you only have the deployment part (without the flow name), tell me how your names look and I can suggest a pattern (e.g., using

like_

or combining with a flow filter).

Yes — in Prefect 3.x that pattern is valid for fetching deployments by a list of names.

DeploymentFilterName(any_=...)

does an exact-name match and works with

client.read_deployments

. Minimal working example with imports:

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import DeploymentFilter, DeploymentFilterName

deployment_names = ["api-service", "worker-task", "data-pipeline"]

async with get_client() as client:
    deployments = await client.read_deployments(
        deployment_filter=DeploymentFilter(
            name=DeploymentFilterName(any_=deployment_names)
        )
    )

# deployments is a list of Deployment objects

Notes: -

any_

matches names exactly. If you want partial/case-insensitive matching, use `like_`:

DeploymentFilter(name=DeploymentFilterName(like_="prod"))

- If you prefer synchronous code, use

get_client(sync_client=True)

and call

client.read_deployments(...)

without

await

.

Great question! In Prefect 3.x, you set default parameters on the deployment, then override them dynamically when you trigger a run (Python or CLI). Here’s the setup and the ways to pass parameters at runtime. Step 1: Define parameters on your flow

from prefect import flow

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

Step 2: Set default parameters when creating the deployment - Via Python

if __name__ == "__main__":
    my_flow.deploy(
        name="my-deployment",
        work_pool_name="my-pool",
        parameters={"name": "Alice", "count": 5},  # defaults
        tags=["production"],
    )

- Or from a repo

flow.from_source(
    source="<https://github.com/org/repo.git>",
    entrypoint="flows.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-pool",
    parameters={"name": "Bob", "count": 3},
)

Step 3: Dynamically override parameters at run time Option A — Python helper: run_deployment (sync or async)

from prefect.deployments import run_deployment

# Trigger immediately with overrides
flow_run = run_deployment(
    name="my_flow/my-deployment",   # "<flow_name>/<deployment_name>"
    parameters={"name": "Eve", "count": 10},
    timeout=0  # return immediately instead of waiting
)

Option B — Python client (async)

from prefect.client.orchestration import get_client

async with get_client() as client:
    flow_run = await client.create_flow_run_from_deployment(
        deployment_id="your-deployment-id",
        parameters={"name": "Zoe", "count": 2},
        tags=["manual"],
        job_variables={"env.MY_FLAG": "true"}  # infra-level settings if needed
    )

Option C — CLI - Individual parameters:

prefect deployments run my_flow/my-deployment \
  -p name="Kai" \
  -p count=7

- JSON blob:

prefect deployments run my_flow/my-deployment \
  --params '{"name": "Mira", "count": 3}'

Notes and tips - Keys in

parameters

must match your flow function signature (extra/unknown keys will cause errors). - Runtime parameters are merged with deployment defaults; runtime wins on conflicts. - Values must be JSON-serializable. - You can also trigger with parameters from the UI: open the deployment → Run → edit parameters. If you tell me how you plan to trigger runs (Python, CLI, or UI) and an example of the parameters you want to pass, I can tailor a minimal snippet for your setup. Helpful docs: - Deployments (concepts)

Short answer: tasks end up stuck in Pending when the flow exits before you’ve waited for all submitted tasks, when dependencies never resolve, or when you’re not running in a flow context. With an “executor” that dynamically builds a DAG of async tasks, you’ll want to submit tasks as Prefect tasks and then explicitly wait for them before the flow returns. A good reference pattern - Define each block as a Prefect task (async is fine). - Orchestrate inside a Prefect flow (sync or async). - Submit tasks with .submit(...) and provide dependencies via either: - passing upstream futures as args/kwargs (implicit dependency + value passing), or -

wait_for=[upstream_futures]

(ordering-only dependency) - Before the flow returns, ensure you wait for all futures or collect their results. Example: dynamic DAG + async tasks, with ordering and value dependencies

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

# Example tasks (async blocks)
@task
async def fetch(x: int) -> int:
    # pretend async I/O
    import asyncio; await asyncio.sleep(0.1)
    return x * 2

@task
async def combine(a: int, b: int) -> int:
    return a + b

# block_specs is your dynamic description of the graph
# Each spec has:
# - key: identifier for this node
# - import_path and func_name to load the task
# - args/kwargs: inputs (you can include upstream futures directly to pass values)
# - upstream_keys: dependencies to respect even if no values are needed
@flow
async def workflow_executor(block_specs: list[dict]):
    futures_by_key = {}

    # Submit in topological order, or loop until all upstreams are scheduled
    pending = block_specs[:]
    while pending:
        ready = []
        for spec in pending:
            upstream_keys = spec.get("upstream_keys", [])
            if all(k in futures_by_key for k in upstream_keys):
                ready.append(spec)

        if not ready:
            raise RuntimeError("Cycle or unresolved dependencies in specs")

        for spec in ready:
            mod = import_module(spec["import_path"])
            task_fn = getattr(mod, spec["func_name"])  # must be a Prefect @task

            upstream_futures = [futures_by_key[k] for k in spec.get("upstream_keys", [])]
            args = spec.get("args", [])
            kwargs = spec.get("kwargs", {})

            # If downstream needs upstream RESULTS as inputs, pass the future(s) directly in args/kwargs.
            # Prefect will resolve to results and set dependencies automatically.
            # If you only need ordering, don't pass as input; use wait_for to enforce dependency.
            fut = task_fn.submit(*args, wait_for=upstream_futures, **kwargs)
            futures_by_key[spec["key"]] = fut
            pending.remove(spec)

    # CRITICAL: wait for all futures before the flow exits
    all_futures = list(futures_by_key.values())
    wait(all_futures)

    # Collect results if you need them
    return {k: f.result() for k, f in futures_by_key.items()}

Common reasons everything stays Pending - The flow returns before waiting: - Fix: call

prefect.futures.wait([...])

or

f.result()

on your futures before the flow completes. - Not running inside a Prefect flow: - Fix: all orchestration must happen inside a function decorated with

@flow

. - Dependencies never resolve: - Fix: ensure your

wait_for

list contains Prefect futures (not

asyncio.Task

, not plain values). - Check upstream futures for failures/timeouts via

future.state

. - Misusing event loops: - Fix: do not create your own event loop. If your flow is async, call it as

asyncio.run(my_flow())

from main, or just call

my_flow()

(Prefect will manage execution). - Concurrency limits: - Fix: verify you’re not blocked by tag/global concurrency limits; check states and logs. When every block is an async task

- You can and should still use

.submit()

to schedule tasks and achieve concurrency under Prefect’s task runner; then use

wait()

or

.result()

to ensure completion before exit. - If you have raw async helper functions (not decorated), you can

await

them or use

asyncio.gather

, but those won’t be tracked as Prefect tasks unless you wrap them with

@task

. Minimal pattern if you don’t need dynamic loading, just async-with-deps:

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

@task
async def a():
    return 1

@task
async def b(x: int):
    return x + 1

@flow
async def f():
    af = a.submit()
    bf = b.submit(af)            # implicit dependency + value passing
    cf = b.submit(0, wait_for=[af, bf])  # explicit ordering, no value dependency
    wait([af, bf, cf])
    return cf.result()

If you can share a small snippet of your current executor (how you load the module, submit tasks, and where you wait), I can pinpoint the exact cause of the Pending state and suggest a targeted fix.

If

node_futures

is

{}

, that’s fine as an initializer — but it must not be passed directly to

wait_for

.

wait_for

expects a single

PrefectFuture

or a list of futures. Passing a dict (especially

{}

) won’t create any dependency and can lead to confusing behavior. Use this pattern when you maintain a dict of futures keyed by node id:

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

@task
async def do_work(x: int) -> int:
    import asyncio; await asyncio.sleep(0.1)
    return x * 2

@flow
async def executor(specs: list[dict]):
    node_futures = {}  # start empty

    pending = specs[:]
    while pending:
        ready = [s for s in pending if all(k in node_futures for k in s.get("upstream_keys", []))]
        if not ready:
            raise RuntimeError("Cycle or unresolved dependencies in specs")

        for spec in ready:
            mod = import_module(spec["import_path"])
            task_fn = getattr(mod, spec["func_name"])  # must be a @task

            upstream_futs = [node_futures[k] for k in spec.get("upstream_keys", [])]
            args = spec.get("args", [])
            kwargs = spec.get("kwargs", {})

            # IMPORTANT: pass a list of futures, not the dict
            fut = task_fn.submit(*args, wait_for=upstream_futs, **kwargs)
            node_futures[spec["key"]] = fut
            pending.remove(spec)

    # IMPORTANT: wait before flow exits
    wait(list(node_futures.values()))
    return {k: f.result() for k, f in node_futures.items()}

Common pitfalls that cause “Pending forever” - Passing the dict itself:

task_fn.submit(..., wait_for=node_futures)

❌ - Do:

wait_for=[node_futures[k] for k in upstream_keys]

✅ - Exiting the flow before waiting for futures: - Do:

wait(list(node_futures.values()))

or call

.result()

on each future. - Not running in a flow context (no

@flow

around the orchestrator). - For deployments: no worker polling the work pool. If the flow run is Pending in the UI, make sure a worker is started for the correct pool/queue and there aren’t blocking concurrency limits/tags mismatches. If you can share the snippet where you: - build

node_futures

- construct

wait_for

- end the flow (do you call

wait

or

.result()

?) …I can pinpoint the exact fix.

Short answer: Prefect will only infer dependencies from what you pass to the task call. If you pass a dict that contains Prefect futures as values (even nested), Prefect will recursively detect those futures, wait for them, and replace them with their results before executing the task. If you just keep futures in an internal dict and don’t pass them to the task (or

wait_for

), Prefect won’t know about the dependency. Concretely, this works (nested dict of futures as an argument):

from prefect import flow, task

@task
def merge(node_results: dict):
    # node_results["node_results"] arrives with actual results, not futures
    return sum(node_results["node_results"].values())

@task
def inc(x): return x + 1

@flow
def f():
    a = inc.submit(1)          # PrefectFuture
    b = inc.submit(2)          # PrefectFuture

    payload = {
        "node_results": {
            "node_1": a,       # future
            "node_2": b        # future
        }
    }
    merged = merge.submit(payload)  # Prefect infers deps from futures inside the dict
    return merged.result()

This also works if you don’t need the upstream values but want ordering only:

# Values here are plain literals; deps enforced via wait_for
merged = merge.submit({"node_results": {"node_1": 0, "node_2": 0}}, wait_for=[a, b])

This does NOT create any dependency:

# Passing your dict object to wait_for does nothing useful
merge.submit(..., wait_for={"node_results": {"node_1": a, "node_2": b}})   # ❌

Key points - Passing PrefectFuture objects anywhere in args/kwargs (including nested dicts/lists/tuples) creates implicit dependencies and Prefect resolves them to results before the task runs. - Use

wait_for=[...futures...]

when you need ordering but don’t want to pass the values. - Don’t pass the futures container dict itself to

wait_for

— give Prefect a list of futures. - Make sure your orchestrator is a Prefect

@flow

and you

wait()

or

.result()

on all futures before the flow exits; otherwise tasks can remain Pending. - Ensure you’re passing Prefect futures (not

asyncio.Task

or other future types).

Likely yes — that nested dict is adding implicit dependencies. How Prefect treats your nested structure - If your task is called with args/kwargs that contain PrefectFuture objects anywhere (even deeply nested), Prefect will: - Register those futures as upstream dependencies, and - Resolve them to their results before your task starts - So if your payload includes futures for node_1 and node_2, your downstream task will not start until both of those futures are finished, even if your

wait_for

only lists a subset. Example (this will wait on both a and b because they’re in the payload)

@task
def merge(node_results: dict): ...

@flow
def f():
    a = t.submit(1)
    b = t.submit(2)
    payload = {"node_results": {"node_1": a, "node_2": b}}
    # Even though wait_for only includes `a`, Prefect will also wait on `b`
    fut = merge.submit(payload, wait_for=[a])

If that’s not what you want, you have a few options: - Use only ordering deps (don’t pass futures in the payload):

payload = {"node_results": {"node_1": 0, "node_2": 0}}  # placeholders, not futures
  fut = merge.submit(payload, wait_for=[a, b])

- Resolve futures to values before passing them (blocks until ready, then no implicit deps remain):

from prefect.futures import resolve_futures_to_results
  payload = {"node_results": {"node_1": a, "node_2": b}}
  payload = resolve_futures_to_results(payload)  # resolves and blocks here
  fut = merge.submit(payload)  # no implicit deps now

- Or pass literal values you already computed:

payload = {"node_results": {"node_1": a.result(), "node_2": b.result()}}
  fut = merge.submit(payload)

Other reasons tasks stay Pending to double-check - Upstream futures in your payload are themselves stuck (failed, waiting on others, or unscheduled). - The flow exits before you

wait()

or call

.result()

on submitted futures. - You accidentally passed non-Prefect futures (e.g.,

asyncio.Task

) in the payload; Prefect won’t await those implicitly. - Dependency cycles in the graph. If you share a minimal snippet (how you build the payload and submit the downstream task), I can point to the exact dependency causing the wait. Docs for reference: - Tasks and dependencies - Futures and waiting

Short answer: not reliably. If you keep calling

.submit()

and let the flow return without waiting (or without returning those futures), Prefect does not guarantee those tasks will finish. When the flow ends, the task runner is torn down and pending tasks can be cancelled. Only futures that are part of the flow’s return value (or explicitly waited on) are guaranteed to be awaited before the flow reaches a terminal state. What to do instead - If you don’t need the results, still wait for completion:

from prefect.futures import wait

  @flow
  def my_flow():
      futures = [my_task.submit(i) for i in range(10)]
      wait(futures)  # ensure they all run to completion before the flow exits

- If you need results, block on them or return them:

@flow
  def my_flow():
      futures = [my_task.submit(i) for i in range(10)]
      return [f.result() for f in futures]  # or just `return futures` to let Prefect resolve them

- If a downstream task references upstream futures (even nested in dicts/lists), Prefect will infer those dependencies and wait on them before starting the downstream. But if you never reference a future and never wait on it, it can be cancelled at teardown. Rules of thumb - Submit → wait (with

wait([...])

) or return the futures. - Don’t rely on “fire-and-forget”

.submit()

inside a flow; it’s not guaranteed. Helpful docs: - Run work concurrently - Futures and waiting

In Prefect 3.x there isn’t an on_success/on_failure keyword; the task callbacks are named on_completion, on_failure, and on_running. You attach them either via the task decorator or with decorator methods on the task object. Prefect passes three arguments to each callback: (task, task_run, state). To pass your own extra parameters, wrap your callback with functools.partial or use a closure/factory. Examples - Attach callbacks at definition time

from functools import partial
from prefect import task, flow

def notify(task, task_run, state, channel):
    print(f"[{channel}] {task.name} -> {state.name}")

@task(
    on_completion=[partial(notify, channel="success-channel")],
    on_failure=[partial(notify, channel="alerts-channel")]
)
def do_work(x: int) -> int:
    return x * 2

@flow
def f():
    do_work(21)

f()

- Attach callbacks after definition

from functools import partial
from prefect import task

@task
def risky(x: int) -> int:
    if x < 0:
        raise ValueError("boom")
    return x

@risky.on_failure
def alert(task, task_run, state):
    print(f"FAILED: {task.name} - {state.message}")

@risky.on_completion
def log_done(task, task_run, state):
    print(f"DONE: {task.name}")

- Async callbacks are fine

import asyncio
from prefect import task

@task(on_completion=[lambda task, task_run, state: asyncio.create_task(asyncio.sleep(0))])
def t():
    return 1

Better:

async def async_hook(task, task_run, state):
    await asyncio.sleep(0.1)

@task(on_completion=[async_hook])
def t():
    return 1

- Passing extra parameters using partial/closures

from functools import partial

def post_to_webhook(task, task_run, state, url, headers=None):
    # send state info somewhere...
    pass

webhook = partial(post_to_webhook, url="https://example.com/hook", headers={"X-TOKEN": "abc"})
@task(on_failure=[webhook])
def t():
    ...

Key details - Callback signature is exactly: (task, task_run, state) -> None or Awaitable[None]. - Provide callbacks as a list: on_completion=[...], on_failure=[...], on_running=[...]. - You can mix sync and async callbacks; Prefect will await async ones. - If you need transaction semantics, there are separate on_commit/on_rollback hooks with different signatures. Docs to skim: - Tasks and hooks: https://docs.prefect.io/latest/guide/concepts/tasks/

TL;DR - Tasks: on_completion and on_failure have been available since Prefect 2.10.0 (and in all 3.x, including 3.0.0). on_running for tasks was added in Prefect 3.5.0. - Flows: on_running has been available since Prefect 2.16.3 (and in all 3.x). Details and references - Task on_completion/on_failure - Introduced in Prefect 2.10.0 - Available in Prefect 3.0.0+ as well - Reference: PR #8605 - Add on_failure and on_completion task hooks - Flow on_running - Introduced in Prefect 2.16.3 - Available in Prefect 3.0.0+ as well - Reference: PR #12153 - Add on_running state change hook (flows) - Task on_running - Introduced in Prefect 3.5.0 (not present in 3.0.0–3.4.x) - References: - PR #19277 - Add on_running hook support for tasks - Prefect 3.5.0 release notes

Short answer - on_crashed exists for flows, not for tasks. - It’s available in both Prefect 2.x and 3.x for flows. - on_failure and on_completion exist for both tasks and flows. - on_running exists for flows in 2.16.3+ and for tasks in 3.5.0+. - In all these hooks, you receive the state object and can read the result from it. Details - Flows - Available hooks: on_completion, on_failure, on_running, on_cancellation, on_crashed. - on_crashed is supported in 2.x and 3.x. - In a flow hook your function signature is

(flow, flow_run, state)

. You can access the result with

state.result(raise_on_failure=False)

. - Tasks - Available hooks: on_completion, on_failure, on_running (task on_running added in 3.5.0; not present in 3.0.0–3.4.x). - There is no on_crashed for tasks. - In a task hook your function signature is

(task, task_run, state)

. You can access the result with

state.result(raise_on_failure=False)

. Notes and caveats - Crashed vs Failed: on_failure only runs for Failed; it does not run for Crashed. Use on_crashed (flows) to catch crashes. - Accessing results: In any hook,

state

is the post-transition state. You can read: -

state.name

,

state.type

,

state.message

-

state.result()

/

await state.aresult()

to get the value (if Completed) or the exception (if Failed). For Crashed, there usually isn’t a result. - Env toggle for certain hooks: Flow on_cancellation/on_crashed hooks respect an internal toggle used by some workers; they can be disabled in certain subprocess contexts. In normal runs they are enabled. If you share your Prefect version and whether you want hooks on tasks or flows, I can give you a precise snippet using the hooks your version supports.

The latest stable Prefect release is 3.6.8 (released Dec 24, 2025). - Full changelog: https://github.com/PrefectHQ/prefect/compare/3.6.7...3.6.8 - All releases: https://github.com/PrefectHQ/prefect/releases Upgrade:

pip install -U prefect

Or check your installed version:

python -c "import prefect, sys; print(prefect.__version__)"

You’ve got a few good sources: - GitHub Releases (official tags, clearly marks pre-releases): https://github.com/PrefectHQ/prefect/releases - PyPI release history (all published versions): https://pypi.org/project/prefect/#history - Prefect docs release notes (curated highlights by version): https://docs.prefect.io/latest/release-notes/ Quick ways to check from your terminal: - Show available versions on PyPI:

pip index versions prefect

- Check what you have installed:

python -c "import prefect; print(prefect.__version__)"

Tip: On GitHub, prefer entries without the “Pre-release” badge; on PyPI, pre-releases are marked and aren’t installed by default unless you opt-in.

Install Prefect 3.5.0 or newer. That’s the first version with the full set of state hooks you’ve asked about, including task on_running. I recommend the latest stable (3.6.8) to get all fixes. Minimums by hook: - Tasks: on_completion/on_failure (>=2.10.0), on_running (>=3.5.0) - Flows: on_completion/on_failure (2.x+), on_running (>=2.16.3), on_cancellation/on_crashed (2.x+) - Note: on_crashed exists for flows only (not tasks) Install/upgrade:

pip install -U "prefect>=3.5.0"
# or pin the latest stable
pip install -U "prefect==3.6.8"

Verify your version:

python -c "import prefect; print(prefect.__version__)"

References: - Prefect releases - Task hooks docs - Flow hooks docs

@Marvin I install 3.6.0 for fixes on those state hooks

Good call — Prefect 3.6.0 includes all the state hooks you asked about: - Tasks: on_completion, on_failure, and on_running - Flows: on_completion, on_failure, on_running, on_cancellation, on_crashed If you’re installing “for fixes,” I’d still recommend the latest 3.6.x (currently 3.6.8) since it includes additional bug fixes. But 3.6.0 will have the features. Quick self-check snippet

from prefect import task, flow

events = []

@task(
    on_running=[lambda t,tr,s: events.append(("task","running",s.name))],
    on_completion=[lambda t,tr,s: events.append(("task","done",s.name))],
    on_failure=[lambda t,tr,s: events.append(("task","failed",s.name))],
)
def t(x): return x

@flow(
    on_running=[lambda f,fr,s: events.append(("flow","running",s.name))],
    on_completion=[lambda f,fr,s: events.append(("flow","done",s.name))],
    on_failure=[lambda f,fr,s: events.append(("flow","failed",s.name))],
    on_crashed=[lambda f,fr,s: events.append(("flow","crashed",s.name))],
    on_cancellation=[lambda f,fr,s: events.append(("flow","canceled",s.name))],
)
def f():
    t(1)
    return "ok"

f()
print(events)

Notes - Callbacks receive `(task/flow, task_run/flow_run, state)`; you can read results via

state.result(raise_on_failure=False)

. - There’s no on_crashed for tasks (flows only). Links - Releases: Prefect releases - Task hooks: Task hooks - Flow hooks: Flow hooks If there’s a specific hook-related bug you’re targeting, share details and I’ll confirm if it’s fixed in 3.6.0 or recommend a patch version.

Awesome. Since you’re on Prefect 3.6.0, you have task on_running/on_completion/on_failure available. Here’s a step-by-step pattern to centralize hooks and pass dynamic parameters cleanly. Step 1: Create a central hooks module - Put shared hooks in one place, e.g. my_project/hooks.py - Use partials to bind per-task static config - Use Prefect runtime to read per-run dynamic context (parameters, tags, names) Example: my_project/hooks.py

from functools import partial
from prefect import get_run_logger
from prefect.runtime import task_run, flow_run

def _result_from_state(state, raise_on_failure=False):
    try:
        return state.result(raise_on_failure=raise_on_failure)
    except Exception as exc:
        return exc

def log_running(task, task_run_obj, state, *, group=None):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"[RUNNING] group={group} task={task.name} run_id={task_run_obj.id} "
                f"flow={flow_run.flow_name} params={task_run.parameters}")

def log_completion(task, task_run_obj, state, *, group=None):
    logger = get_run_logger()
    value = _result_from_state(state, raise_on_failure=False)
    <http://logger.info|logger.info>(f"[COMPLETED] group={group} task={task.name} value={value}")

def log_failure(task, task_run_obj, state, *, group=None, alert=False):
    logger = get_run_logger()
    err = _result_from_state(state, raise_on_failure=False)
    logger.error(f"[FAILED] group={group} task={task.name} error={err} message={state.message}")
    if alert:
        # send alert to Slack/Email/etc. (pseudo)
        logger.error("Alert sent")

# Helpers to pre-bind per-task static config
def hooks_for_group(group: str, *, alert_on_failure: bool = True):
    return {
        "on_running": [partial(log_running, group=group)],
        "on_completion": [partial(log_completion, group=group)],
        "on_failure": [partial(log_failure, group=group, alert=alert_on_failure)],
    }

Step 2: Create a reusable decorator to attach hooks to many tasks - This keeps all 20 tasks consistent and avoids duplication. Example: my_project/decorators.py

from prefect import task
from my_project.hooks import hooks_for_group

def instrumented_task(group: str, **task_kwargs):
    h = hooks_for_group(group)
    def wrap(fn):
        return task(
            on_running=h["on_running"],
            on_completion=h["on_completion"],
            on_failure=h["on_failure"],
            **task_kwargs,
        )(fn)
    return wrap

Step 3: Use the decorator across your task modules - Each task can specify its group (or any static knob you want to bind via partial). Example: tasks in different folders

# module_a/tasks.py
from my_project.decorators import instrumented_task

@instrumented_task(group="etl")
def extract():
    return {"rows": 100}

# module_b/tasks.py
from my_project.decorators import instrumented_task

@instrumented_task(group="ml", retries=1)
def train(model_name: str, epochs: int = 10):
    return {"model": model_name, "epochs": epochs}

Step 4: Pass dynamic parameters to hooks (per-run) without changing the hook signature - Use Prefect runtime in your hook: - task_run.parameters gives you the parameters for this specific task run - task_run.tags for routing/tenancy flags - flow_run.parameters if you need flow-level inputs - Alternatively, return structured results from your task and read them in the hook via state.result(raise_on_failure=False). Examples of dynamic parameters inside hooks - Based on task params:

def log_completion(task, task_run_obj, state, *, group=None):
    logger = get_run_logger()
    value = state.result(raise_on_failure=False)
    priority = task_run.parameters.get("priority", "normal")
    <http://logger.info|logger.info>(f"[COMPLETED] group={group} priority={priority} task={task.name} value={value}")

- Based on flow params: ``` def log_running(task, task_run_obj, state, *, group=None): logger = get_run_logger() tenant = flow_run.parameters.get("tenant", "default")

logger.info(f"[RUNNING] tenant={tenant} group={group} task={task.name}")

- Based on task result payload:

def log_failure(task, task_run_obj, state, *, group=None, alert=False): logger = get_run_logger() outcome = state.result(raise_on_failure=False) # exception or result # If your task returns {"meta": {...}} you can read meta here for routing

Step 5: If you already have Task objects, use with_options to attach the same centralized hooks

from my_project.hooks import hooks_for_group h = hooks_for_group("analytics", alert_on_failure=True) task_a = task_a.with_options( on_running=h["on_running"], on_completion=h["on_completion"], on_failure=h["on_failure"], )

Step 6: Verify quickly

from prefect import flow from module_a.tasks import extract from module_b.tasks import train @flow def pipeline(): extract() train("resnet", epochs=5, priority="high") pipeline() ``` Notes and best practices - Hook signature must remain (task, task_run, state). Extra params come from partials (static, per-task) or Prefect runtime (dynamic, per-run). - Prefer async hooks for network I/O or wrap sync HTTP calls with a non-blocking pattern if needed. - Hooks shouldn’t raise; exceptions are logged and ignored. - For very global configuration (e.g., webhook URL), consider environment variables or Prefect Variables/Blocks; read them inside the hook. References - Tasks and hooks: https://docs.prefect.io/latest/guide/concepts/tasks/ - Runtime context (flow_run/task_run): https://docs.prefect.io/latest/guide/reference/runtime/

Short answer: No. In Prefect 3.x, if you set a task’s

wait_for

to multiple upstream tasks and any one of them fails/crashes, the downstream task will not start. It stays in a Pending state named “NotReady” because an upstream did not reach COMPLETED. If you want the child to run even when some parents fail, wrap those specific upstream futures with

allow_failure

. You can do this either in the arguments or directly in

wait_for

. Example without allow_failure (child never starts if any parent fails):

from prefect import flow, task

@task
def parent_ok(i): 
    return i

@task
def parent_fail():
    raise ValueError("boom")

@task
def child():
    print("I will not run if any parent failed.")

@flow
def my_flow():
    f1 = parent_ok.submit(1)
    f2 = parent_fail.submit()
    f3 = parent_ok.submit(3)
    # Because one parent fails, child will remain Pending("NotReady")
    child(wait_for=[f1, f2, f3])

my_flow()

Example with allow_failure (child will run even if a specific parent fails):

from prefect import flow, task, allow_failure

@task
def parent_ok(i): 
    return i

@task
def parent_fail():
    raise ValueError("boom")

@task
def child():
    print("I will run once the non-failing parents complete, even if one failed.")

@flow
def my_flow():
    f1 = parent_ok.submit(1)
    f2 = parent_fail.submit()
    f3 = parent_ok.submit(3)
    # Wrap only the failing parent with allow_failure
    child(wait_for=[f1, allow_failure(f2), f3])

my_flow()

Notes: - You can selectively wrap only the upstreams you’re okay ignoring on failure; the others must still complete successfully before the child runs. - Task hooks don’t change scheduling behavior—they’re great for alerts/side effects, but they won’t cause a downstream to start. Docs and references: - How to run work concurrently (covers

wait_for

): How to run work concurrently - Task concepts (state/dependency behavior): Task concepts - Annotations API (

allow_failure

): Annotations API If you share a small snippet of your flow, I can suggest exactly where to apply

allow_failure

based on which parents you want to tolerate failing.