Lester Cayabyab
09/16/2025, 2:05 PMMarvin
09/16/2025, 2:05 PMMarvin
09/16/2025, 2:06 PMfrom prefect.client.orchestration import get_client
# name must be "flow_name/deployment_name", e.g., "etl-flow/prod"
async def get_deployment_by_name(name: str):
async with get_client() as client:
return await client.read_deployment_by_name(name)
# Convenience wrapper if you have names separately
async def get_deployment(flow_name: str, deployment_name: str):
return await get_deployment_by_name(f"{flow_name}/{deployment_name}")
Sync version:
from prefect.client.orchestration import SyncPrefectClient
def get_deployment_by_name_sync(name: str):
with SyncPrefectClient() as client:
return client.read_deployment_by_name(name)
Usage example:
deployment = await get_deployment_by_name("my-flow/prod")
print(deployment.id, deployment.name)
If you’d prefer a version that handles “not found” more gracefully, I can add try/except around the call—just say the behavior you want.Lester Cayabyab
09/16/2025, 2:09 PMMarvin
09/16/2025, 2:09 PMMarvin
09/16/2025, 2:10 PMread_deployment_by_name("flow_name/deployment_name")
maps to that REST endpoint.