Hello - I have a flow containing a task that has a max_retries value of 1. I want to kick off another flow after the task has finished:

flow.set_dependencies(create_my_flow_run(), upstream_tasks=[task_with_1_retry])

But this only runs

create_my_flow_run

after the retry has finished, I want to run the flow after the first failure, so I tried adding:

@task
...
try:
    something_that_might_fail() 
finally:
    StartFlowRun(flow_name=my_flow.name).run()

To the bottom of

task_with_1_retry

- but getting the error:

Unexpected error while running flow: KeyError('Task slug case(1111)-1 is not found in the current Flow. This is usually caused by a mismatch between the flow version stored in the Prefect backend and the flow that was loaded from storage.

Any ideas how to achieve what I described? Thanks!

Hi @Heather - apologies for the delayed response. From your question it looks as though you’re still on Prefect version 1? It’s a while since I’ve used that but setting

upstream_tasks

might help? You can see more in this discourse question: https://discourse.prefect.io/t/how-can-i-define-state-dependencies-between-tasks/69