@Marvin In my Flow I call 2 subflows. How can I create a dependency between the 2 so that in the flow run ui there is a dependency arrow from the first subflow to the second subflow?

To create a dependency between two subflows in Prefect 3.x, you need to ensure that the second subflow is called as a result of the first subflow's completion. This effectively creates a dependency arrow in the flow run UI. Here's a simple way to achieve this using an imperative programming style:

python
from prefect import flow

@flow
def first_subflow():
    # Your logic for the first subflow
    pass

@flow
def second_subflow():
    # Your logic for the second subflow
    pass

@flow
def main_flow():
    # Execute the first subflow
    first_result = first_subflow()
    
    # Use the result of the first subflow to trigger the second subflow
    if first_result is not None:
        second_subflow()

if __name__ == "__main__":
    main_flow()

This code ensures that

second_subflow

is only called after

first_subflow

completes, creating a dependency. For more details, you can explore Prefect's flow documentation: Prefect Flows. If you need more advanced dependency management, consider using tags or other Prefect features like triggers.

@Marvin this does not seem to work. Can this be done with upstreamtasks?

Creating a dependency between subflows using

upstream_tasks

is indeed a feasible approach in Prefect. The

upstream_tasks

parameter can be used to define dependencies explicitly between tasks or subflows within the same flow run. Here's how you can structure your flow with this concept:

python
from prefect import flow, task, get_run_logger

@task
def task_one():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Running task one")
    return "result from task one"

@task
def task_two():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Running task two")

@flow
def first_subflow():
    result = task_one()
    return result

@flow
def second_subflow(upstream_result):
    task_two(upstream_result)

@flow
def main_flow():
    # Run the first subflow and store its result
    first_result = first_subflow()
    
    # Pass the result of the first subflow to the second subflow
    second_subflow(first_result)

if __name__ == "__main__":
    main_flow()

In this setup,

second_subflow

depends on the completion of

first_subflow

, as it takes the result of

first_subflow

as an input, thus establishing an explicit dependency. For more detailed information, you can refer to Prefect's tasks documentation.