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 a1f81efde91 Fix asset materialization dropping partition date on
partitioned Dag runs (#69314)
a1f81efde91 is described below
commit a1f81efde915fb1c024b684280c38d1de0253138
Author: Wei Lee <[email protected]>
AuthorDate: Fri Jul 3 22:45:19 2026 +0800
Fix asset materialization dropping partition date on partitioned Dag runs
(#69314)
---
.../api_fastapi/core_api/routes/public/assets.py | 1 +
.../core_api/routes/public/test_assets.py | 36 ++++++++++++++++++++++
2 files changed, 37 insertions(+)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
index 15af36445dd..a19f4c7da77 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
@@ -464,6 +464,7 @@ def materialize_asset(
triggering_user_name=user.get_name(),
state=DagRunState.QUEUED,
partition_key=params["partition_key"],
+ partition_date=params["partition_date"],
note=params["note"],
session=session,
)
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 33c43785a55..0470b0b1e4e 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
@@ -41,7 +41,9 @@ from airflow.models.dagrun import DagRun
from airflow.models.serialized_dag import SerializedDagModel
from airflow.models.trigger import Trigger
from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk import Asset
from airflow.timetables.simple import PartitionedAtRuntime
+from airflow.timetables.trigger import CronPartitionTimetable
from airflow.utils.session import provide_session
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunType
@@ -1684,6 +1686,40 @@ class TestPostAssetMaterialize(TestAssets):
assert response.status_code == 400
assert "must not contain '..'" in response.json()["detail"]
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_should_respond_200_with_partition_date_for_partitioned_dag(
+ self, test_client, dag_maker, session
+ ):
+ """Materializing a Dag with a real partitioned timetable must populate
partition_date.
+
+ Regression guard: before this fix, `partition_date` resolved by
`validate_context` was
+ dropped when creating the run, unlike the sibling
`/dags/{dag_id}/dagRuns` trigger route.
+ """
+ partitioned_dag_id = "test_materialize_populates_partition_date"
+ asset = Asset(name="materialize_partition_date_asset",
uri="s3://bucket/materialize-partition-date")
+ with dag_maker(
+ dag_id=partitioned_dag_id,
+ schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"),
+ start_date=DEFAULT_DATE,
+ session=session,
+ serialized=True,
+ ):
+ EmptyOperator(task_id="task", outlets=[asset])
+ session.commit()
+
+ asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri
== asset.uri))
+
+ response = test_client.post(
+ f"/assets/{asset_id}/materialize",
+ json={"partition_key": "2025-06-01T00:00:00"},
+ )
+ assert response.status_code == 200
+
+ dag_run = session.scalar(select(DagRun).where(DagRun.dag_id ==
partitioned_dag_id))
+ assert dag_run is not None
+ assert dag_run.partition_key == "2025-06-01T00:00:00"
+ assert dag_run.partition_date == timezone.datetime(2025, 6, 1)
+
class TestGetAssetQueuedEvents(TestQueuedEventEndpoint):
@pytest.mark.usefixtures("time_freezer")
