This is an automated email from the ASF dual-hosted git repository. vatsrahul1001 pushed a commit to branch v3-3-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 2afe211fd2eaf70cf90082e724ace45f8d3a4fe3 Author: Wei Lee <[email protected]> AuthorDate: Sat Jul 4 02:30:03 2026 +0800 Fix asset materialization dropping partition date on partitioned Dag runs (#69314) (#69339) --- .../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")
