Handle ValueError for trigger_dag_run API request

[MaksYermak]

In this PR I have added a code which handle ValueError exception for trigger_dag_run API request and raise HTTPException with 400 http code. These changes improve error messaging from this request and user can see clear message in task logs.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[SameerMesiah97]

I think the rationale could be better explained. ValueError is far too broad in this codepath and catching it can make real issues under a generic 400 exception.

[SameerMesiah97]

Catching all instances of ValueError is a bit too broad. Since it wraps full trigger_dag invocation, this could convert unrelated internal ValueErrors into 400 responses and mask real server-side issues.

Could you elaborate more on the scenarios under which you expected to encounter a ValueError? Maybe it is better to narrow this to that?

[MaksYermak]

Hi @SameerMesiah97 ,
I have tried to cover these ValueErrors https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/serialization/definitions/dag.py#L544-L581 from create_dagrun function which run inside trigger_dag function under trigger_dar_run API call. Because in the current time users see this error:

Top level error
Traceback (most recent call last):
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/sdk/execution_time/task_runner.py", line 920, in run
    result = _execute_task(context=context, ti=ti, log=log)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/sdk/execution_time/task_runner.py", line 1307, in _execute_task
    result = ctx.run(execute, context=context)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/sdk/bases/operator.py", line 416, in wrapper
    return func(self, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/providers/standard/operators/trigger_dagrun.py", line 258, in execute
    self._trigger_dag_af_3(
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/providers/standard/operators/trigger_dagrun.py", line 269, in _trigger_dag_af_3
    raise DagRunTriggerException(
airflow.exceptions.DagRunTriggerException

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/sdk/execution_time/task_runner.py", line 1450, in main
    state, _, error = run(ti, context, log)
                      ^^^^^^^^^^^^^^^^^^^^^
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/composer/patches/metrics/monkey_patching/airflow_sdk_execution_time_task_runner.py", line 36, in wrapper
    state, msg, error = f(ti, *args, **kwargs)
                        ^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/sdk/execution_time/task_runner.py", line 940, in run
    msg, state = _handle_trigger_dag_run(drte, context, ti, log)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/sdk/execution_time/task_runner.py", line 1053, in _handle_trigger_dag_run
    comms_msg = SUPERVISOR_COMMS.send(
                ^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/sdk/execution_time/comms.py", line 207, in send
    return self._get_response()
           ^^^^^^^^^^^^^^^^^^^^
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/sdk/execution_time/comms.py", line 271, in _get_response
    return self._from_frame(frame)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/python3.11/lib/python3.11/site-packages/airflow/sdk/execution_time/comms.py", line 258, in _from_frame
    raise AirflowRuntimeError(error=err)
airflow.sdk.exceptions.AirflowRuntimeError: API_SERVER_ERROR: {'status_code': 500, 'message': 'Server returned error', 'detail': {'message': 'Internal server error', 'correlation-id': '002dab06-ee5b-8be7-a321-7b13b5311915'}}

with 500 status code and unclear error message:

{'status_code': 500, 'message': 'Server returned error', 'detail': {'message': 'Internal server error', 'correlation-id': '002dab06-ee5b-8be7-a321-7b13b5311915'}}

[potiuk]

@MaksYermak A few things need addressing before review — see our Pull Request quality criteria.

• ❌ Pre-commit / static checks. See docs.

No rush.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.