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 15af36445dd95..a19f4c7da7797 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 33c43785a55a8..0470b0b1e4e51 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.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 @@ def test_should_respond_400_on_invalid_dag_run_id(self, test_client): 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")