Hi, Do you have example of create_flow_run_from_deployment, wait for completion of flow run before continue to next line of code execution in 2.0? I was using create_flow_run and wait_for_flow_run in 1.0

Hello Patrick, here is an example for creating a flow run from deployment: https://discourse.prefect.io/t/how-to-create-a-flow-run-from-deployment-orchestrator-pattern/803

I tried this example, it does not wait for completion. It is running async mode. I want it to wait for flow run to be completed. My 1.0 code looks like this:

flow_run_ids = create_flow_run(
        flow_name=unmapped("LiveLots-ETL"),
        project_name=unmapped(f"{prefect_project}"),
        run_name=unmapped("Flow Test-LiveLots-ETL"),
        idempotency_key=datetime.now().strftime("%m/%d/%Y:%H:%M:%S"),
        parameters={"config_file": config_file, "bucket": bucket, "prefix": prefix},
    )



    wait_for_flow =
   wait_for_flow_run(flow_run_ids, raise_final_state=unmapped(True))

Ah, understood. One moment while we do some research on our end to troubleshoot this. We also received your email, but we can continue the conversation here if that works for you.

Yes this works as well

I think that this article may be helpful for you to accomplish what you have in the 1.0 example, in the context of 2.0. In 2.0, subflows can be called from the parent flow upon execution. This can be set up so that downstream subflows are dependent on the state of those that are upstream. Note that in Orion, subflows are run sequentially by default. In addition, if you'd like to pass different parameters to the subflows, you could create a for loop for your parent flow which passes in parameters to the subflows at runtime. Here is an example of how to accomplish this:

from prefect import flow

@flow(name = "Flow A")
def flow_A(params):
    print(f"I am flow A, and I received these parameters: {params}")

@flow(name = "Flow B")
def flow_b():
    print(f"I am flow B, I run after Flow A")

params_list = [{"config_file": "config_file", "bucket": "bucket", "prefix": "prefix"},
                {"config_file": "config_file2", "bucket2": "bucket2", "prefix": "prefix2"},
                {"config_file": "config_file3", "bucket2": "bucket3", "prefix": "prefix3"}]

@flow(name = "Parent Flow")
def parent_flow(params_list):
    for params in params_list:
        flow_A(params)
    #Runs after Flow A is done
    flow_b()

parent_flow(params_list)

I tried this method and it works. However, what I really want is to run a deployed flow and wait for completion. Is this possible in 2.0? If not, will it be in future release?

Hey @Patrick Tan The short answer to this questions is yes, we'll be expanding on orchestration patters more in the future, that said you could probably accomplish this with native python:

@flow
def p_flow():
    flow_a = create_flow_run_from_deployment()
    flow_a_state = None

    while flow_a_state != "Completed":
        time.sleep(sometimeinseconds)
           flow_a_state = task_to_get_flow_run_state(flow_a)

Not a working example but more of a conceptual idea, one of the benefits of 2.0 is that if it's accomplishable in native python there's likely a way to address it in 2.0