Skip to content

Commit 0120ba7

Browse files
authored
API: Return 400 instead of 500 from materialize_asset for invalid validation input (apache#67445)
The materialize_asset POST endpoint passed user input (dag_run_id, logical_date/data_interval pairing, partition_key) straight through to MaterializeAssetBody.validate_context() and dag.create_dagrun(), which raise ValueError / ParamValidationError on invalid input (e.g. dag_run_id containing '..', invalid partition_key type, logical_date/data_interval mismatch). Those exceptions escaped uncaught and were returned as 500 Internal Server Error, even though the route's OpenAPI spec already documents 400 for this case. Wrap the validate_context / create_dagrun calls in try/except and re-raise as HTTPException(400) with the validator's message in the detail, matching the sibling pattern in dag_run.trigger_dag_run. Regression test asserts dag_run_id='bad..id' returns 400, not 500.
1 parent 896c031 commit 0120ba7

2 files changed

Lines changed: 32 additions & 15 deletions

File tree

  • airflow-core

airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py

Lines changed: 19 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@
7373
from airflow.api_fastapi.logging.decorators import action_logging
7474
from airflow.assets.manager import asset_manager
7575
from airflow.configuration import conf
76+
from airflow.exceptions import ParamValidationError
7677
from airflow.models.asset import (
7778
AssetAliasModel,
7879
AssetDagRunQueue,
@@ -443,21 +444,24 @@ def materialize_asset(
443444
f"Dag with dag_id: '{dag_id}' does not allow asset materialization runs",
444445
)
445446

446-
params = (body or MaterializeAssetBody()).validate_context(dag)
447-
return dag.create_dagrun(
448-
run_id=params["run_id"],
449-
logical_date=params["logical_date"],
450-
data_interval=params["data_interval"],
451-
run_after=params["run_after"],
452-
conf=params["conf"],
453-
run_type=DagRunType.ASSET_MATERIALIZATION,
454-
triggered_by=DagRunTriggeredByType.REST_API,
455-
triggering_user_name=user.get_name(),
456-
state=DagRunState.QUEUED,
457-
partition_key=params["partition_key"],
458-
note=params["note"],
459-
session=session,
460-
)
447+
try:
448+
params = (body or MaterializeAssetBody()).validate_context(dag)
449+
return dag.create_dagrun(
450+
run_id=params["run_id"],
451+
logical_date=params["logical_date"],
452+
data_interval=params["data_interval"],
453+
run_after=params["run_after"],
454+
conf=params["conf"],
455+
run_type=DagRunType.ASSET_MATERIALIZATION,
456+
triggered_by=DagRunTriggeredByType.REST_API,
457+
triggering_user_name=user.get_name(),
458+
state=DagRunState.QUEUED,
459+
partition_key=params["partition_key"],
460+
note=params["note"],
461+
session=session,
462+
)
463+
except (ParamValidationError, ValueError) as e:
464+
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from e
461465

462466

463467
@assets_router.get(

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1609,6 +1609,19 @@ def test_should_respond_403_when_user_cannot_trigger_dag(self, test_client):
16091609
user=mock.ANY,
16101610
)
16111611

1612+
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
1613+
def test_should_respond_400_on_invalid_dag_run_id(self, test_client):
1614+
"""A dag_run_id containing '..' triggers ValueError in DagRun.validate_run_id.
1615+
1616+
It must surface as 400 BAD_REQUEST, not 500 INTERNAL_SERVER_ERROR.
1617+
"""
1618+
response = test_client.post(
1619+
"/assets/1/materialize",
1620+
json={"dag_run_id": "bad..id"},
1621+
)
1622+
assert response.status_code == 400
1623+
assert "must not contain '..'" in response.json()["detail"]
1624+
16121625

16131626
class TestGetAssetQueuedEvents(TestQueuedEventEndpoint):
16141627
@pytest.mark.usefixtures("time_freezer")

0 commit comments

Comments
 (0)