Creates ConcurrencyOptions pydantic model.

docs/3.0/api-ref/rest-api/server/schema.json

@@ -14593,6 +14593,15 @@
                 "title": "ConcurrencyLimitCreate",
                 "description": "Data used by the Prefect REST API to create a concurrency limit."
             },
+            "ConcurrencyLimitStrategy": {
+                "type": "string",
+                "enum": [
+                    "ENQUEUE",
+                    "CANCEL_NEW"
+                ],
+                "title": "ConcurrencyLimitStrategy",
+                "description": "Enumeration of concurrency collision strategies."
+            },
             "ConcurrencyLimitV2": {
                 "properties": {
                     "id": {
@@ -14804,6 +14813,19 @@
                 "title": "ConcurrencyLimitV2Update",
                 "description": "Data used by the Prefect REST API to update a v2 concurrency limit."
             },
+            "ConcurrencyOptions": {
+                "properties": {
+                    "collision_strategy": {
+                        "$ref": "#/components/schemas/ConcurrencyLimitStrategy"
+                    }
+                },
+                "type": "object",
+                "required": [
+                    "collision_strategy"
+                ],
+                "title": "ConcurrencyOptions",
+                "description": "Class for storing the concurrency config in database."
+            },
             "Constant": {
                 "properties": {
                     "input_type": {
@@ -16401,6 +16423,17 @@
                         "title": "Concurrency Limit",
                         "description": "The deployment's concurrency limit."
                     },
+                    "concurrency_options": {
+                        "anyOf": [
+                            {
+                                "$ref": "#/components/schemas/ConcurrencyOptions"
+                            },
+                            {
+                                "type": "null"
+                            }
+                        ],
+                        "description": "The deployment's concurrency options."
+                    },
                     "parameters": {
                         "anyOf": [
                             {

src/prefect/server/database/migrations/MIGRATION-NOTES.md

@@ -8,6 +8,11 @@ Each time a database migration is written, an entry is included here with:
 
 This gives us a history of changes and will create merge conflicts if two migrations are made at once, flagging situations where a branch needs to be updated before merging.
 
+# Adds `concurrency_options` to `Deployments`
+
+SQLite: `7d6350aea855`
+Postgres: `555ed31b284d`
+
 # Add `concurrency_limit` to `Deployments`
 SQLite: `f93e1439f022`
 Postgres:`97429116795e`

src/prefect/server/database/migrations/versions/postgresql/2024_09_11_090317_555ed31b284d_add_concurrency_options.py

@@ -0,0 +1,33 @@
+"""add concurrency options
+
+Revision ID: 555ed31b284d
+Revises: 97429116795e
+Create Date: 2024-09-11 09:03:17.744587
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+import prefect
+from prefect.server.schemas.core import ConcurrencyOptions
+
+# revision identifiers, used by Alembic.
+revision = "555ed31b284d"
+down_revision = "97429116795e"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "deployment",
+        sa.Column(
+            "concurrency_options",
+            prefect.server.utilities.database.Pydantic(ConcurrencyOptions),
+            nullable=True,
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("deployment", "concurrency_options")

src/prefect/server/database/migrations/versions/sqlite/2024_09_11_090106_7d6350aea855_add_concurrency_options.py

@@ -0,0 +1,34 @@
+"""add concurrency options
+
+Revision ID: 7d6350aea855
+Revises: f93e1439f022
+Create Date: 2024-09-11 09:01:06.678866
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+import prefect
+from prefect.server.schemas.core import ConcurrencyOptions
+
+# revision identifiers, used by Alembic.
+revision = "7d6350aea855"
+down_revision = "f93e1439f022"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    with op.batch_alter_table("deployment", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "concurrency_options",
+                prefect.server.utilities.database.Pydantic(ConcurrencyOptions),
+                nullable=True,
+            )
+        )
+
+
+def downgrade():
+    with op.batch_alter_table("deployment", schema=None) as batch_op:
+        batch_op.drop_column("concurrency_options")

src/prefect/server/database/orm_models.py

@@ -878,7 +878,18 @@ def job_variables(self):
     )
 
     concurrency_limit: Mapped[Union[int, None]] = mapped_column(
-        sa.Integer, default=None, nullable=True
+        sa.Integer,
+        server_default=None,
+        nullable=True,
+        default=None,
+    )
+    concurrency_options: Mapped[
+        Union[schemas.core.ConcurrencyOptions, None]
+    ] = mapped_column(
+        Pydantic(schemas.core.ConcurrencyOptions),
+        server_default=None,
+        nullable=True,
+        default=None,
     )
     tags: Mapped[List[str]] = mapped_column(
         JSON, server_default="[]", default=list, nullable=False

src/prefect/server/models/deployments.py

@@ -85,6 +85,12 @@ async def create_deployment(
     insert_values = deployment.model_dump_for_orm(
         exclude_unset=True, exclude={"schedules"}
     )
+    if isinstance(deployment.concurrency_limit, schemas.core.ConcurrencyLimitConfig):
+        concurrency_options = insert_values.pop("concurrency_limit")
+        insert_values["concurrency_options"] = {
+            "collision_strategy": concurrency_options.collision_strategy,
+        }
+        insert_values["concurrency_limit"] = concurrency_options.concurrency
 
     # The job_variables field in client and server schemas is named
     # infra_overrides in the database.
@@ -94,7 +100,14 @@ async def create_deployment(
 
     conflict_update_fields = deployment.model_dump_for_orm(
         exclude_unset=True,
-        exclude={"id", "created", "created_by", "schedules", "job_variables"},
+        exclude={
+            "id",
+            "created",
+            "created_by",
+            "schedules",
+            "job_variables",
+            "concurrency_limit",
+        },
     )
     if job_variables:
         conflict_update_fields["infra_overrides"] = job_variables

src/prefect/server/schemas/actions.py

@@ -270,6 +270,9 @@ def remove_old_fields(cls, values):
     concurrency_limit: Optional[PositiveInteger] = Field(
         default=None, description="The deployment's concurrency limit."
     )
+    concurrency_options: Optional[schemas.core.ConcurrencyOptions] = Field(
+        default=None, description="The deployment's concurrency options."
+    )
     parameters: Optional[Dict[str, Any]] = Field(
         default=None,
         description="Parameters for flow runs scheduled by the deployment.",

src/prefect/server/schemas/core.py

@@ -48,7 +48,12 @@
     PositiveInteger,
     StrictVariableValue,
 )
-from prefect.utilities.collections import dict_to_flatdict, flatdict_to_dict, listrepr
+from prefect.utilities.collections import (
+    AutoEnum,
+    dict_to_flatdict,
+    flatdict_to_dict,
+    listrepr,
+)
 from prefect.utilities.names import generate_slug, obfuscate
 
 if TYPE_CHECKING:
@@ -146,6 +151,32 @@ class UpdatedBy(BaseModel):
     )
 
 
+class ConcurrencyLimitStrategy(AutoEnum):
+    """
+    Enumeration of concurrency collision strategies.
+    """
+
+    ENQUEUE = AutoEnum.auto()
+    CANCEL_NEW = AutoEnum.auto()
+
+
+class ConcurrencyLimitConfig(BaseModel):
+    """
+    Options for configuring deployment concurrency limits.
+    """
+
+    concurrency: int
+    collision_strategy: ConcurrencyLimitStrategy
+
+
+class ConcurrencyOptions(BaseModel):
+    """
+    Class for storing the concurrency config in database.
+    """
+
+    collision_strategy: ConcurrencyLimitStrategy
+
+
 class FlowRun(ORMBaseModel):
     """An ORM representation of flow run data."""
 
@@ -546,9 +577,9 @@ class Deployment(ORMBaseModel):
     schedules: List[DeploymentSchedule] = Field(
         default_factory=list, description="A list of schedules for the deployment."
     )
-    concurrency_limit: Optional[PositiveInteger] = Field(
-        default=None, description="The concurrency limit for the deployment."
-    )
+    concurrency_limit: Optional[
+        Union[NonNegativeInteger, ConcurrencyLimitConfig]
+    ] = Field(default=None, description="The concurrency limit for the deployment.")
     job_variables: Dict[str, Any] = Field(
         default_factory=dict,
         description="Overrides to apply to flow run infrastructure at runtime.",

tests/server/models/test_deployments.py

@@ -244,6 +244,25 @@ async def test_create_deployment_with_updated_by(self, session, flow):
         )
         assert updated_deployment.updated_by.type == new_updated_by.type
 
+    async def test_create_deployment_with_concurrency_limit(self, session, flow):
+        concurrency_options = schemas.core.ConcurrencyLimitConfig(
+            concurrency=10,
+            collision_strategy="ENQUEUE",
+        )
+        deployment = await models.deployments.create_deployment(
+            session=session,
+            deployment=schemas.core.Deployment(
+                name="My Deployment",
+                flow_id=flow.id,
+                concurrency_limit=concurrency_options,
+            ),
+        )
+        assert deployment.concurrency_limit == concurrency_options.concurrency
+        assert (
+            deployment.concurrency_options.collision_strategy
+            == concurrency_options.collision_strategy
+        )
+
 
 class TestReadDeployment:
     async def test_read_deployment(self, session, flow, flow_function):
@@ -1068,6 +1087,27 @@ async def test_updating_deployment_does_not_duplicate_work_queue(
         assert wq is not None
         assert wq.work_pool == work_pool
 
+    async def test_update_deployment_with_concurrency_limit(
+        self,
+        session,
+        deployment,
+    ):
+        await models.deployments.update_deployment(
+            session=session,
+            deployment_id=deployment.id,
+            deployment=schemas.actions.DeploymentUpdate(
+                concurrency_limit=42,
+                concurrency_options=schemas.core.ConcurrencyOptions(
+                    collision_strategy="CANCEL_NEW"
+                ),
+            ),
+        )
+        updated_deployment = await models.deployments.read_deployment(
+            session=session, deployment_id=deployment.id
+        )
+        assert updated_deployment.concurrency_limit == 42
+        assert updated_deployment.concurrency_options.collision_strategy == "CANCEL_NEW"
+
 
 @pytest.fixture
 async def deployment_schedules(