[v3-2-test] Fix get_team_name_dep creating wasted async sessions when multi_team=False (#65275)

airflow-core/src/airflow/api_fastapi/execution_api/security.py

@@ -74,10 +74,8 @@
 from fastapi.params import Security as SecurityParam
 from fastapi.routing import APIRoute
 from fastapi.security import HTTPBearer, SecurityScopes
-from sqlalchemy import select
 
 from airflow.api_fastapi.auth.tokens import JWTValidator
-from airflow.api_fastapi.common.db.common import AsyncSessionDep
 from airflow.api_fastapi.execution_api.datamodels.token import TIToken
 from airflow.api_fastapi.execution_api.deps import DepContainer
 
@@ -222,16 +220,20 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.allowed_token_types = frozenset(token_scopes) if token_scopes else frozenset({"execution"})
 
 
-async def get_team_name_dep(session: AsyncSessionDep, token=CurrentTIToken) -> str | None:
+async def get_team_name_dep(token=CurrentTIToken) -> str | None:
     """Return the team name associated to the task (if any)."""
     from airflow.configuration import conf
-    from airflow.models import DagModel, TaskInstance
-    from airflow.models.dagbundle import DagBundleModel
-    from airflow.models.team import Team
 
     if not conf.getboolean("core", "multi_team"):
         return None
 
+    from sqlalchemy import select
+
+    from airflow.models import DagModel, TaskInstance
+    from airflow.models.dagbundle import DagBundleModel
+    from airflow.models.team import Team
+    from airflow.utils.session import create_session_async
+
     stmt = (
         select(Team.name)
         .select_from(TaskInstance)
@@ -240,4 +242,5 @@ async def get_team_name_dep(session: AsyncSessionDep, token=CurrentTIToken) -> s
         .join(DagBundleModel.teams)
         .where(TaskInstance.id == token.id)
     )
-    return await session.scalar(stmt)
+    async with create_session_async() as session:
+        return await session.scalar(stmt)

airflow-core/tests/unit/api_fastapi/execution_api/test_security.py

@@ -16,14 +16,20 @@
 # under the License.
 from __future__ import annotations
 
+from unittest.mock import MagicMock, patch
 from uuid import UUID
 
 import pytest
 from fastapi import APIRouter, FastAPI, Request, Security
 from fastapi.testclient import TestClient
 
 from airflow.api_fastapi.execution_api.datamodels.token import TIToken
-from airflow.api_fastapi.execution_api.security import ExecutionAPIRoute, _jwt_bearer, require_auth
+from airflow.api_fastapi.execution_api.security import (
+    ExecutionAPIRoute,
+    _jwt_bearer,
+    get_team_name_dep,
+    require_auth,
+)
 
 
 class TestExecutionAPIRoute:
@@ -136,3 +142,21 @@ def test_execution_token_accepted_on_both_routes(self, token_type_app):
         run = client.get(f"/task-instances/{self.TI_ID}/run", headers={"Authorization": "Bearer fake"})
         assert state.status_code == 200
         assert run.status_code == 200
+
+
+class TestGetTeamNameDep:
+    """Tests for get_team_name_dep avoiding unnecessary async sessions."""
+
+    @pytest.mark.asyncio
+    async def test_returns_none_without_session_when_multi_team_disabled(self):
+        """When multi_team=False, no async session should be created."""
+        token = MagicMock(spec=TIToken)
+
+        with (
+            patch("airflow.configuration.conf.getboolean", return_value=False),
+            patch("airflow.utils.session.create_session_async") as mock_create_session,
+        ):
+            result = await get_team_name_dep(token=token)
+
+        assert result is None
+        mock_create_session.assert_not_called()