This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 9322e902156 [v3-3-test] Route SQLAlchemyError through global exception
handler (#69267) (#70236)
9322e902156 is described below
commit 9322e9021568210d1550691d4677016a15aa5fdc
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Jul 22 16:33:26 2026 +0200
[v3-3-test] Route SQLAlchemyError through global exception handler (#69267)
(#70236)
* Route SQLAlchemyError through global exception handler
* Replace manual traceback with logging
* Update airflow-core/src/airflow/api_fastapi/common/exceptions.py
* Add test for ti_run and include traceback when expose_stacktrace=True
* Add test for ti_run and include traceback when expose_stacktrace=True
* fix referenced before assignment problems
* Handle execution API database errors consistently
* Handle execution API database errors consistently
* Refactor duplicated database error handling into helper
* fix code format
* Trigger CI rerun
* Move database error response handling into handler class
---------
(cherry picked from commit 7c03dabdcddfda99e8b663bd254ce18637278407)
Co-authored-by: fat-catTW <[email protected]>
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
---
airflow-core/src/airflow/api_fastapi/app.py | 3 +-
.../src/airflow/api_fastapi/common/exceptions.py | 48 ++++++++++++-----
.../src/airflow/api_fastapi/core_api/app.py | 7 ---
.../src/airflow/api_fastapi/execution_api/app.py | 2 +
.../execution_api/routes/task_instances.py | 14 ++---
.../unit/api_fastapi/common/test_exceptions.py | 40 +++++++++++++++
.../unit/api_fastapi/execution_api/test_app.py | 18 ++++++-
.../versions/head/test_task_instances.py | 60 +++++++++++++++++++++-
8 files changed, 159 insertions(+), 33 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/app.py
b/airflow-core/src/airflow/api_fastapi/app.py
index a4dbc3c9b72..e4d1e40efbb 100644
--- a/airflow-core/src/airflow/api_fastapi/app.py
+++ b/airflow-core/src/airflow/api_fastapi/app.py
@@ -27,9 +27,9 @@ from fastapi import FastAPI
from fastapi.routing import Mount
from airflow.api_fastapi.common.dagbag import create_dag_bag
+from airflow.api_fastapi.common.exceptions import init_error_handlers
from airflow.api_fastapi.core_api.app import (
init_config,
- init_error_handlers,
init_flask_plugins,
init_middlewares,
init_views,
@@ -128,7 +128,6 @@ def create_app(apps: str = "all") -> FastAPI:
if "all" in apps_list or "execution" in apps_list:
task_exec_api_app = create_task_execution_api_app()
task_exec_api_app.state.dag_bag = dag_bag
- init_error_handlers(task_exec_api_app)
app.mount("/execution", task_exec_api_app)
if "all" in apps_list or "core" in apps_list:
diff --git a/airflow-core/src/airflow/api_fastapi/common/exceptions.py
b/airflow-core/src/airflow/api_fastapi/common/exceptions.py
index 86af9e062cd..00cae2bbdd1 100644
--- a/airflow-core/src/airflow/api_fastapi/common/exceptions.py
+++ b/airflow-core/src/airflow/api_fastapi/common/exceptions.py
@@ -23,15 +23,15 @@ from abc import ABC, abstractmethod
from enum import Enum
from typing import Generic, TypeVar
-from fastapi import HTTPException, Request, status
-from sqlalchemy.exc import DatabaseError, DataError, IntegrityError
+from fastapi import FastAPI, HTTPException, Request, status
+from sqlalchemy.exc import DataError, IntegrityError, SQLAlchemyError
from airflow.configuration import conf
from airflow.exceptions import DeserializationError
from airflow.utils.strings import get_random_string
T = TypeVar("T", bound=Exception)
-DBError = TypeVar("DBError", bound=DatabaseError)
+DBError = TypeVar("DBError", bound=SQLAlchemyError)
log = logging.getLogger(__name__)
@@ -70,35 +70,41 @@ class _DatabaseErrorHandler(BaseErrorHandler[DBError]):
def _should_handle(self, exc: DBError) -> bool:
return True
- def exception_handler(self, request: Request, exc: DBError):
- if not self._should_handle(exc):
- return
+ def _raise_database_error_response(self, exc: DBError) -> None:
+ statement = getattr(exc, "statement", "hidden")
+ orig_error = getattr(exc, "orig", "hidden")
exception_id = get_random_string()
stacktrace = "".join(traceback.format_tb(exc.__traceback__))
- log_message = f"Error with id {exception_id}, statement:
{exc.statement}\n{stacktrace}"
+ log_message = f"Error with id {exception_id}, statement:
{statement}\n{stacktrace}"
log.error(log_message)
+
if conf.get("api", "expose_stacktrace") == "True":
message = log_message
- statement = str(exc.statement)
- orig_error = str(exc.orig)
+ statement_out = str(statement)
+ orig_error_out = str(orig_error)
else:
message = (
"Serious error when handling your request. Check logs for more
details - "
f"you will find it in api server when you look for ID
{exception_id}"
)
- statement = "hidden"
- orig_error = "hidden"
+ statement_out = "hidden"
+ orig_error_out = "hidden"
raise HTTPException(
status_code=self.status_code,
detail={
"reason": self.reason,
- "statement": statement,
- "orig_error": orig_error,
+ "statement": statement_out,
+ "orig_error": orig_error_out,
"message": message,
},
)
+ def exception_handler(self, request: Request, exc: DBError):
+ if not self._should_handle(exc):
+ return
+ self._raise_database_error_response(exc)
+
class _UniqueConstraintErrorHandler(_DatabaseErrorHandler[IntegrityError]):
"""Translate a unique-constraint ``IntegrityError`` into a 409, matched
per database dialect."""
@@ -158,8 +164,24 @@ class
DagErrorHandler(BaseErrorHandler[DeserializationError]):
)
+class SQLAlchemyErrorHandler(_DatabaseErrorHandler[SQLAlchemyError]):
+ """Generic handler for SQLAlchemyError -> 500 responses."""
+
+ status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+ reason = "Database error"
+
+ def __init__(self):
+ super().__init__(SQLAlchemyError)
+
+
ERROR_HANDLERS: list[BaseErrorHandler] = [
_UniqueConstraintErrorHandler(),
DataErrorHandler(),
+ SQLAlchemyErrorHandler(),
DagErrorHandler(),
]
+
+
+def init_error_handlers(app: FastAPI) -> None:
+ for handler in ERROR_HANDLERS:
+ app.add_exception_handler(handler.exception_cls,
handler.exception_handler)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/app.py
b/airflow-core/src/airflow/api_fastapi/core_api/app.py
index 27213a4c0a2..961f13fbc71 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/app.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/app.py
@@ -170,13 +170,6 @@ def init_config(app: FastAPI) -> None:
app.state.secret_key = get_signing_key("api", "secret_key")
-def init_error_handlers(app: FastAPI) -> None:
- from airflow.api_fastapi.common.exceptions import ERROR_HANDLERS
-
- for handler in ERROR_HANDLERS:
- app.add_exception_handler(handler.exception_cls,
handler.exception_handler)
-
-
def init_middlewares(app: FastAPI) -> None:
from airflow.api_fastapi.app import get_auth_manager
from airflow.api_fastapi.auth.middlewares.refresh_token import
JWTRefreshMiddleware
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/app.py
b/airflow-core/src/airflow/api_fastapi/execution_api/app.py
index 5a87cdb81e0..f9e7cb17250 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/app.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/app.py
@@ -287,6 +287,7 @@ def _inject_trace_context_dep(routes, mode: str) -> None:
def create_task_execution_api_app(lifespan: svcs.fastapi.lifespan = lifespan)
-> FastAPI:
"""Create FastAPI app for task execution API."""
+ from airflow.api_fastapi.common.exceptions import init_error_handlers
from airflow.api_fastapi.execution_api.routes import execution_api_router
from airflow.api_fastapi.execution_api.versions import bundle
from airflow.configuration import conf
@@ -314,6 +315,7 @@ def create_task_execution_api_app(lifespan:
svcs.fastapi.lifespan = lifespan) ->
_inject_trace_context_dep(execution_api_router.routes, mode)
app.generate_and_include_versioned_routers(execution_api_router)
+ init_error_handlers(app)
# As we are mounted as a sub app, we don't get any logs for unhandled
exceptions without this!
@app.exception_handler(Exception)
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index c1bac796023..34c3dc35406 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -319,10 +319,8 @@ def ti_run(
# Let the app-level DataErrorHandler return a 422 (not the opaque 500
below).
raise
except SQLAlchemyError:
- log.exception("Error marking Task Instance state as running")
- raise HTTPException(
- status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Database error occurred"
- )
+ # Defer to app-level SQLAlchemyError handler (returns HTTP 500).
+ raise
# JWTReissueMiddleware also writes Refreshed-API-Token but skips workload
tokens, so we set it here for the workload→execution swap.
if token.claims.scope == "workload":
@@ -502,11 +500,9 @@ def ti_update_state(
except DataError:
# Let DataErrorHandler return a 422 (not the opaque 500 below).
raise
- except SQLAlchemyError as e:
- log.error("Error updating Task Instance state", error=str(e))
- raise HTTPException(
- status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Database error occurred"
- )
+ except SQLAlchemyError:
+ # Defer to app-level SQLAlchemyError handler (returns HTTP 500).
+ raise
if updated_state == TaskInstanceState.SUCCESS:
if conf.getboolean("state_store", "clear_on_success"):
diff --git a/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py
b/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py
index 7f470e2f019..b9b18c893d0 100644
--- a/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py
+++ b/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py
@@ -497,6 +497,46 @@ class TestDataErrorHandler:
assert detail["statement"] == "hidden"
assert detail["orig_error"] == "hidden"
+ @conf_vars({("api", "expose_stacktrace"): "False"})
+ @patch("airflow.api_fastapi.common.exceptions.get_random_string",
return_value=MOCKED_ID)
+ def test_sqlalchemy_error_dispatched_through_fastapi_app(self,
mock_get_random_string) -> None:
+ """End-to-end: a route raising SQLAlchemyError returns 500 via the
registered handler."""
+ from sqlalchemy.exc import SQLAlchemyError
+
+ app = FastAPI()
+ for h in ERROR_HANDLERS:
+ app.add_exception_handler(h.exception_cls, h.exception_handler)
+
+ @app.post("/test")
+ def trigger_error():
+ raise SQLAlchemyError("boom")
+
+ response = TestClient(app, raise_server_exceptions=False).post("/test")
+ assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+ detail = response.json()["detail"]
+ assert detail["reason"] == "Database error"
+ assert detail["message"] == MESSAGE
+
+ @conf_vars({("api", "expose_stacktrace"): "True"})
+ @patch("airflow.api_fastapi.common.exceptions.get_random_string",
return_value=MOCKED_ID)
+ def test_sqlalchemy_error_includes_traceback_with_stacktrace(self,
mock_get_random_string) -> None:
+ """End-to-end: SQLAlchemyError exposes traceback details when
stacktrace logging is enabled."""
+ from sqlalchemy.exc import SQLAlchemyError
+
+ app = FastAPI()
+ for h in ERROR_HANDLERS:
+ app.add_exception_handler(h.exception_cls, h.exception_handler)
+
+ @app.post("/test")
+ def trigger_error():
+ raise SQLAlchemyError("boom")
+
+ response = TestClient(app, raise_server_exceptions=False).post("/test")
+ assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+ detail = response.json()["detail"]
+ assert detail["reason"] == "Database error"
+ assert "trigger_error" in detail["message"]
+
class TestDagErrorHandler:
@pytest.mark.parametrize(
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/test_app.py
b/airflow-core/tests/unit/api_fastapi/execution_api/test_app.py
index d4b9ce5ac88..bb2d2d557dc 100644
--- a/airflow-core/tests/unit/api_fastapi/execution_api/test_app.py
+++ b/airflow-core/tests/unit/api_fastapi/execution_api/test_app.py
@@ -24,11 +24,12 @@ from uuid import UUID
import httpx
import pytest
-from fastapi import Request
+from fastapi import Request, status
from fastapi.params import Security as SecurityParam
from fastapi.routing import APIRoute
from fastapi.testclient import TestClient
from opentelemetry import context as otel_context, propagate as otel_propagate
+from sqlalchemy.exc import SQLAlchemyError
from airflow.api_fastapi.execution_api.app import (
InProcessExecutionAPI,
@@ -64,6 +65,21 @@ def test_access_api_contract(client):
assert response.headers["airflow-api-version"] == bundle.versions[0].value
+@conf_vars({("api", "expose_stacktrace"): "False"})
[email protected]("airflow.api_fastapi.common.exceptions.get_random_string",
return_value="test-error-id")
+def
test_direct_execution_api_app_handles_sqlalchemy_errors(mock_get_random_string):
+ app = create_task_execution_api_app()
+
+ @app.get("/test-sqlalchemy-error")
+ def trigger_error():
+ raise SQLAlchemyError("boom")
+
+ response = TestClient(app,
raise_server_exceptions=False).get("/test-sqlalchemy-error")
+
+ assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+ assert response.json()["detail"]["reason"] == "Database error"
+
+
def test_ti_self_routes_have_task_instance_id_param(client):
"""Every route with ti:self scope must have a {task_instance_id} path
parameter."""
app = client.app
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index 542ce7eaaf1..8a152bebe0d 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
@@ -18,6 +18,7 @@
from __future__ import annotations
from datetime import datetime
+from types import SimpleNamespace
from typing import TYPE_CHECKING
from unittest import mock
from uuid import UUID, uuid4
@@ -1497,7 +1498,64 @@ class TestTIUpdateState:
mock_register_asset_changes_in_db.return_value = None
response =
client.patch(f"/execution/task-instances/{ti.id}/state", json=payload)
assert response.status_code == 500
- assert response.json()["detail"] == "Database error occurred"
+ detail = response.json()["detail"]
+ assert isinstance(detail, dict)
+ assert detail.get("reason") == "Database error"
+
+ def test_ti_run_database_error(self, client, session,
create_task_instance):
+ """
+ Test that a database error is handled correctly when starting the Task
Instance.
+ """
+ ti = create_task_instance(
+ task_id="test_ti_run_database_error",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ dag_id=str(uuid4()),
+ )
+ session.commit()
+
+ payload = {
+ "state": "running",
+ "hostname": "hostname",
+ "unixname": "unixname",
+ "pid": 123,
+ "start_date": "2024-10-31T12:00:00Z",
+ }
+
+ with mock.patch(
+ "airflow.api_fastapi.common.db.common.Session.execute",
+ side_effect=[
+ mock.Mock(
+ one=mock.Mock(
+ return_value=SimpleNamespace(
+ state="queued",
+ dag_id="dag",
+ run_id="run",
+ task_id="task",
+ map_index=-1,
+ try_number=1,
+ max_tries=0,
+ start_date=None,
+ next_method=None,
+ hostname=None,
+ unixname=None,
+ pid=None,
+ next_kwargs=None,
+ logical_date=timezone.utcnow(),
+ owners="test_owner",
+ )
+ )
+ ),
+ SQLAlchemyError("Database error"),
+ ],
+ ):
+ response = client.patch(f"/execution/task-instances/{ti.id}/run",
json=payload)
+
+ assert response.status_code == 500
+ detail = response.json()["detail"]
+ assert isinstance(detail, dict)
+ assert detail.get("reason") == "Database error"
@pytest.mark.parametrize("queues_enabled", [False, True])
def test_ti_update_state_to_deferred(
