Charles Pereira
04/23/2026, 7:38 PMMarvin
04/23/2026, 7:38 PMMarvin
04/23/2026, 7:39 PMtimeout_seconds argument on the flow:
from prefect import flow
import time
@flow(timeout_seconds=300)
def my_flow():
time.sleep(600)
If the run exceeds that threshold, Prefect marks the flow run as Failed with a timeout error.
You can also override it on a copied flow with `with_options()`:
from prefect import flow
@flow(timeout_seconds=300)
def my_flow():
pass
fast_fail_flow = my_flow.with_options(timeout_seconds=60)
fast_fail_flow()
A couple of important notes:
- timeout_seconds is in seconds
- You can set timeouts on tasks too, separately from the flow
- In Prefect 3.x, this is configured in Python on the flow definition; it’s not typically set as a deployment-only knob
Example with both flow and task timeouts:
from prefect import flow, task
import time
@task(timeout_seconds=30)
def slow_task():
time.sleep(100)
@flow(timeout_seconds=120)
def my_flow():
slow_task()
If you want, I can also show:
- how this behaves in a deployment
- how to cancel or retry long-running flows
- how to enforce this for all flows in a project pattern-wise