https://prefect.io logo
Channels
Powered by
#prefect-server
Title
# prefect-server
c

Casey Green

08/24/2021, 3:35 PM
Hi. I'm trying to write a flow that has a nested for-while loop, where the while loop contains multiple tasks. I don't think this is possible - it's sort of a combination of apply_map and task looping with signal.LOOP. Any pointers/ideas for how to solve this pattern? Or is it definitely not possible? Here's what kinda what the code looks like:
for_while_flow.py
k

Kevin Kho

08/24/2021, 3:52 PM
Hey @Casey Green, been thinking about this for a bit and I don’t think there is a way to get this to work because LOOPING itself is confined to one task. Looping over multiple tasks breaks the acyclic nature of the DAG. So I think you need to combine this into one big 
run_job
task, and then LOOP there. You can just do 
trigger_job.run()
and 
wait_to_complete.run()
inside the big task, but they will just be like functions instead of Prefect tasks.
c

Casey Green

08/24/2021, 3:54 PM
gotcha, that makes sense since this wouldn't be a valid DAG! thanks 🙏
If I were to make this a flow instead, is it possible to automatically trigger retries at the flow level? e.g.
Copy code
with Flow("test") as flow:
    flow_name = Parameter("flow_name", required=True)
    handle = trigger_job(job_name)
    result = wait_to_complete(handle)
    max_runs = 3

    # maybe this isn't a case... perhaps a terminal_state_handler that inspects the result and sets the state appropriately?
    with case(should_re_run(result), True):
        # trigger flow failure, but automatically retry.
Kinda looks and feels like a code smell 🤷‍♂️
I guess really what I'm trying to do is to retry an entire sub-branch of a DAG
which maybe is a weird/invalid thing to want to do 🤷‍♂️
k

Kevin Kho

08/24/2021, 4:39 PM
We don’t have retries at the Flow run level but if it is invoked through a 
StartFlowRun
task or 
create_flow_run
task, then we have retries, but the concept of a flow retry is not definitive because some users expect everything to run (even successful tasks) and some expect to run from where it left off. I think retries on these two Prefect tasks can be used for both scenarios by supplying a 
idempotency_key
c

Casey Green

08/24/2021, 5:00 PM
oh interesting, I'll take a look at this task. this might work
Powered by