Nicholas Pini
10/01/2025, 2:14 PMMarvin
10/01/2025, 2:14 PMMarvin
10/01/2025, 2:18 PMfrom prefect.client.orchestration import get_client
from prefect.exceptions import ObjectNotFound
async def deployment_exists(flow_name: str, deployment_name: str) -> bool:
full_name = f"{flow_name}/{deployment_name}" # format: flow/deployment
async with get_client() as client:
try:
await client.read_deployment_by_name(full_name)
return True
except ObjectNotFound:
return False
Synchronous (if you prefer blocking I/O):
from prefect.client.orchestration import SyncPrefectClient
from prefect.exceptions import ObjectNotFound
def deployment_exists_sync(flow_name: str, deployment_name: str) -> bool:
full_name = f"{flow_name}/{deployment_name}"
with SyncPrefectClient() as client:
try:
client.read_deployment_by_name(full_name)
return True
except ObjectNotFound:
return False
If you already have the deployment ID:
from prefect.client.orchestration import get_client
from prefect.exceptions import ObjectNotFound
async def deployment_exists_by_id(deployment_id: str) -> bool:
async with get_client() as client:
try:
await client.read_deployment(deployment_id)
return True
except ObjectNotFound:
return False
Notes and docs:
- Name format is "<flow_name>/<deployment_name>" (workspace is implied by your active profile): Read deployment by name
- Read by ID: Read deployment
- List deployments (if you prefer listing and filtering): List deploymentsBring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.
Powered by