Repository: PrefectHQ/prefect
Issue category: Feature Request

Add ability to configure `ConcurrenyOptions` to the CLI
### Describe the current behavior

Currently users can set the the concurrency_limit for a deployment when creating it with the CLI by specifying the limit with and int as so: `deploy "my-deployment -cl 1". 

Upcoming work #15291, adds options for handling concurrency limiting which we will need add to the CLI.

### Describe the proposed behavior

I think that it while it's possible to create a custom type in typer and configure them as so `deploy "my-deployment" -cl 1 --concurrency-options = {'collisions-strategy' = 'cancel-new'}`, it makes for a better user experience if we flatten it. This could be done for each subsequent options that gets added and along with sane defaults, users wouldn't be setting a lot of these.

### Example Use

`deploy "my-deployment" -cl 42 --collision-strategy "cancel-new"`

### Additional context

_No response_


Functions to edit:
src/prefect/cli/deploy.py:deploy

Patch:
diff --git a/src/prefect/cli/deploy.py b/src/prefect/cli/deploy.py
index cf15c23ab6ee..cfa321482e55 100644
--- a/src/prefect/cli/deploy.py
+++ b/src/prefect/cli/deploy.py
@@ -41,6 +41,7 @@
 )
 from prefect.cli.root import app, is_interactive
 from prefect.client.schemas.actions import DeploymentScheduleCreate
+from prefect.client.schemas.objects import ConcurrencyLimitConfig
 from prefect.client.schemas.schedules import (
     CronSchedule,
     IntervalSchedule,
@@ -242,6 +243,11 @@ async def deploy(
         "--concurrency-limit",
         help=("The maximum number of concurrent runs for this deployment."),
     ),
+    concurrency_limit_collision_strategy: str = typer.Option(
+        None,
+        "--collision-strategy",
+        help="Configure the behavior for runs once the concurrency limit is reached. Falls back to `ENQUEUE` if unset.",
+    ),
     work_pool_name: str = SettingsOption(
         PREFECT_DEFAULT_WORK_POOL_NAME,
         "-p",
@@ -371,12 +377,23 @@ async def deploy(
         job_variables = list()
     job_variables.extend(variables)
 
+    concurrency_limit_config = (
+        None
+        if concurrency_limit is None
+        else concurrency_limit
+        if concurrency_limit_collision_strategy is None
+        else ConcurrencyLimitConfig(
+            limit=concurrency_limit,
+            collision_strategy=concurrency_limit_collision_strategy,
+        ).model_dump()
+    )
+
     options = {
         "entrypoint": entrypoint,
         "description": description,
         "version": version,
         "tags": tags,
-        "concurrency_limit": concurrency_limit,
+        "concurrency_limit": concurrency_limit_config,
         "work_pool_name": work_pool_name,
         "work_queue_name": work_queue_name,
         "variables": job_variables,

Test patch: diff --git a/tests/cli/test_deploy.py b/tests/cli/test_deploy.py
index 661494a4ce4a..1f455b62306f 100644
--- a/tests/cli/test_deploy.py
+++ b/tests/cli/test_deploy.py
@@ -276,7 +276,7 @@ async def test_project_deploy(self, project_dir, prefect_client: PrefectClient):
         await run_sync_in_worker_thread(
             invoke_and_assert,
             command=(
-                "deploy ./flows/hello.py:my_flow -n test-name -p test-pool -cl 42 --version"
+                "deploy ./flows/hello.py:my_flow -n test-name -p test-pool --version"
                 " 1.0.0 -v env=prod -t foo-bar"
             ),
             expected_code=0,
@@ -293,7 +293,6 @@ async def test_project_deploy(self, project_dir, prefect_client: PrefectClient):
         assert deployment.work_pool_name == "test-pool"
         assert deployment.version == "1.0.0"
         assert deployment.tags == ["foo-bar"]
-        assert deployment.global_concurrency_limit.limit == 42
         assert deployment.job_variables == {"env": "prod"}
         assert deployment.enforce_parameter_schema
 
@@ -446,6 +445,69 @@ async def test_deploy_does_not_prompt_storage_when_pull_step_exists(
             ],
         )
 
+    @pytest.mark.parametrize(
+        "cli_options,expected_limit,expected_strategy",
+        [
+            pytest.param("-cl 42", 42, None, id="limit-only"),
+            pytest.param(
+                "-cl 42 --collision-strategy CANCEL_NEW",
+                42,
+                "CANCEL_NEW",
+                id="limit-and-strategy",
+            ),
+            pytest.param(
+                "--collision-strategy CANCEL_NEW",
+                None,
+                None,
+                id="strategy-only",
+            ),
+        ],
+    )
+    async def test_deploy_with_concurrency_limit_and_options(
+        self,
+        project_dir,
+        prefect_client: PrefectClient,
+        cli_options,
+        expected_limit,
+        expected_strategy,
+    ):
+        await prefect_client.create_work_pool(
+            WorkPoolCreate(name="test-pool", type="test")
+        )
+        await run_sync_in_worker_thread(
+            invoke_and_assert,
+            command=(
+                "deploy ./flows/hello.py:my_flow -n test-deploy-concurrency-limit -p test-pool "
+                + cli_options
+                # "-cl 42 --collision-strategy CANCEL_NEW"
+            ),
+            expected_code=0,
+            expected_output_contains=[
+                "An important name/test-deploy-concurrency-limit",
+                "prefect worker start --pool 'test-pool'",
+            ],
+        )
+
+        deployment = await prefect_client.read_deployment_by_name(
+            "An important name/test-deploy-concurrency-limit"
+        )
+        assert deployment.name == "test-deploy-concurrency-limit"
+        assert deployment.work_pool_name == "test-pool"
+
+        if expected_limit is not None:
+            assert deployment.global_concurrency_limit is not None
+            assert deployment.global_concurrency_limit.limit == expected_limit
+        else:
+            assert deployment.global_concurrency_limit is None
+
+        if expected_strategy is not None:
+            assert deployment.concurrency_options is not None
+            assert (
+                deployment.concurrency_options.collision_strategy == expected_strategy
+            )
+        else:
+            assert deployment.concurrency_options is None
+
     class TestGeneratedPullAction:
         async def test_project_deploy_generates_pull_action(
             self, work_pool, prefect_client, uninitialized_project_dir