@Marvin I have a flow which triggers a sub flows. I would like my sub flow to always run even if the upstream sub flow was failed. how can I do it?

In Prefect, a subflow run is created when a flow function is called inside the execution of another flow. The primary flow is the "parent" flow and the flow created within the parent is the "child" flow or "subflow". Subflows behave like normal flow runs and will block execution of the parent flow until completion. However, asynchronous subflows can be run in parallel by using AnyIO task groups or asyncio.gather. To ensure that a subflow always runs even if the upstream subflow fails, you can use the

skip_on_upstream_skip

parameter in the

Flow

class. This parameter, when set to

False

, will ensure that the subflow is always run, regardless of the status of the upstream tasks. Here is an example:

from prefect import Flow, task

@task
def upstream_task():
    # This task might fail
    pass

@task
def subflow_task():
    # This task should always run
    pass

with Flow("My Flow") as flow:
    upstream_task = upstream_task()
    subflow_task = subflow_task(skip_on_upstream_skip=False)

flow.run()

In this example,

subflow_task

will always run, even if

upstream_task

fails.