<@ULVA73B9P> How to retry flow using python api ?
# ask-marvin
r
@Marvin How to retry flow using python api ?
m
thought for 463.5 seconds
Assuming Prefect 3.x — here are the main ways to retry a flow using the Python API. 1) Configure automatic retries on the flow
Copy code
from prefect import flow

@flow(retries=3, retry_delay_seconds=10)
def my_flow():
    # your logic
    ...
Notes: -
retries
= number of times to retry the whole flow if it fails -
retry_delay_seconds
= fixed delay (int/float) between retries - Advanced strategies like exponential backoff or jitter are not applied at the flow level; use task-level retries for those - Docs: Retries how-to 2) Manually retry an existing failed flow run (Python API) Set the run back to Scheduled so a worker will pick it up again:
Copy code
import asyncio
from prefect.client.orchestration import get_client
from prefect.states import Scheduled

async def retry_flow_run(flow_run_id: str):
    async with get_client() as client:
        await client.set_flow_run_state(flow_run_id, state=Scheduled())

# asyncio.run(retry_flow_run("<your-flow-run-id>"))
Docs: Manually retry a flow run 3) Adjust the current flow’s retry policy at runtime (advanced) You can update the current run’s empirical policy while the flow is running:
Copy code
from prefect import flow
from prefect.context import FlowRunContext
from prefect.client.orchestration import get_client
from prefect.server.schemas.core import FlowRunPolicy
from prefect.server.schemas.actions import FlowRunUpdate

@flow
async def my_flow():
    ctx = FlowRunContext.get()
    async with get_client() as client:
        await client.update_flow_run(
            ctx.flow_run.id,
            FlowRunUpdate(empirical_policy=FlowRunPolicy(retries=5, retry_delay=20)),
        )
    ...
A few clarifications - Flow-level retries only support a single fixed delay; exponential backoff and jitter are task-only features. For advanced retry strategies, set them on tasks within the flow. - API refs: prefect.flows.Flow, FlowRunPolicy schema If you had a different scenario in mind (e.g., conditional retry, rescheduling at a specific time, or retrying from within a task), let me know and I can tailor examples.