ashb · Oct 1, 2026 · Sep 28, 2026 · Oct 1, 2026
@@ -27,6 +27,7 @@
 from fastapi.routing import Mount
 from starlette.middleware import Middleware
 
+from airflow import settings
 from airflow.api_fastapi.common.dagbag import create_dag_bag
 from airflow.api_fastapi.common.exceptions import init_error_handlers
 from airflow.api_fastapi.common.http_access_log import HttpAccessLogMiddleware
@@ -100,6 +101,7 @@ def _initialize_api_server_stats() -> None:
 async def lifespan(app: FastAPI):
     _initialize_api_server_stats()
     async with AsyncExitStack() as stack:
+        stack.push_async_callback(settings.dispose_async_engine)
         for route in app.routes:
             if isinstance(route, Mount) and isinstance(route.app, FastAPI):
                 await stack.enter_async_context(

@@ -38,6 +38,7 @@
 from opentelemetry import context as otel_context, propagate as otel_propagate
 from starlette.middleware.base import BaseHTTPMiddleware
 
+from airflow import settings
 from airflow.api_fastapi.auth.tokens import (
     JWTGenerator,
     JWTValidator,
@@ -437,6 +438,7 @@ def transport(self) -> httpx.WSGITransport:
 
         # https://github.com/abersheeran/a2wsgi/discussions/64
         async def start_lifespan(cm: AsyncExitStack, app: FastAPI):
+            cm.push_async_callback(settings.dispose_async_engine)
             await cm.enter_async_context(app.router.lifespan_context(app))
 
         cm = AsyncExitStack()

diff --git a/airflow-core/src/airflow/settings.py b/airflow-core/src/airflow/settings.py
@@ -427,13 +427,7 @@ def create_async_metadata_engine(
 
 
 def _configure_async_session() -> None:
-    """
-    Configure async SQLAlchemy session.
-
-    This exists so tests can reconfigure the session. How SQLAlchemy configures
-    this does not work well with Pytest and you can end up with issues when the
-    session and runs in a different event loop from the test itself.
-    """
+    """Configure the async engine and session factory."""
     global AsyncSession, async_engine
 
     if not SQL_ALCHEMY_CONN_ASYNC:
@@ -670,6 +664,12 @@ def dispose_orm(do_log: bool = True):
         AsyncSession = None
 
 
+async def dispose_async_engine() -> None:
+    """Close checked-in connections on their event loop, retaining the engine and session factory."""
+    if async_engine is not None:
+        await async_engine.dispose()
+
+
 def reconfigure_orm(disable_connection_pool=False, pool_class=None):
     """Properly close database connections and re-configure ORM."""
     dispose_orm()

diff --git a/airflow-core/tests/unit/api_fastapi/auth/managers/simple/conftest.py b/airflow-core/tests/unit/api_fastapi/auth/managers/simple/conftest.py
@@ -75,4 +75,5 @@ def test_client():
             ): "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager"
         }
     ):
-        return TestClient(create_app("core"))
+        with TestClient(create_app("core")) as client:
+            yield client
diff --git a/airflow-core/tests/unit/api_fastapi/conftest.py b/airflow-core/tests/unit/api_fastapi/conftest.py
@@ -133,11 +133,12 @@ def _authed_test_client(app: FastAPI, request):
             ),
         )
     with mock.patch("airflow.models.revoked_token.RevokedToken.is_revoked", return_value=False):
-        yield TestClient(
+        with TestClient(
             app,
             headers={"Authorization": f"Bearer {token}"},
             base_url=f"{BASE_URL}{get_api_path(request)}",
-        )
+        ) as test_client:
+            yield test_client
 
 
 @pytest.fixture
@@ -167,22 +168,28 @@ def fresh_test_client(request):
 
 @pytest.fixture
 def unauthenticated_test_client(request, _isolated_shared_app):
-    return TestClient(_isolated_shared_app, base_url=f"{BASE_URL}{get_api_path(request)}")
+    with TestClient(_isolated_shared_app, base_url=f"{BASE_URL}{get_api_path(request)}") as test_client:
+        yield test_client
 
 
 @pytest.fixture
-def unauthorized_test_client(request, _isolated_shared_app):
-    app = _isolated_shared_app
-    auth_manager: SimpleAuthManager = app.state.auth_manager
+def unauthorized_headers(_isolated_shared_app):
+    auth_manager: SimpleAuthManager = _isolated_shared_app.state.auth_manager
     token = auth_manager._get_token_signer().generate(
         auth_manager.serialize_user(SimpleAuthManagerUser(username="dummy", role=None))
     )
+    return {"Authorization": f"Bearer {token}"}
+
+
+@pytest.fixture
+def unauthorized_test_client(request, _isolated_shared_app, unauthorized_headers):
     with mock.patch("airflow.models.revoked_token.RevokedToken.is_revoked", return_value=False):
