Reject invalid partition keys in the create asset events API

airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py

@@ -19,14 +19,24 @@
 
 from collections.abc import Iterable
 from datetime import datetime
-from typing import TYPE_CHECKING
-
-from pydantic import AliasPath, AwareDatetime, ConfigDict, Field, JsonValue, NonNegativeInt, field_validator
+from typing import TYPE_CHECKING, Annotated
+
+from pydantic import (
+    AliasPath,
+    AwareDatetime,
+    ConfigDict,
+    Field,
+    JsonValue,
+    NonNegativeInt,
+    StringConstraints,
+    field_validator,
+)
 
 from airflow._shared.secrets_masker import redact
 from airflow._shared.timezones import timezone
 from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
 from airflow.api_fastapi.core_api.datamodels.dag_run import TriggerDAGRunPostBody
+from airflow.models.base import ID_LEN
 from airflow.utils.types import DagRunType
 
 if TYPE_CHECKING:
@@ -190,7 +200,9 @@ class CreateAssetEventsBody(StrictBaseModel):
     """Create asset events request."""
 
     asset_id: int
-    partition_key: str | None = None
+    # pattern (not strip_whitespace) so the value isn't mutated — must stay byte-identical to what
+    # `_validate_outlet_event_partition_keys` in the Execution API accepts for the same raw input.
+    partition_key: Annotated[str, StringConstraints(pattern=r"\S", max_length=ID_LEN)] | None = None
     extra: dict = Field(default_factory=dict)
     access_control: AssetEventAccessControl | None = None
 

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -13264,6 +13264,8 @@ components:
         partition_key:
           anyOf:
           - type: string
+            maxLength: 250
+            pattern: \S
           - type: 'null'
           title: Partition Key
         extra:

airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -2662,7 +2662,9 @@ export const $CreateAssetEventsBody = {
         partition_key: {
             anyOf: [
                 {
-                    type: 'string'
+                    type: 'string',
+                    maxLength: 250,
+                    pattern: '\\S'
                 },
                 {
                     type: 'null'

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py

@@ -37,6 +37,7 @@
     DagScheduleAssetReference,
     TaskOutletAssetReference,
 )
+from airflow.models.base import ID_LEN
 from airflow.models.dagrun import DagRun
 from airflow.models.serialized_dag import SerializedDagModel
 from airflow.models.trigger import Trigger
@@ -1336,6 +1337,29 @@ def test_invalid_attr_not_allowed(self, test_client, session):
 
         assert response.status_code == 422
 
+    @pytest.mark.parametrize(
+        ("partition_key", "expected_status_code"),
+        [
+            pytest.param("", 422, id="empty"),
+            pytest.param("   ", 422, id="whitespace_only"),
+            pytest.param("a" * (ID_LEN + 1), 422, id="too_long"),
+            pytest.param("2026-03-23", 200, id="valid"),
+            pytest.param(None, 200, id="none"),
+        ],
+    )
+    def test_partition_key_validation(self, test_client, session, partition_key, expected_status_code):
+        (asset,) = self.create_assets(num=1, session=session)
+        event_payload = {"asset_id": asset.id, "partition_key": partition_key}
+        response = test_client.post("/assets/events", json=event_payload)
+        assert response.status_code == expected_status_code
+
+    def test_partition_key_preserves_surrounding_whitespace(self, test_client, session):
+        (asset,) = self.create_assets(num=1, session=session)
+        event_payload = {"asset_id": asset.id, "partition_key": "  2026-03-23  "}
+        response = test_client.post("/assets/events", json=event_payload)
+        assert response.status_code == 200
+        assert response.json()["partition_key"] == "  2026-03-23  "
+
     @pytest.mark.usefixtures("time_freezer")
     @pytest.mark.enable_redact
     def test_should_mask_sensitive_extra(self, test_client, session):

airflow-ctl/src/airflowctl/api/datamodels/generated.py

@@ -431,6 +431,10 @@ class ConnectionTestResponse(BaseModel):
     message: Annotated[str, Field(title="Message")]
 
 
+class PartitionKey(RootModel[str]):
+    root: Annotated[str, Field(max_length=250, pattern="\\S", title="Partition Key")]
+
+
 class CreateAssetEventsBody(BaseModel):
     """
     Create asset events request.
@@ -440,7 +444,7 @@ class CreateAssetEventsBody(BaseModel):
         extra="forbid",
     )
     asset_id: Annotated[int, Field(title="Asset Id")]
-    partition_key: Annotated[str | None, Field(title="Partition Key")] = None
+    partition_key: Annotated[PartitionKey | None, Field(title="Partition Key")] = None
     extra: Annotated[dict[str, Any] | None, Field(title="Extra")] = None
     access_control: AssetEventAccessControl | None = None