Expose enforce parameter schema flag on individual flow runs (#14718)

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

@@ -15990,6 +15990,18 @@
                         "type": "object",
                         "title": "Parameters"
                     },
+                    "enforce_parameter_schema": {
+                        "anyOf": [
+                            {
+                                "type": "boolean"
+                            },
+                            {
+                                "type": "null"
+                            }
+                        ],
+                        "title": "Enforce Parameter Schema",
+                        "description": "Whether or not to enforce the parameter schema on this run."
+                    },
                     "context": {
                         "type": "object",
                         "title": "Context"

src/prefect/client/schemas/actions.py

@@ -377,6 +377,10 @@ class DeploymentFlowRunCreate(ActionBaseModel):
     parameters: Dict[str, Any] = Field(
         default_factory=dict, description="The parameters for the flow run."
     )
+    enforce_parameter_schema: Optional[bool] = Field(
+        default=None,
+        description="Whether or not to enforce the parameter schema on this run.",
+    )
     context: Dict[str, Any] = Field(
         default_factory=dict, description="The context for the flow run."
     )

src/prefect/server/api/deployments.py

@@ -709,7 +709,14 @@ async def create_flow_run_from_deployment(
                 detail=f"Error hydrating flow run parameters: {exc}",
             )
 
-        if deployment.enforce_parameter_schema:
+        # default
+        enforce_parameter_schema = deployment.enforce_parameter_schema
+
+        # run override
+        if flow_run.enforce_parameter_schema is not None:
+            enforce_parameter_schema = flow_run.enforce_parameter_schema
+
+        if enforce_parameter_schema:
             if not isinstance(deployment.parameter_openapi_schema, dict):
                 raise HTTPException(
                     status.HTTP_409_CONFLICT,
@@ -759,6 +766,7 @@ async def create_flow_run_from_deployment(
                     "tags",
                     "infrastructure_document_id",
                     "work_queue_name",
+                    "enforce_parameter_schema",
                 }
             ),
             flow_id=deployment.flow_id,

src/prefect/server/schemas/actions.py

@@ -566,6 +566,10 @@ class DeploymentFlowRunCreate(ActionBaseModel):
         examples=["my-flow-run"],
     )
     parameters: Dict[str, Any] = Field(default_factory=dict)
+    enforce_parameter_schema: Optional[bool] = Field(
+        default=None,
+        description="Whether or not to enforce the parameter schema on this run.",
+    )
     context: Dict[str, Any] = Field(default_factory=dict)
     infrastructure_document_id: Optional[UUID] = Field(None)
     empirical_policy: schemas.core.FlowRunPolicy = Field(

tests/server/orchestration/api/test_deployments.py

@@ -2843,6 +2843,29 @@ async def test_create_flow_run_enforces_parameter_schema(
 
         assert response.status_code == 201
 
+    async def test_create_flow_run_respects_per_run_validation_flag(
+        self,
+        deployment_with_parameter_schema,
+        client,
+    ):
+        response = await client.post(
+            f"/deployments/{deployment_with_parameter_schema.id}/create_flow_run",
+            json={"parameters": {"x": 1}},
+        )
+
+        assert response.status_code == 409
+        assert (
+            "Validation failed for field 'x'. Failure reason: 1 is not of type 'string'"
+            in response.text
+        )
+
+        response = await client.post(
+            f"/deployments/{deployment_with_parameter_schema.id}/create_flow_run",
+            json={"parameters": {"x": 1}, "enforce_parameter_schema": False},
+        )
+
+        assert response.status_code == 201
+
     async def test_create_flow_run_does_not_enforce_parameter_schema_when_enforcement_is_toggled_off(
         self,
         deployment_with_parameter_schema,