@Marvin in Prefect 2.X is it possible to trigger a flow run of another flow and detach? I'd like to trigger another flow with parameters from another flow but I don't want to block until the flow is finished, I want to exit after the flow has been triggered.

In Prefect 2.x, you can indeed trigger a flow run from another flow and detach it, meaning you don't have to wait for the triggered flow to complete before continuing with the original flow. This is achieved using the

create_flow_run

function from the Prefect Client. Here's a basic example:

python
from prefect import flow, task
from prefect.client import Client

@task
def trigger_flow():
    client = Client()
    flow_run_id = client.create_flow_run(
        flow_name="your_target_flow_name",
        parameters={"param1": "value1", "param2": "value2"}
    )
    return flow_run_id

@flow
def main_flow():
    trigger_flow()

if __name__ == "__main__":
    main_flow()

In this example,

create_flow_run

is used to start another flow with specified parameters. The flow run ID is returned, but the main flow does not wait for this flow to complete, allowing it to exit immediately after triggering the other flow. For more detailed information, you can refer to the Prefect 2.x documentation on flows.

@Marvin

from prefect.client import Client

leads to an import error. Also, the method

create_flow_run

does not support passing a

flow_name

parameter.

It seems there was a mistake in my previous response. The

Client

class does not exist in Prefect 2.x, and the method

create_flow_run

does not support a

flow_name

parameter directly. Instead, you should use the

get_client

function and work with deployments to trigger another flow. Here's the corrected approach: 1. Use `get_client`: This function provides access to the Prefect client. 2. Trigger a flow run using a deployment: You need to create a deployment for the flow you want to trigger. Once you have a deployment, you can trigger a flow run using the deployment ID. Here's a basic example:

python
from prefect import flow, get_client

@flow
async def trigger_another_flow():
    async with get_client() as client:
        deployment_id = "your-deployment-id"
        flow_run = await client.create_flow_run_from_deployment(deployment_id=deployment_id, parameters={"param1": "value1"})
        print(f"Triggered flow run: {flow_run.id}")

@flow
def main_flow():
    trigger_another_flow()

if __name__ == "__main__":
    main_flow()

In this example, replace

"your-deployment-id"

with the actual deployment ID of the flow you want to trigger. This method triggers the flow asynchronously and exits immediately, not waiting for the triggered flow to complete. For more details, you can refer to the Prefect 2.x documentation on using the client.