https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> how can i extend the flow decorator t...
# ask-marvin
p
@Marvin how can i extend the flow decorator to set certain defaults, so i can use @my_default_flow in the actual code?
m
✅ thought for 95.3 seconds
The easiest pattern is to wrap 
flow
with 
functools.partial
.
Copy code
python
from functools import partial
from prefect import flow

my_default_flow = partial(
    flow,
    retries=3,
    timeout_seconds=300,
    log_prints=True,
)

@my_default_flow
def my_flow(x: int) -> int:
    return x + 1
You can also still override defaults per flow:
Copy code
python
@my_default_flow(timeout_seconds=30)
def fast_flow():
    return "ok"
Why this is the recommended approach - It uses the normal Prefect 3.x decorator behavior - It keeps your code simple - Prefect preserves function metadata/signature for the resulting flow object If you need extra logic beyond setting defaults, you can wrap it in a function:
Copy code
python
from prefect import flow

def my_default_flow(__fn=None, **kwargs):
    defaults = {
        "retries": 3,
        "timeout_seconds": 300,
        "log_prints": True,
    }
    defaults.update(kwargs)
    return flow(__fn, **defaults)

@my_default_flow
def a_flow():
    return 1

@my_default_flow(timeout_seconds=10)
def another_flow():
    return 2
If you want to support both 
@my_default_flow
and 
@my_default_flow(...)
, that wrapper pattern works well. Rule of thumb - Just defaults → use 
partial
- Defaults + custom validation/logging → use a wrapper function - I would not start by subclassing internals unless you really need it Example with validation:
Copy code
python
from prefect import flow

def my_default_flow(__fn=None, **kwargs):
    defaults = {
        "retries": 3,
        "timeout_seconds": 300,
        "log_prints": True,
    }
    defaults.update(kwargs)

    if defaults["timeout_seconds"] <= 0:
        raise ValueError("timeout_seconds must be positive")

    return flow(__fn, **defaults)
One caveat: avoid putting shared mutable defaults in there, like hook lists you plan to mutate. If you want, I can give you a fully typed version of 
my_default_flow
that preserves nicer IDE hints.
p
yes give it fully typed
@Marvin yes fully typed please
m
✅ thought for 19.4 seconds
Yep — here’s a fully typed version that supports both: - 
@my_default_flow
- 
@my_default_flow(...)
and preserves the wrapped function’s parameter/return typing using 
ParamSpec
and 
TypeVar
.
Copy code
python
from collections.abc import Callable
from typing import TypeVar, ParamSpec, overload, Any

from prefect import flow
from prefect.flows import Flow

P = ParamSpec("P")
R = TypeVar("R")


@overload
def my_default_flow(__fn: Callable[P, R]) -> Flow[P, R]:
    ...


@overload
def my_default_flow(
    __fn: None = None,
    *,
    name: str | None = None,
    version: str | None = None,
    flow_run_name: str | Callable[[], str] | None = None,
    retries: int | None = None,
    retry_delay_seconds: int | float | None = None,
    description: str | None = None,
    timeout_seconds: int | float | None = None,
    validate_parameters: bool = True,
    persist_result: bool | None = None,
    cache_result_in_memory: bool = True,
    log_prints: bool | None = None,
) -> Callable[[Callable[P, R]], Flow[P, R]]:
    ...


def my_default_flow(
    __fn: Callable[P, R] | None = None,
    *,
    name: str | None = None,
    version: str | None = None,
    flow_run_name: str | Callable[[], str] | None = None,
    retries: int | None = None,
    retry_delay_seconds: int | float | None = None,
    description: str | None = None,
    timeout_seconds: int | float | None = None,
    validate_parameters: bool = True,
    persist_result: bool | None = None,
    cache_result_in_memory: bool = True,
    log_prints: bool | None = None,
) -> Flow[P, R] | Callable[[Callable[P, R]], Flow[P, R]]:
    return flow(
        __fn,
        name=name,
        version=version,
        flow_run_name=flow_run_name,
        retries=3 if retries is None else retries,
        retry_delay_seconds=retry_delay_seconds,
        description=description,
        timeout_seconds=300 if timeout_seconds is None else timeout_seconds,
        validate_parameters=validate_parameters,
        persist_result=persist_result,
        cache_result_in_memory=cache_result_in_memory,
        log_prints=True if log_prints is None else log_prints,
    )
Usage:
Copy code
python
@my_default_flow
def flow_a(x: int) -> int:
    return x + 1


@my_default_flow(timeout_seconds=30, retries=5)
def flow_b(name: str) -> str:
    return f"hello {name}"
A few notes: Why this version is nice - IDEs can infer the original function signature better - It matches the normal decorator ergonomics - You can centralize your house defaults in one place If you want more Prefect options You can add more keyword arguments from 
flow(...)
as needed, like: - 
task_runner
- 
result_storage
- 
result_serializer
- 
on_completion
- 
on_failure
- 
on_cancellation
- 
on_crashed
- 
on_running
I kept the example shorter so it stays maintainable. One practical recommendation Only include the options you actually want to standardize. If you mirror the entire 
@flow
signature, your wrapper can become tedious to maintain across Prefect upgrades. If you want, I can also give you: 1. a full signature version that includes nearly all 
@flow
kwargs, or 2. a generic helper factory, like 
make_flow_decorator(retries=3, timeout_seconds=300)
.
Open in Slack
PreviousNext
Powered by