Aniket Rawat
04/02/2026, 5:46 AMMarvin
04/02/2026, 5:46 AMMarvin
04/02/2026, 5:49 AMprefect deploy
- Cons: These are private modules (underscore-prefixed) and may change between releases
Example (deploy everything defined in prefect.yaml):
import asyncio
from pathlib import Path
from rich.console import Console
from prefect.cli.deploy._config import _load_deploy_configs_and_actions
from prefect.cli.deploy._core import _run_single_deploy
async def deploy_from_yaml(yaml_path: Path = Path("prefect.yaml")):
console = Console()
deploy_configs, actions = _load_deploy_configs_and_actions(
prefect_file=yaml_path,
console=console,
)
for deploy_config in deploy_configs:
await _run_single_deploy(
deploy_config=deploy_config,
actions=actions,
prefect_file=yaml_path,
console=console,
is_interactive=lambda: False,
)
if __name__ == "__main__":
asyncio.run(deploy_from_yaml())
2) Use the public API to create/update deployments from code (bypass YAML)
- Pros: Public, stable surface; no private imports
- Cons: You’ll need to read your YAML and map fields onto the deployment yourself
Minimal example:
import yaml
from pathlib import Path
from prefect.flows import load_flow_from_entrypoint
from prefect.deployments.runner import RunnerDeployment
def deploy_from_yaml_manually(yaml_path: Path = Path("prefect.yaml"), name: str | None = None):
data = yaml.safe_load(yaml_path.read_text())
deployments = data.get("deployments", [])
if not deployments:
raise ValueError("No deployments found in prefect.yaml")
# Optionally select a specific deployment by name
cfg = next((d for d in deployments if d.get("name") == name), deployments[0])
flow = load_flow_from_entrypoint(cfg["entrypoint"])
deployment = RunnerDeployment(
name=cfg["name"],
flow_name=flow.name,
entrypoint=cfg.get("entrypoint"),
work_pool_name=(cfg.get("work_pool") or {}).get("name"),
work_queue_name=(cfg.get("work_pool") or {}).get("work_queue_name"),
parameters=cfg.get("parameters", {}),
description=cfg.get("description") or flow.description,
version=cfg.get("version"),
tags=cfg.get("tags") or [],
job_variables=(cfg.get("work_pool") or {}).get("job_variables") or {},
schedules=cfg.get("schedules"),
concurrency_limit=cfg.get("concurrency_limit"),
paused=cfg.get("paused"),
triggers=cfg.get("triggers") or [],
)
deployment_id = deployment.apply(schedules=cfg.get("schedules"))
return deployment_id
if __name__ == "__main__":
deploy_from_yaml_manually()
Notes:
- If you need build/push/pull steps from prefect.yaml (e.g., building/pushing a Docker image), prefer option 1 so those steps run exactly as the CLI would.
- If you don’t need those steps, option 2 is a clean, public approach.
- For new work in Prefect 3.x, you can also define deployments directly in code with flow.from_source(...).deploy(...), skipping prefect.yaml entirely.
If you can share whether you need build/push/pull steps honored from the YAML, I can tailor the snippet to your exact use case.