<@ULVA73B9P> is there a way to deploy deployment y...
# marvin-ai
a
@Marvin is there a way to deploy deployment yaml files programmatically (without the CLI)?
m
thought for 174.0 seconds
Short answer: Yes. In Prefect 3.x there isn’t a single public “apply prefect.yaml” function, but you have two solid options to do this programmatically without invoking the CLI. 1) Use the same internal helpers the CLI uses (most faithful to prefect.yaml) - Pros: Honors build/push/pull steps, schedules, triggers, etc., exactly like
prefect deploy
- Cons: These are private modules (underscore-prefixed) and may change between releases Example (deploy everything defined in prefect.yaml):
Copy code
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:
Copy code
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.