hey! this is a minimal example of a flow that i am trying to pass

retries

to:

@flow(name="my-dbt-flow", retries=5)
def my_dbt_flow(run_dbt: bool = True):
    if run_dbt:
        # Runs a Prefect flow until completion
        dbt_flow = create_flow_run(flow_name="dbt", wait=True)

        if dbt_flow.state_name == 'Completed':
            do_something(wait_for=[dbt_flow])
        else:
            raise Exception("DBT did not complete successfully.")

however, having deployed it and tested it, upon reaching the

Exception

ive raised, it does not perform a retry, rather just sits in

Failed

. what am i doing wrong?

That looks correct. Have you tried adding

retry_delay_seconds

? Example for that here https://docs.prefect.io/latest/tutorial/flows/