-        yield TestClient(
-            app,
-            headers={"Authorization": f"Bearer {token}"},
+        with TestClient(
+            _isolated_shared_app,
+            headers=unauthorized_headers,
             base_url=f"{BASE_URL}{get_api_path(request)}",
-        )
+        ) as test_client:
+            yield test_client
 
 
 @pytest.fixture

diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_auth.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_auth.py
@@ -22,6 +22,7 @@
 
 import jwt
 import pytest
+from fastapi.testclient import TestClient
 
 from airflow.api_fastapi.auth.managers.base_auth_manager import COOKIE_NAME_JWT_TOKEN
 from airflow.models.revoked_token import RevokedToken
@@ -169,10 +170,7 @@ def cleanup_revoked_tokens(self):
         clear_db_revoked_tokens()
 
     @pytest.fixture
-    def logout_client(self):
-        """A test client without the is_revoked mock so revocation tests hit the real DB."""
-        from fastapi.testclient import TestClient
-
+    def logout_app(self):
         from airflow.api_fastapi.app import create_app
 
         with conf_vars(
@@ -183,8 +181,13 @@ def logout_client(self):
                 ): "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager"
             }
         ):
-            app = create_app()
-            yield TestClient(app, base_url="http://testserver/api/v2")
+            yield create_app()
+
+    @pytest.fixture
+    def logout_client(self, logout_app):
+        """A test client without the is_revoked mock so revocation tests hit the real DB."""
+        with TestClient(logout_app, base_url="http://testserver/api/v2") as client:
+            yield client
 
     def test_logout_revokes_token(self, logout_client):
         """Test that logout revokes the JWT token and persists it in the database."""
@@ -282,7 +285,7 @@ def test_logout_revokes_every_credential_presented(self, logout_client):
         assert RevokedToken.is_revoked("test-jti-both-bearer") is True
         assert RevokedToken.is_revoked("test-jti-both-cookie") is True
 
-    def test_logout_revokes_both_even_when_a_trusted_user_is_cached(self, logout_client):
+    def test_logout_revokes_both_even_when_a_trusted_user_is_cached(self, logout_app):
         """The trusted-middleware shortcut must not change what logout revokes.
 
         On protected routes `get_user()` can return a user cached by JWTRefreshMiddleware
@@ -291,7 +294,7 @@ def test_logout_revokes_both_even_when_a_trusted_user_is_cached(self, logout_cli
         """
         from airflow.api_fastapi.core_api.security import USER_INJECTED_BY_TRUSTED_MIDDLEWARE
 
-        auth_manager = logout_client.app.state.auth_manager
+        auth_manager = logout_app.state.auth_manager
         bearer_token = self._mint(auth_manager, "test-jti-trusted-bearer")
         cookie_token = self._mint(auth_manager, "test-jti-trusted-cookie")
 
@@ -300,9 +303,12 @@ async def _inject(request, call_next):
             request.state.user_authenticated_via = USER_INJECTED_BY_TRUSTED_MIDDLEWARE
             return await call_next(request)
 
