diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py b/airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py index 2b178e97fc220..966a13d8dc817 100644 --- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py +++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py @@ -153,6 +153,14 @@ def trigger_dag_run( "message": f"A run already exists for Dag '{dag_id}' with run_id '{run_id}'", }, ) + except ValueError as e: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + detail={ + "reason": "value_error", + "message": str(e), + }, + ) from e @router.post( diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py index 337bc1c90b8d5..31be920bd25fe 100644 --- a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py +++ b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py @@ -17,6 +17,8 @@ from __future__ import annotations +import re + import pytest import time_machine from fastapi import Request @@ -260,6 +262,31 @@ async def auth_as_parent_ti(request: Request) -> TIToken: child_run = session.scalars(select(DagRun).where(DagRun.run_id == child_run_id)).one() assert child_run.triggering_user_name == parent_triggering_user_name + def test_trigger_dag_run_value_error(self, client, session, dag_maker): + """Test that error is raised when a DAG Run has ValueError.""" + + dag_id = "test_trigger_dag_run_value_error" + run_id = "manual__{test_run_id}" + logical_date = timezone.datetime(2026, 5, 22) + + with dag_maker(dag_id=dag_id, session=session, serialized=True): + EmptyOperator(task_id="test_task") + + session.commit() + + response = client.post( + f"/execution/dag-runs/{dag_id}/{run_id}", + json={"logical_date": logical_date.isoformat()}, + ) + + detail_message = response.json()["detail"]["message"] + detail_reason = response.json()["detail"]["reason"] + + pattern = r"The run_id provided '.*' does not match regex pattern '.*' or '.*'" + assert re.search(pattern, detail_message) + assert detail_reason == "value_error" + assert response.status_code == 400 + class TestDagRunClear: def setup_method(self):