Repository: PrefectHQ/prefect
Issue category: Feature Request

Remove `prefect deployment apply` CLI from the `main` branch
Remove the `prefect.cli.deployment.apply` function and any references to it. Any private code that was used solely for that function should also be deleted. This CLI command will be included in 2.x releases, but will not be included in the 3.0 release.


Functions to edit:
src/prefect/cli/deployment.py:apply

Patch:
diff --git a/src/prefect/cli/deployment.py b/src/prefect/cli/deployment.py
index 398fd457653a..401c3dc80d70 100644
--- a/src/prefect/cli/deployment.py
+++ b/src/prefect/cli/deployment.py
@@ -22,7 +22,7 @@
 from prefect.cli._types import PrefectTyper
 from prefect.cli._utilities import exit_with_error, exit_with_success
 from prefect.cli.root import app
-from prefect.client.orchestration import ServerType, get_client
+from prefect.client.orchestration import get_client
 from prefect.client.schemas.filters import FlowFilter
 from prefect.client.schemas.objects import DeploymentSchedule
 from prefect.client.schemas.schedules import (
@@ -32,7 +32,7 @@
 )
 from prefect.client.utilities import inject_client
 from prefect.context import PrefectObjectRegistry, registry_from_script
-from prefect.deployments import Deployment, load_deployments_from_yaml
+from prefect.deployments import load_deployments_from_yaml
 from prefect.exceptions import (
     ObjectAlreadyExists,
     ObjectNotFound,
@@ -1103,135 +1103,6 @@ def _load_deployments(path: Path, quietly=False) -> PrefectObjectRegistry:
     return specs
 
 
-@deployment_app.command(
-    deprecated=True,
-    deprecated_start_date="Mar 2024",
-    deprecated_name="deployment apply",
-    deprecated_help="Use 'prefect deploy' to deploy flows via YAML instead.",
-)
-async def apply(
-    paths: List[str] = typer.Argument(
-        ...,
-        help="One or more paths to deployment YAML files.",
-    ),
-    upload: bool = typer.Option(
-        False,
-        "--upload",
-        help=(
-            "A flag that, when provided, uploads this deployment's files to remote"
-            " storage."
-        ),
-    ),
-    work_queue_concurrency: int = typer.Option(
-        None,
-        "--limit",
-        "-l",
-        help=(
-            "Sets the concurrency limit on the work queue that handles this"
-            " deployment's runs"
-        ),
-    ),
-):
-    """
-    Create or update a deployment from a YAML file.
-    """
-    deployment = None
-    async with get_client() as client:
-        for path in paths:
-            try:
-                deployment = await Deployment.load_from_yaml(path)
-                app.console.print(
-                    f"Successfully loaded {deployment.name!r}", style="green"
-                )
-            except Exception as exc:
-                exit_with_error(
-                    f"'{path!s}' did not conform to deployment spec: {exc!r}"
-                )
-
-            assert deployment
-
-            await create_work_queue_and_set_concurrency_limit(
-                deployment.work_queue_name,
-                deployment.work_pool_name,
-                work_queue_concurrency,
-            )
-
-            if upload:
-                if (
-                    deployment.storage
-                    and "put-directory" in deployment.storage.get_block_capabilities()
-                ):
-                    file_count = await deployment.upload_to_storage()
-                    if file_count:
-                        app.console.print(
-                            (
-                                f"Successfully uploaded {file_count} files to"
-                                f" {deployment.location}"
-                            ),
-                            style="green",
-                        )
-                else:
-                    app.console.print(
-                        (
-                            f"Deployment storage {deployment.storage} does not have"
-                            " upload capabilities; no files uploaded."
-                        ),
-                        style="red",
-                    )
-            await check_work_pool_exists(
-                work_pool_name=deployment.work_pool_name, client=client
-            )
-
-            if client.server_type != ServerType.CLOUD and deployment.triggers:
-                app.console.print(
-                    (
-                        "Deployment triggers are only supported on "
-                        f"Prefect Cloud. Triggers defined in {path!r} will be "
-                        "ignored."
-                    ),
-                    style="red",
-                )
-
-            deployment_id = await deployment.apply()
-            app.console.print(
-                (
-                    f"Deployment '{deployment.flow_name}/{deployment.name}'"
-                    f" successfully created with id '{deployment_id}'."
-                ),
-                style="green",
-            )
-
-            if PREFECT_UI_URL:
-                app.console.print(
-                    "View Deployment in UI:"
-                    f" {PREFECT_UI_URL.value()}/deployments/deployment/{deployment_id}"
-                )
-
-            if deployment.work_pool_name is not None:
-                await _print_deployment_work_pool_instructions(
-                    work_pool_name=deployment.work_pool_name, client=client
-                )
-            elif deployment.work_queue_name is not None:
-                app.console.print(
-                    "\nTo execute flow runs from this deployment, start an agent that"
-                    f" pulls work from the {deployment.work_queue_name!r} work queue:"
-                )
-                app.console.print(
-                    f"$ prefect agent start -q {deployment.work_queue_name!r}",
-                    style="blue",
-                )
-            else:
-                app.console.print(
-                    (
-                        "\nThis deployment does not specify a work queue name, which"
-                        " means agents will not be able to pick up its runs. To add a"
-                        " work queue, edit the deployment spec and re-run this command,"
-                        " or visit the deployment in the UI."
-                    ),
-                    style="red",
-                )
-
-
 @deployment_app.command()
 async def delete(
     name: Optional[str] = typer.Argument(

Test patch: diff --git a/tests/cli/deployment/test_deployment_cli.py b/tests/cli/deployment/test_deployment_cli.py
index 57c19166a434..7b79206102b8 100644
--- a/tests/cli/deployment/test_deployment_cli.py
+++ b/tests/cli/deployment/test_deployment_cli.py
@@ -5,12 +5,10 @@
 
 from prefect import flow
 from prefect.client.orchestration import PrefectClient
-from prefect.deployments import Deployment
 from prefect.server.schemas.filters import DeploymentFilter, DeploymentFilterId
 from prefect.server.schemas.schedules import IntervalSchedule
 from prefect.settings import (
     PREFECT_API_SERVICES_TRIGGERS_ENABLED,
-    PREFECT_UI_URL,
     temporary_settings,
 )
 from prefect.testing.cli import invoke_and_assert
@@ -32,234 +30,6 @@ def fn():
     return fn
 
 
-def test_deployment_apply_prints_deprecation_warning(tmp_path, patch_import):
-    Deployment.build_from_flow(
-        flow=my_flow,
-        name="TEST",
-        flow_name="my_flow",
-        output=str(tmp_path / "test.yaml"),
-        work_queue_name="prod",
-    )
-
-    invoke_and_assert(
-        [
-            "deployment",
-            "apply",
-            str(tmp_path / "test.yaml"),
-        ],
-        temp_dir=tmp_path,
-        expected_output_contains=(
-            "WARNING: The 'deployment apply' command has been deprecated.",
-            "It will not be available after Sep 2024.",
-            "Use 'prefect deploy' to deploy flows via YAML instead.",
-        ),
-    )
-
-
-class TestOutputMessages:
-    def test_message_with_work_queue_name_from_python_build(
-        self, patch_import, tmp_path
-    ):
-        d = Deployment.build_from_flow(
-            flow=my_flow,
-            name="TEST",
-            flow_name="my_flow",
-            output=str(tmp_path / "test.yaml"),
-            work_queue_name="prod",
-        )
-        invoke_and_assert(
-            [
-                "deployment",
-                "apply",
-                str(tmp_path / "test.yaml"),
-            ],
-            expected_output_contains=[
-                (
-                    "To execute flow runs from this deployment, start an agent "
-                    f"that pulls work from the {d.work_queue_name!r} work queue:"
-                ),
-                f"$ prefect agent start -q {d.work_queue_name!r}",
-            ],
-        )
-
-    def test_message_with_prefect_agent_work_pool(
-        self, patch_import, tmp_path, prefect_agent_work_pool
-    ):
-        Deployment.build_from_flow(
-            flow=my_flow,
-            name="TEST",
-            flow_name="my_flow",
-            output=str(tmp_path / "test.yaml"),
-            work_pool_name=prefect_agent_work_pool.name,
-        )
-        invoke_and_assert(
-            [
-                "deployment",
-                "apply",
-                str(tmp_path / "test.yaml"),
-            ],
-            expected_output_contains=[
-                (
-                    "To execute flow runs from this deployment, start an agent that"
-                    f" pulls work from the {prefect_agent_work_pool.name!r} work pool:"
-                ),
-                f"$ prefect agent start -p {prefect_agent_work_pool.name!r}",
-            ],
-        )
-
-    def test_message_with_process_work_pool(
-        self, patch_import, tmp_path, process_work_pool
-    ):
-        Deployment.build_from_flow(
-            flow=my_flow,
-            name="TEST",
-            flow_name="my_flow",
-            output=str(tmp_path / "test.yaml"),
-            work_pool_name=process_work_pool.name,
-        )
-        invoke_and_assert(
-            [
-                "deployment",
-                "apply",
-                str(tmp_path / "test.yaml"),
-            ],
-            expected_output_contains=[
-                (
-                    "To execute flow runs from this deployment, start a worker "
-                    f"that pulls work from the {process_work_pool.name!r} work pool:"
-                ),
-                f"$ prefect worker start -p {process_work_pool.name!r}",
-            ],
-        )
-
-    def test_message_with_process_work_pool_without_workers_enabled(
-        self, patch_import, tmp_path, process_work_pool, disable_workers
-    ):
-        Deployment.build_from_flow(
-            flow=my_flow,
-            name="TEST",
-            flow_name="my_flow",
-            output=str(tmp_path / "test.yaml"),
-            work_pool_name=process_work_pool.name,
-        )
-        invoke_and_assert(
-            [
-                "deployment",
-                "apply",
-                str(tmp_path / "test.yaml"),
-            ],
-            expected_output_contains=[
-                (
-                    "\nTo execute flow runs from this deployment, please enable "
-                    "the workers CLI and start a worker that pulls work from the "
-                    f"{process_work_pool.name!r} work pool:"
-                ),
-                (
-                    "$ prefect config set PREFECT_EXPERIMENTAL_ENABLE_WORKERS=True\n"
-                    f"$ prefect worker start -p {process_work_pool.name!r}"
-                ),
-            ],
-        )
-
-    def test_linking_to_deployment_in_ui(
-        self,
-        patch_import,
-        tmp_path,
-        monkeypatch,
-    ):
-        with temporary_settings({PREFECT_UI_URL: "http://foo/bar"}):
-            Deployment.build_from_flow(
-                flow=my_flow,
-                name="TEST",
-                flow_name="my_flow",
-                output=str(tmp_path / "test.yaml"),
-                work_queue_name="prod",
-            )
-            invoke_and_assert(
-                [
-                    "deployment",
-                    "apply",
-                    str(tmp_path / "test.yaml"),
-                ],
-                expected_output_contains="http://foo/bar/deployments/deployment/",
-            )
-
-    def test_updating_work_queue_concurrency_from_python_build(
-        self, patch_import, tmp_path
-    ):
-        Deployment.build_from_flow(
-            flow=my_flow,
-            name="TEST",
-            flow_name="my_flow",
-            output=str(tmp_path / "test.yaml"),
-            work_queue_name="prod",
-        )
-        invoke_and_assert(
-            [
-                "deployment",
-                "apply",
-                str(tmp_path / "test.yaml"),
-                "-l",
-                "42",
-            ],
-            expected_output_contains=[
-                "Updated concurrency limit on work queue 'prod' to 42",
-            ],
-        )
-
-    def test_message_with_missing_work_queue_name(self, patch_import, tmp_path):
-        Deployment.build_from_flow(
-            flow=my_flow,
-            name="TEST",
-            flow_name="my_flow",
-            output=str(tmp_path / "test.yaml"),
-            work_queue_name=None,
-        )
-        invoke_and_assert(
-            [
-                "deployment",
-                "apply",
-                str(tmp_path / "test.yaml"),
-            ],
-            expected_output_contains=(
-                (
-                    "This deployment does not specify a work queue name, which means"
-                    " agents will not be able to pick up its runs. To add a work queue,"
-                    " edit the deployment spec and re-run this command, or visit the"
-                    " deployment in the UI."
-                ),
-            ),
-        )
-
-    def test_message_with_missing_nonexistent_work_pool(self, patch_import, tmp_path):
-        Deployment.build_from_flow(
-            flow=my_flow,
-            name="TEST",
-            flow_name="my_flow",
-            output=str(tmp_path / "test.yaml"),
-            work_pool_name="gibberish",
-            schedule=IntervalSchedule(interval=60),
-        )
-        invoke_and_assert(
-            [
-                "deployment",
-                "apply",
-                str(tmp_path / "test.yaml"),
-            ],
-            expected_code=1,
-            expected_output_contains=(
-                [
-                    (
-                        "This deployment specifies a work pool name of 'gibberish', but"
-                        " no such work pool exists."
-                    ),
-                    "To create a work pool via the CLI:",
-                    "$ prefect work-pool create 'gibberish'",
-                ]
-            ),
-        )
-
-
 class TestDeploymentSchedules:
     @pytest.fixture(autouse=True)
     def enable_triggers(self):