Matthew Seligson

    Matthew Seligson

    7 months ago
    My flow A has a StartFlowRun task that runs flow B. If I’m looking at a flow run of flow A in the UI, how can I navigate to the corresponding run of flow B?
    Anna Geller

    Anna Geller

    7 months ago
    There is not a straightforward way to do it in Prefect 1.0. One way to navigate to a child flow run is through logs: within your parent flow run logs, you should see something like this:
    Flow run created: <https://cloud.prefect.io/anna-prefect/flow-run/d3f27dd8-1e5e-40af-94cf-46766ef080f3>
    Having said that, we know this link between parent and child flow run could be improved and Orion provides a first-class way to navigate between parent and child flows. If you are interested to learn more, check out this livestream recording:

    https://www.youtube.com/watch?v=w4mc43X27kE&amp;t=2613s

    What you can also do is adding stream_logs=True to see child flow run logs directly in the parent flow run logs.
    from prefect import Flow
    from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
    
    with Flow("parent_flow") as flow:
        child_flow_run_id = create_flow_run(
            flow_name="child_flow_name", run_name="custom_run_name"
        )
        child_flowrunview = wait_for_flow_run(
            child_flow_run_id, raise_final_state=True, stream_logs=True
        )
    But note that this is not available in the StartFlowRun task. For more, see this Discourse topic
    Matthew Seligson

    Matthew Seligson

    7 months ago
    Can I only wait for my subflow to complete in Orion?
    Kevin Kho

    Kevin Kho

    7 months ago
    You mean subflow in Orion but main flow in current Prefect?
    Anna Geller

    Anna Geller

    7 months ago
    When using the default SequentialTaskRunner, Orion actually waits till your child flow run completes before starting any downstream tasks/subflows, try this example:
    from prefect import flow
    import time
    
    
    @flow
    def subflow(x):
        print("I am a subflow - waiting for 5 seconds to show some processing!")
        time.sleep(5)
        return x
    
    
    @flow
    def second_subflow(x):
        print("I am a second subflow!")
        return x
    
    
    @flow
    def main_flow():
        data = subflow(5)
        second_subflow(data)
    
    
    if __name__ == "__main__":
        main_flow()