如何使用Prefect API从已创建的部署中触发Flow运行？

你需要编写一个接收部署名称或ID的函数来执行Prefect Flow，但使用create_flow_run_from_deployment时持续报错。原代码如下：

async def run_prefect_flow(flow_name: str):
    try:
        async with get_client() as client:
            deployment = await client.read_deployment_by_name(flow_name)
            flow_run = await client.create_flow_run_from_deployment(deployment.id)
            return flow_run
    except PrefectHTTPStatusError as e:
        print(f"Ocorreu um erro ao tentar executar a flow {flow_name}: {e}")
        return None

可能的错误原因及修正方案 #

1. 部署名称格式错误
   Prefect部署的完整名称格式为{flow-name}/{deployment-name}，仅传入my-deployment-test可能无法匹配到部署。可以通过prefect deployment ls命令查看所有部署的完整名称。

2. 无需手动查询部署ID
   create_flow_run_from_deployment支持直接传入部署名称或ID作为参数，无需先调用read_deployment_by_name获取ID，简化代码同时避免额外的查询错误。

3. 异步函数调用方式错误
   若在同步环境中调用该异步函数，必须使用asyncio.run()来触发执行，否则异步逻辑无法正常运行。

修正后的代码 #

import asyncio
from prefect import get_client
from prefect.exceptions import PrefectHTTPStatusError

async def run_prefect_flow(deployment_identifier: str):
    try:
        async with get_client() as client:
            # 直接传入部署名称或ID创建Flow运行
            flow_run = await client.create_flow_run_from_deployment(
                deployment_id_or_name=deployment_identifier
            )
            return flow_run
    except PrefectHTTPStatusError as e:
        print(f"执行部署 {deployment_identifier} 时出错: {e}")
        return None

# 同步环境下的调用示例
if __name__ == "__main__":
    # 替换为你的部署完整名称或ID
    flow_run = asyncio.run(run_prefect_flow("my-flow/my-deployment-test"))
    if flow_run:
        print(f"Flow运行已创建，ID: {flow_run.id}")

额外说明 #

• 如果使用部署ID，直接传入ID字符串即可，函数会自动识别类型。
• 若在异步上下文（如FastAPI、Starlette路由）中调用，无需使用asyncio.run()，直接await run_prefect_flow(...)即可。

内容的提问来源于stack exchange，提问作者Victor Longui