Copilot commented on code in PR #65162:
URL: https://github.com/apache/airflow/pull/65162#discussion_r3074463620
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py:
##########
@@ -453,7 +453,7 @@ def get_node_summaries() -> Iterable[dict[str, Any]]:
)
def get_grid_ti_summaries_stream(
dag_id: str,
- session: SessionDep,
+ session: Annotated[Session, Depends(_get_session)],
Review Comment:
`_get_session` is a module-private helper (leading underscore) but is now
imported/used directly from another module. To keep dependency usage consistent
and avoid coupling to a private symbol, consider exposing a public
request-scoped session dependency (e.g. `RequestSessionDep = Annotated[Session,
Depends(_get_session)]`) in `common.db.common` and using that here (or
rename/export the helper) instead of importing `_get_session` directly.
##########
airflow-core/tests/unit/api_fastapi/core_api/test_app.py:
##########
@@ -16,13 +16,71 @@
# under the License.
from __future__ import annotations
+import typing
+
import pytest
+from fastapi.params import Depends as DependsClass
+from fastapi.responses import StreamingResponse
+from starlette.routing import Mount
+
+from airflow.api_fastapi.app import create_app
from tests_common.test_utils.db import clear_db_jobs
pytestmark = pytest.mark.db_test
+def _get_all_api_routes(app):
+ """Recursively yield all APIRoutes from the app and its mounted
sub-apps."""
+ for route in getattr(app, "routes", []):
+ if isinstance(route, Mount) and hasattr(route, "app"):
+ yield from _get_all_api_routes(route.app)
+ if hasattr(route, "endpoint"):
+ yield route
+
+
+class TestStreamingEndpointSessionScope:
+ def test_no_streaming_endpoint_uses_function_scoped_depends(self):
+ """Streaming endpoints must not use function-scoped generator
dependencies.
+
+ FastAPI's ``function_stack`` (used for ``scope="function"``
dependencies)
+ is torn down after the route handler returns but *before* the response
body
+ is sent. For ``StreamingResponse`` endpoints the response body is
produced
+ by a generator that runs during sending, so any generator dependency
with
+ ``scope="function"`` will have its cleanup run before the generator
+ executes. This causes the generator to silently reopen the session via
+ autobegin, and the resulting connection is never returned to the pool.
+ """
+ app = create_app()
+ violations = []
+ for route in _get_all_api_routes(app):
+ try:
+ hints = typing.get_type_hints(route.endpoint,
include_extras=True)
+ except Exception:
+ continue
+ if hints.get("return") is not StreamingResponse:
+ continue
Review Comment:
This test only flags routes whose *return type annotation* is exactly
`StreamingResponse`, and it silently skips any endpoint where
`typing.get_type_hints()` raises. That combination can allow streaming handlers
to evade the check (e.g. endpoints that return a `StreamingResponse` without an
explicit return annotation, or future endpoints whose annotations can’t be
resolved at runtime). Consider identifying streaming routes via the FastAPI
route metadata (e.g. `APIRoute.response_class is StreamingResponse`) and/or
failing the test when type-hint inspection can’t be performed so the guardrail
can’t be bypassed unintentionally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
