This is an automated email from the ASF dual-hosted git repository.
Lee-W pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 01b425a06f6 Reject invalid partition keys in the create asset events
API (#69453)
01b425a06f6 is described below
commit 01b425a06f6998950cada673582a17b6b2264b41
Author: Wei Lee <[email protected]>
AuthorDate: Wed Jul 8 09:53:26 2026 +0800
Reject invalid partition keys in the create asset events API (#69453)
---
.../api_fastapi/core_api/datamodels/assets.py | 20 ++++++++++++++----
.../core_api/openapi/v2-rest-api-generated.yaml | 2 ++
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 4 +++-
.../core_api/routes/public/test_assets.py | 24 ++++++++++++++++++++++
.../src/airflowctl/api/datamodels/generated.py | 6 +++++-
5 files changed, 50 insertions(+), 6 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
index 753ea94f1d9..3be0707bee1 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
@@ -19,14 +19,24 @@ from __future__ import annotations
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
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 334fe6dae8d..97ae27faff5 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -13276,6 +13276,8 @@ components:
partition_key:
anyOf:
- type: string
+ maxLength: 250
+ pattern: \S
- type: 'null'
title: Partition Key
extra:
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index ec5024d981e..55b25b2c975 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/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'
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
index 112138b46ab..1b452924ffe 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
@@ -37,6 +37,7 @@ from airflow.models.asset import (
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
@@ -1417,6 +1418,29 @@ class TestPostAssetEvents(TestAssets):
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):
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index fa352e61140..8ce734890ad 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/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
