Thread
#prefect-server
    Casey Green

    Casey Green

    1 year ago
    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:
    Kevin Kho

    Kevin Kho

    1 year ago
    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.
    Casey Green

    Casey Green

    1 year ago
    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.
    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 🤷‍♂️
    Kevin Kho

    Kevin Kho

    1 year ago
    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
    Casey Green

    Casey Green

    1 year ago
    oh interesting, I'll take a look at this task. this might work