feat : Add SDK support for YAML-based deployments via deploy_from_yaml

src/prefect/deployments/__init__.py

@@ -1,21 +1,22 @@
 from typing import TYPE_CHECKING
 from prefect._internal.compatibility.migration import getattr_migration
 
-
 if TYPE_CHECKING:
     from .flow_runs import arun_deployment, run_deployment
     from .base import initialize_project
     from .runner import deploy
+    from .yaml import deploy_from_yaml
 
 _public_api: dict[str, tuple[str, str]] = {
     "initialize_project": (__spec__.parent, ".base"),
     "arun_deployment": (__spec__.parent, ".flow_runs"),
     "run_deployment": (__spec__.parent, ".flow_runs"),
     "deploy": (__spec__.parent, ".runner"),
+    "deploy_from_yaml": (__spec__.parent, ".yaml"),
 }
 
 # Declare API for type-checkers
-__all__ = ["initialize_project", "deploy", "arun_deployment", "run_deployment"]
+__all__ = ["initialize_project", "deploy", "arun_deployment", "run_deployment","deploy_from_yaml",]
 
 
 def __getattr__(attr_name: str) -> object:

src/prefect/deployments/yaml.py

@@ -0,0 +1,72 @@
+from pathlib import Path
+from typing import List
+from uuid import UUID
+
+
+async def deploy_from_yaml(path: str) -> List[UUID]:
+    """
+    Deploy flows defined in a prefect.yaml file via the SDK.
+
+    Args:
+        path: Path to the prefect.yaml file.
+
+    Returns:
+        List of UUIDs for the created/updated deployments.
+
+    Example:
+        import asyncio
+        from prefect.deployments import deploy_from_yaml
+
+        asyncio.run(deploy_from_yaml("prefect.yaml"))
+    """
+    from prefect.cli.deploy._config import (
+        _load_deploy_configs_and_actions,
+        _pick_deploy_configs,
+    )
+    from prefect.cli.deploy._core import _run_multi_deploy, _run_single_deploy
+
+    yaml_path = Path(path)
+    if not yaml_path.exists():
+        raise FileNotFoundError(f"No prefect.yaml found at: {path}")
+
+    class _SilentConsole:
+        """Suppresses Rich console output when deploying via SDK."""
+        def print(self, *args, **kwargs):
+            pass
+
+    console = _SilentConsole()
+
+    all_deploy_configs, actions = _load_deploy_configs_and_actions(
+        prefect_file=yaml_path,
+        console=console,
+    )
+
+    deploy_configs = _pick_deploy_configs(
+        all_deploy_configs,
+        names=[],
+        deploy_all=True,
+        console=console,
+        is_interactive=False,
+    )
+
+    if not deploy_configs:
+        raise ValueError("No deployments found in prefect.yaml")
+
+    if len(deploy_configs) > 1:
+        await _run_multi_deploy(
+            deploy_configs=deploy_configs,
+            actions=actions,
+            deploy_all=True,
+            prefect_file=yaml_path,
+            console=console,
+            is_interactive=False,
+        )
+    else:
+        await _run_single_deploy(
+            deploy_config=deploy_configs[0],
+            actions=actions,
+            options={},
+            prefect_file=yaml_path,
+            console=console,
+            is_interactive=False,
+        )

src/prefect/testing/test_yaml_deploy.py

@@ -0,0 +1,79 @@
+import pytest
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+
+@pytest.fixture
+def prefect_yaml(tmp_path):
+    content = """
+deployments:
+  - name: test-deployment
+    entrypoint: myflow.py:my_flow
+    work_pool:
+      name: my-pool
+"""
+    f = tmp_path / "prefect.yaml"
+    f.write_text(content)
+    return str(f)
+
+
+@pytest.mark.asyncio
+async def test_deploy_from_yaml_file_not_found():
+    from prefect.deployments.yaml import deploy_from_yaml
+    with pytest.raises(FileNotFoundError, match="No prefect.yaml found"):
+        await deploy_from_yaml("nonexistent.yaml")
+
+
+@pytest.mark.asyncio
+async def test_deploy_from_yaml_no_deployments(tmp_path):
+    f = tmp_path / "prefect.yaml"
+    f.write_text("deployments: []")
+    with patch(
+        "prefect.cli.deploy._config._load_deploy_configs_and_actions",
+        return_value=([], []),
+    ), patch(
+        "prefect.cli.deploy._config._pick_deploy_configs",
+        return_value=[],
+    ):
+        from prefect.deployments.yaml import deploy_from_yaml
+        with pytest.raises(ValueError, match="No deployments found"):
+            await deploy_from_yaml(str(f))
+
+
+@pytest.mark.asyncio
+async def test_deploy_from_yaml_single(prefect_yaml):
+    mock_config = {"name": "test-deployment", "entrypoint": "myflow.py:my_flow"}
+    with patch(
+        "prefect.cli.deploy._config._load_deploy_configs_and_actions",
+        return_value=([mock_config], []),
+    ), patch(
+        "prefect.cli.deploy._config._pick_deploy_configs",
+        return_value=[mock_config],
+    ), patch(
+        "prefect.cli.deploy._core._run_single_deploy",
+        new_callable=AsyncMock,
+    ) as mock_single:
+        from prefect.deployments.yaml import deploy_from_yaml
+        await deploy_from_yaml(prefect_yaml)
+        mock_single.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_deploy_from_yaml_multi(prefect_yaml):
+    mock_configs = [
+        {"name": "deployment-1", "entrypoint": "myflow.py:flow_one"},
+        {"name": "deployment-2", "entrypoint": "myflow.py:flow_two"},
+    ]
+    with patch(
+        "prefect.cli.deploy._config._load_deploy_configs_and_actions",
+        return_value=(mock_configs, []),
+    ), patch(
+        "prefect.cli.deploy._config._pick_deploy_configs",
+        return_value=mock_configs,
+    ), patch(
+        "prefect.cli.deploy._core._run_multi_deploy",
+        new_callable=AsyncMock,
+    ) as mock_multi:
+        from prefect.deployments.yaml import deploy_from_yaml
+        await deploy_from_yaml(prefect_yaml)
+        mock_multi.assert_called_once()