-        logout_client.app.middleware("http")(_inject)
-        logout_client.cookies.set(COOKIE_NAME_JWT_TOKEN, cookie_token)
-        with patch.object(auth_manager, "get_url_logout", return_value=None):
+        logout_app.middleware("http")(_inject)
+        with (
+            TestClient(logout_app, base_url="http://testserver/api/v2") as logout_client,
+            patch.object(auth_manager, "get_url_logout", return_value=None),
+        ):
+            logout_client.cookies.set(COOKIE_NAME_JWT_TOKEN, cookie_token)
             response = logout_client.get(
                 "/auth/logout",
                 headers={"Authorization": f"Bearer {bearer_token}"},

diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
@@ -23,7 +23,6 @@
 
 import pendulum
 import pytest
-from fastapi.testclient import TestClient
 from sqlalchemy import and_, func, select
 from sqlalchemy.exc import OperationalError, ProgrammingError
 
@@ -83,18 +82,13 @@ def clean_db():
 
 
 @pytest.fixture
-def dag_reader_test_client(test_client):
+def dag_reader_headers(test_client):
     """A caller who may read the Dags but not write them: viewer is below the role edits require."""
     auth_manager = test_client.app.state.auth_manager
     token = auth_manager._get_token_signer().generate(
         auth_manager.serialize_user(SimpleAuthManagerUser(username="reader", role="viewer"))
     )
-    with mock.patch("airflow.models.revoked_token.RevokedToken.is_revoked", return_value=False):
-        yield TestClient(
-            test_client.app,
-            headers={"Authorization": f"Bearer {token}"},
-            base_url=str(test_client.base_url),
-        )
+    return {"Authorization": f"Bearer {token}"}
 
 
 def make_dags():
@@ -1599,22 +1593,24 @@ def test_pause_backfill_404_when_the_dag_is_unreadable(self, session, unauthoriz
         response = unauthorized_test_client.put(f"/backfills/{backfill.id}/pause")
         assert response.status_code == 404
 
-    def test_pause_backfill_403(self, session, dag_reader_test_client):
+    def test_pause_backfill_403(self, session, dag_reader_headers, test_client):
         (dag,) = self._create_dag_models()
         from_date = timezone.utcnow()
         to_date = timezone.utcnow()
         backfill = Backfill(dag_id=dag.dag_id, from_date=from_date, to_date=to_date)
         session.add(backfill)
         session.commit()
-        response = dag_reader_test_client.put(f"/backfills/{backfill.id}/pause")
+        response = test_client.put(f"/backfills/{backfill.id}/pause", headers=dag_reader_headers)
         assert response.status_code == 403
 
     def test_pause_backfill_unknown_id_is_not_authorized_by_a_body_dag_id(
-        self, session, dag_reader_test_client
+        self, session, dag_reader_headers, test_client
     ):
         (dag,) = self._create_dag_models()
         session.commit()
-        response = dag_reader_test_client.put(f"/backfills/{231984098}/pause", json={"dag_id": dag.dag_id})
+        response = test_client.put(
+            f"/backfills/{231984098}/pause", json={"dag_id": dag.dag_id}, headers=dag_reader_headers
+        )
         assert response.status_code == 404
         assert response.json().get("detail") == "Backfill not found"
 

diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py
@@ -1732,16 +1732,15 @@ def test_get_status_requires_token_header(self, test_client):
         assert response.status_code == 422
 
     @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
-    def test_get_status_unauthorized_user_does_not_leak_row(
-        self, test_client, unauthorized_test_client, session
-    ):
+    def test_get_status_unauthorized_user_does_not_leak_row(self, test_client, unauthorized_headers, session):
         """A user without rights on the conn_id never sees the row payload via GET-by-token."""
         post_response = test_client.post("/connections/enqueue-test", json=self.TEST_REQUEST_BODY)
         assert post_response.status_code == 202
         token = post_response.json()["token"]
 
-        response = unauthorized_test_client.get(
-            "/connections/enqueue-test", headers={"Airflow-Connection-Test-Token": token}
+        response = test_client.get(
+            "/connections/enqueue-test",
+            headers={**unauthorized_headers, "Airflow-Connection-Test-Token": token},
         )
         assert response.status_code in (401, 403, 404)
         body = (

diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_bundles.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_bundles.py
@@ -21,7 +21,6 @@
 from unittest import mock
 
 import pytest
-from fastapi.testclient import TestClient
 from itsdangerous import URLSafeSerializer
 from sqlalchemy import insert, update
 
@@ -275,7 +274,7 @@ def deny_dagless_view(*, access_view, user, team_name=None):
 
 
 @pytest.fixture
-def viewer_client(test_client, readable_dag_ids):
+def viewer_headers(test_client, readable_dag_ids):
     """
     A viewer with the same readable Dags: may read import errors, but not the admin-gated view.
 
@@ -286,12 +285,7 @@ def viewer_client(test_client, readable_dag_ids):
     token = auth_manager._get_token_signer().generate(
         auth_manager.serialize_user(SimpleAuthManagerUser(username="viewer", role="viewer"))
     )
-    with mock.patch("airflow.models.revoked_token.RevokedToken.is_revoked", return_value=False):
-        yield TestClient(
-            test_client.app,
-            headers={"Authorization": f"Bearer {token}"},
-            base_url=str(test_client.base_url),
-        )
+    return {"Authorization": f"Bearer {token}"}
 
 
 class TestGetDagBundles:
@@ -325,15 +319,15 @@ def test_excludes_a_bundle_with_no_dags_without_the_admin_view(self, dag_scoped_
         assert DAGLESS_BUNDLE not in [bundle["name"] for bundle in body["dag_bundles"]]
         assert body["total_entries"] == 3
 
-    def test_hides_a_bundle_with_no_dags_from_a_viewer(self, viewer_client):
+    def test_hides_a_bundle_with_no_dags_from_a_viewer(self, viewer_headers, test_client):
         """
         The bundle name and its version are the disclosure, so a viewer must not get them.
 
         Same reasoning as the admin gate on import errors for a file that never registered a Dag:
         with no Dag to authorize against, there is nothing weaker than the admin view to fall back
         on.
         """
-        body = viewer_client.get("/dagBundles").json()
+        body = test_client.get("/dagBundles", headers=viewer_headers).json()
 
         assert DAGLESS_BUNDLE not in [bundle["name"] for bundle in body["dag_bundles"]]
         # Absent from the count too, so its existence does not leak through pagination.
@@ -381,12 +375,12 @@ def test_lists_a_bundle_whose_dags_are_all_stale(self, dag_scoped_client, sessio
 
         assert GIT_BUNDLE in [bundle["name"] for bundle in body["dag_bundles"]]
 
-    def test_returns_nothing_when_no_dag_is_readable(self, viewer_client):
+    def test_returns_nothing_when_no_dag_is_readable(self, viewer_headers, test_client):
         # The fixture already patched this attribute, so retarget its mock rather than nesting a
         # second autospec patch over it.
-        viewer_client.app.state.auth_manager.get_authorized_dag_ids.return_value = set()
+        test_client.app.state.auth_manager.get_authorized_dag_ids.return_value = set()
 
-        body = viewer_client.get("/dagBundles").json()
+        body = test_client.get("/dagBundles", headers=viewer_headers).json()
 
         assert body == {"dag_bundles": [], "total_entries": 0}
 
@@ -483,7 +477,9 @@ def test_deactivated_bundle_is_still_listed(self, dag_scoped_client):
         assert bundle["active"] is False
         assert bundle["version"] == "deadbeef"
 
-    def test_import_error_count_authorizes_on_the_same_terms_as_import_errors(self, viewer_client):
+    def test_import_error_count_authorizes_on_the_same_terms_as_import_errors(
+        self, viewer_headers, test_client
+    ):
         """
         Count on the same terms as ``GET /importErrors``, not "every row for this bundle".
 
@@ -498,7 +494,7 @@ def test_import_error_count_authorizes_on_the_same_terms_as_import_errors(self,
 
         Dropping either restriction takes the count to 2, so one assertion pins both halves.
         """
-        body = viewer_client.get("/dagBundles").json()
+        body = test_client.get("/dagBundles", headers=viewer_headers).json()
         bundle = next(b for b in body["dag_bundles"] if b["name"] == GIT_BUNDLE)
 
         assert bundle["import_error_count"] == 1
@@ -755,15 +751,18 @@ def test_shows_a_bundle_with_no_dags_to_a_caller_holding_the_admin_view(self, ad
     def test_404_for_an_unknown_bundle(self, dag_scoped_client):
         assert dag_scoped_client.get("/dagBundles/no_such_bundle").status_code == 404
 
-    def test_import_error_count_is_gated_like_the_collection(self, admin_client, viewer_client):
+    def test_import_error_count_is_gated_like_the_collection(self, admin_client, viewer_headers):
         """
         The admin sees the unregistered-file error as well; the viewer sees only the registered one.
 
         Same two-part authorization as ``GET /importErrors``, asserted here so the detail route
         cannot drift from the collection route it shares a helper with.
         """
         assert admin_client.get(f"/dagBundles/{GIT_BUNDLE}").json()["import_error_count"] == 2
-        assert viewer_client.get(f"/dagBundles/{GIT_BUNDLE}").json()["import_error_count"] == 1
+        assert (
+            admin_client.get(f"/dagBundles/{GIT_BUNDLE}", headers=viewer_headers).json()["import_error_count"]
+            == 1
+        )
 
     def test_import_error_count_is_withheld_without_permission(self, admin_client):
         auth_manager = admin_client.app.state.auth_manager
@@ -856,14 +855,14 @@ def test_a_file_that_registered_no_dag_is_listed_with_its_error(self, admin_clie
         assert by_path[UNREGISTERED_FILE]["last_parsed_time"] is None
         assert by_path[UNREGISTERED_FILE]["last_parse_duration"] is None
 
-    def test_excludes_a_file_whose_dag_is_not_readable(self, admin_client, viewer_client):
+    def test_excludes_a_file_whose_dag_is_not_readable(self, admin_client, viewer_headers):
         """``UNREADABLE_FILE`` is registered, so only the readable-Dag filter keeps it out."""
-        for client in (admin_client, viewer_client):
-            body = client.get(f"/dagBundles/{GIT_BUNDLE}/files").json()
+        for headers in ({}, viewer_headers):
+            body = admin_client.get(f"/dagBundles/{GIT_BUNDLE}/files", headers=headers).json()
             assert UNREADABLE_FILE not in {file["relative_fileloc"] for file in body["dag_bundle_files"]}
 
-    def test_viewer_does_not_see_the_unregistered_file(self, viewer_client):
-        body = viewer_client.get(f"/dagBundles/{GIT_BUNDLE}/files").json()
+    def test_viewer_does_not_see_the_unregistered_file(self, viewer_headers, test_client):
+        body = test_client.get(f"/dagBundles/{GIT_BUNDLE}/files", headers=viewer_headers).json()
 
         assert [file["relative_fileloc"] for file in body["dag_bundle_files"]] == [REGISTERED_FILE]
         assert body["total_entries"] == 1