failed WRONG_FIX UNSUBMITTED wrong_fix_unsubmitted(None) · None tool calls · 0 s · fastapi/fastapi
🐛 Fix support for `StreamingResponse`s with dependencies with `yield` or `UploadFile`s, close after the response is done 🐛 Fix support for `StreamingResponse`s with dependencies with `yield` or `UploadFile`s, close after the response is done --- There's (again) an `AsyncExitStack` middleware, now used mainly to close objects after the response is done. There's also a new wrapper duplicating some logic in Starlette but creating an `AsyncExitStack` for the exit code in dependencies with `yield`, for regular responses and WebSockets as well. The exit code of dependencies with `yield` is now executed after the response is done.
| # | Tool | Arguments | Result |
|---|---|---|---|
| No trace captured. | |||
--- a/docs_src/dependencies/tutorial013_an_py310.py
+++ b/docs_src/dependencies/tutorial013_an_py310.py
@@ -0,0 +1,38 @@
+import time
+from typing import Annotated
+
+from fastapi import Depends, FastAPI, HTTPException
+from fastapi.responses import StreamingResponse
+from sqlmodel import Field, Session, SQLModel, create_engine
+
+engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")
+
+
+class User(SQLModel, table=True):
+ id: int | None = Field(default=None, primary_key=True)
+ name: str
+
+
+app = FastAPI()
+
+
+def get_session():
+ with Session(engine) as session:
+ yield session
+
+
+def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
+ user = session.get(User, user_id)
+ if not user:
+ raise HTTPException(status_code=403, detail="Not authorized")
+
+
+def generate_stream(query: str):
+ for ch in query:
+ yield ch
+ time.sleep(0.1)
+
+
+@app.get("/generate", dependencies=[Depends(get_user)])
+def generate(query: str):
+ return StreamingResponse(content=generate_stream(query))
--- a/docs_src/dependencies/tutorial014_an_py310.py
+++ b/docs_src/dependencies/tutorial014_an_py310.py
@@ -0,0 +1,39 @@
+import time
+from typing import Annotated
+
+from fastapi import Depends, FastAPI, HTTPException
+from fastapi.responses import StreamingResponse
+from sqlmodel import Field, Session, SQLModel, create_engine
+
+engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")
+
+
+class User(SQLModel, table=True):
+ id: int | None = Field(default=None, primary_key=True)
+ name: str
+
+
+app = FastAPI()
+
+
+def get_session():
+ with Session(engine) as session:
+ yield session
+
+
+def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
+ user = session.get(User, user_id)
+ if not user:
+ raise HTTPException(status_code=403, detail="Not authorized")
+ session.close()
+
+
+def generate_stream(query: str):
+ for ch in query:
+ yield ch
+ time.sleep(0.1)
+
+
+@app.get("/generate", dependencies=[Depends(get_user)])
+def generate(query: str):
+ return StreamingResponse(content=generate_stream(query))
--- a/fastapi/applications.py
+++ b/fastapi/applications.py
@@ -22,6 +22,7 @@
)
from fastapi.exceptions import RequestValidationError, WebSocketRequestValidationError
from fastapi.logger import logger
+from fastapi.middleware.asyncexitstack import AsyncExitStackMiddleware
from fastapi.openapi.docs import (
get_redoc_html,
get_swagger_ui_html,
@@ -36,10 +37,12 @@
from starlette.exceptions import HTTPException
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.middleware.errors import ServerErrorMiddleware
+from starlette.middleware.exceptions import ExceptionMiddleware
from starlette.requests import Request
from starlette.responses import HTMLResponse, JSONResponse, Response
from starlette.routing import BaseRoute
-from starlette.types import ASGIApp, Lifespan, Receive, Scope, Send
+from starlette.types import ASGIApp, ExceptionHandler, Lifespan, Receive, Scope, Send
from typing_extensions import Annotated, Doc, deprecated
AppType = TypeVar("AppType", bound="FastAPI")
@@ -990,6 +993,54 @@ class Item(BaseModel):
self.middleware_stack: Union[ASGIApp, None] = None
self.setup()
+ def build_middleware_stack(self) -> ASGIApp:
+ # Duplicate/override from Starlette to add AsyncExitStackMiddleware
+ # inside of ExceptionMiddleware, inside of custom user middlewares
+ debug = self.debug
+ error_handler = None
+ exception_handlers: dict[Any, ExceptionHandler] = {}
+
+ for key, value in self.exception_handlers.items():
+ if key in (500, Exception):
+ error_handler = value
+ else:
+ exception_handlers[key] = value
+
+ middleware = (
+ [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
+ + self.user_middleware
+ + [
+ Middleware(
+ ExceptionMiddleware, handlers=exception_handlers, debug=debug
+ ),
+ # Add FastAPI-specific AsyncExitStackMiddleware for closing files.
+ # Before this was also used for closing dependencies with yield but
+ # those now have their own AsyncExitStack, to properly support
+ # streaming responses while keeping compatibility with the previous
+ # versions (as of writing 0.117.1) that allowed doing
+ # except HTTPException inside a dependency with yield.
+ # This needs to happen after user middlewares because those create a
+ # new contextvars context copy by using a new AnyIO task group.
+ # This AsyncExitStack preserves the context for contextvars, not
+ # strictly necessary for closing files but it was one of the original
+ # intentions.
+ # If the AsyncExitStack lived outside of the custom middlewares and
+ # contextvars were set, for example in a dependency with 'yield'
+ # in that internal contextvars context, the values would not be
+ # available in the outer context of the AsyncExitStack.
+ # By placing the middleware and the AsyncExitStack here, inside all
+ # user middlewares, the same context is used.
+ # This is currently not needed, only for closing files, but used to be
+ # important when dependencies with yield were closed here.
+ Middleware(AsyncExitStackMiddleware),
+ ]
+ )
+
+ app = self.router
+ for cls, args, kwargs in reversed(middleware):
+ app = cls(app, *args, **kwargs)
+ return app
+
def openapi(self) -> Dict[str, Any]:
"""
Generate the OpenAPI schema of the application. This is called by FastAPI
--- a/fastapi/middleware/asyncexitstack.py
+++ b/fastapi/middleware/asyncexitstack.py
@@ -0,0 +1,18 @@
+from contextlib import AsyncExitStack
+
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+
+# Used mainly to close files after the request is done, dependencies are closed
+# in their own AsyncExitStack
+class AsyncExitStackMiddleware:
+ def __init__(
+ self, app: ASGIApp, context_name: str = "fastapi_middleware_astack"
+ ) -> None:
+ self.app = app
+ self.context_name = context_name
+
+ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ async with AsyncExitStack() as stack:
+ scope[self.context_name] = stack
+ await self.app(scope, receive, send)
--- a/fastapi/routing.py
+++ b/fastapi/routing.py
@@ -1,5 +1,6 @@
import dataclasses
import email.message
+import functools
import inspect
import json
import sys
@@ -8,6 +9,7 @@
from typing import (
Any,
AsyncIterator,
+ Awaitable,
Callable,
Collection,
Coroutine,
@@ -59,6 +61,8 @@
)
from pydantic import BaseModel
from starlette import routing
+from starlette._exception_handler import wrap_app_handling_exceptions
+from starlette._utils import is_async_callable
from starlette.concurrency import run_in_threadpool
from starlette.exceptions import HTTPException
from starlette.requests import Request
@@ -68,11 +72,9 @@
Match,
compile_path,
get_name,
- request_response,
- websocket_session,
)
from starlette.routing import Mount as Mount # noqa
-from starlette.types import AppType, ASGIApp, Lifespan, Scope
+from starlette.types import AppType, ASGIApp, Lifespan, Receive, Scope, Send
from starlette.websockets import WebSocket
from typing_extensions import Annotated, Doc, deprecated
@@ -82,6 +84,73 @@
from asyncio import iscoroutinefunction
+# Copy of starlette.routing.request_response modified to include the
+# dependencies' AsyncExitStack
+def request_response(
+ func: Callable[[Request], Union[Awaitable[Response], Response]],
+) -> ASGIApp:
+ """
+ Takes a function or coroutine `func(request) -> response`,
+ and returns an ASGI application.
+ """
+ f: Callable[[Request], Awaitable[Response]] = (
+ func if is_async_callable(func) else functools.partial(run_in_threadpool, func) # type:ignore
+ )
+
+ async def app(scope: Scope, receive: Receive, send: Send) -> None:
+ request = Request(scope, receive, send)
+
+ async def app(scope: Scope, receive: Receive, send: Send) -> None:
+ # Starts customization
+ response_awaited = False
+ async with AsyncExitStack() as stack:
+ scope["fastapi_inner_astack"] = stack
+ # Same as in Starlette
+ response = await f(request)
+ await response(scope, receive, send)
+ # Continues customization
+ response_awaited = True
+ if not response_awaited:
+ raise FastAPIError(
+ "Response not awaited. There's a high chance that the "
+ "application code is raising an exception and a dependency with yield "
+ "has a block with a bare except, or a block with except Exception, "
+ "and is not raising the exception again. Read more about it in the "
+ "docs: https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/#dependencies-with-yield-and-except"
+ )
+
+ # Same as in Starlette
+ await wrap_app_handling_exceptions(app, request)(scope, receive, send)
+
+ return app
+
+
+# Copy of starlette.routing.websocket_session modified to include the
+# dependencies' AsyncExitStack
+def websocket_session(
+ func: Callable[[WebSocket], Awaitable[None]],
+) -> ASGIApp:
+ """
+ Takes a coroutine `func(session)`, and returns an ASGI application.
+ """
+ # assert asyncio.iscoroutinefunction(func), "WebSocket endpoints must be async"
+
+ async def app(scope: Scope, receive: Receive, send: Send) -> None:
+ session = WebSocket(scope, receive=receive, send=send)
+
+ async def app(scope: Scope, receive: Receive, send: Send) -> None:
+ # Starts customization
+ async with AsyncExitStack() as stack:
+ scope["fastapi_inner_astack"] = stack
+ # Same as in Starlette
+ await func(session)
+
+ # Same as in Starlette
+ await wrap_app_handling_exceptions(app, session)(scope, receive, send)
+
+ return app
+
+
def _prepare_response_content(
res: Any,
*,
@@ -246,119 +315,120 @@ def get_request_handler(
async def app(request: Request) -> Response:
response: Union[Response, None] = None
- async with AsyncExitStack() as file_stack:
- try:
- body: Any = None
- if body_field:
- if is_body_form:
- body = await request.form()
- file_stack.push_async_callback(body.close)
- else:
- body_bytes = await request.body()
- if body_bytes:
- json_body: Any = Undefined
- content_type_value = request.headers.get("content-type")
- if not content_type_value:
- json_body = await request.json()
- else:
- message = email.message.Message()
- message["content-type"] = content_type_value
- if message.get_content_maintype() == "application":
- subtype = message.get_content_subtype()
- if subtype == "json" or subtype.endswith("+json"):
- json_body = await request.json()
- if json_body != Undefined:
- body = json_body
- else:
- body = body_bytes
- except json.JSONDecodeError as e:
- validation_error = RequestValidationError(
- [
- {
- "type": "json_invalid",
- "loc": ("body", e.pos),
- "msg": "JSON decode error",
- "input": {},
- "ctx": {"error": e.msg},
- }
- ],
- body=e.doc,
- )
- raise validation_error from e
- except HTTPException:
- # If a middleware raises an HTTPException, it should be raised again
- raise
- except Exception as e:
- http_error = HTTPException(
- status_code=400, detail="There was an error parsing the body"
- )
- raise http_error from e
- errors: List[Any] = []
- async with AsyncExitStack() as async_exit_stack:
- solved_result = await solve_dependencies(
- request=request,
- dependant=dependant,
- body=body,
- dependency_overrides_provider=dependency_overrides_provider,
- async_exit_stack=async_exit_stack,
- embed_body_fields=embed_body_fields,
+ file_stack = request.scope.get("fastapi_middleware_astack")
+ assert isinstance(file_stack, AsyncExitStack), (
+ "fastapi_middleware_astack not found in request scope"
+ )
+
+ # Read body and auto-close files
+ try:
+ body: Any = None
+ if body_field:
+ if is_body_form:
+ body = await request.form()
+ file_stack.push_async_callback(body.close)
+ else:
+ body_bytes = await request.body()
+ if body_bytes:
+ json_body: Any = Undefined
+ content_type_value = request.headers.get("content-type")
+ if not content_type_value:
+ json_body = await request.json()
+ else:
+ message = email.message.Message()
+ message["content-type"] = content_type_value
+ if message.get_content_maintype() == "application":
+ subtype = message.get_content_subtype()
+ if subtype == "json" or subtype.endswith("+json"):
+ json_body = await request.json()
+ if json_body != Undefined:
+ body = json_body
+ else:
+ body = body_bytes
+ except json.JSONDecodeError as e:
+ validation_error = RequestValidationError(
+ [
+ {
+ "type": "json_invalid",
+ "loc": ("body", e.pos),
+ "msg": "JSON decode error",
+ "input": {},
+ "ctx": {"error": e.msg},
+ }
+ ],
+ body=e.doc,
+ )
+ raise validation_error from e
+ except HTTPException:
+ # If a middleware raises an HTTPException, it should be raised again
+ raise
+ except Exception as e:
+ http_error = HTTPException(
+ status_code=400, detail="There was an error parsing the body"
+ )
+ raise http_error from e
+
+ # Solve dependencies and run path operation function, auto-closing dependencies
+ errors: List[Any] = []
+ async_exit_stack = request.scope.get("fastapi_inner_astack")
+ assert isinstance(async_exit_stack, AsyncExitStack), (
+ "fastapi_inner_astack not found in request scope"
+ )
+ solved_result = await solve_dependencies(
+ request=request,
+ dependant=dependant,
+ body=body,
+ dependency_overrides_provider=dependency_overrides_provider,
+ async_exit_stack=async_exit_stack,
+ embed_body_fields=embed_body_fields,
+ )
+ errors = solved_result.errors
+ if not errors:
+ raw_response = await run_endpoint_function(
+ dependant=dependant,
+ values=solved_result.values,
+ is_coroutine=is_coroutine,
+ )
+ if isinstance(raw_response, Response):
+ if raw_response.background is None:
+ raw_response.background = solved_result.background_tasks
+ response = raw_response
+ else:
+ response_args: Dict[str, Any] = {
+ "background": solved_result.background_tasks
+ }
+ # If status_code was set, use it, otherwise use the default from the
+ # response class, in the case of redirect it's 307
+ current_status_code = (
+ status_code if status_code else solved_result.response.status_code
)
- errors = solved_result.errors
- if not errors:
- raw_response = await run_endpoint_function(
- dependant=dependant,
- values=solved_result.values,
- is_coroutine=is_coroutine,
- )
- if isinstance(raw_response, Response):
- if raw_response.background is None:
- raw_response.background = solved_result.background_tasks
- response = raw_response
- else:
- response_args: Dict[str, Any] = {
- "background": solved_result.background_tasks
- }
- # If status_code was set, use it, otherwise use the default from the
- # response class, in the case of redirect it's 307
- current_status_code = (
- status_code
- if status_code
- else solved_result.response.status_code
- )
- if current_status_code is not None:
- response_args["status_code"] = current_status_code
- if solved_result.response.status_code:
- response_args["status_code"] = (
- solved_result.response.status_code
- )
- content = await serialize_response(
- field=response_field,
- response_content=raw_response,
- include=response_model_include,
- exclude=response_model_exclude,
- by_alias=response_model_by_alias,
- exclude_unset=response_model_exclude_unset,
- exclude_defaults=response_model_exclude_defaults,
- exclude_none=response_model_exclude_none,
- is_coroutine=is_coroutine,
- )
- response = actual_response_class(content, **response_args)
- if not is_body_allowed_for_status_code(response.status_code........................................E
==================================== ERRORS ====================================
_______________ ERROR at setup of test_get_no_item[tutorial008c] _______________
request = <SubRequest 'mod' for <Function test_get_no_item[tutorial008c]>>
@pytest.fixture(
name="mod",
params=[
"tutorial008c",
"tutorial008c_an",
pytest.param("tutorial008c_an_py39", marks=needs_py39),
],
)
def get_mod(request: pytest.FixtureRequest):
> mod = importlib.import_module(f"docs_src.dependencies.{request.param}")
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
tests/test_tutorial/test_dependencies/test_tutorial008c.py:20:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/Users/jp/.local/share/uv/python/cpython-3.13.15-macos-aarch64-none/lib/python3.13/importlib/__init__.py:88: in import_module
return _bootstrap._gcd_import(name[level:], package, level)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
<frozen importlib._bootstrap>:1395: in _gcd_import
???
<frozen importlib._bootstrap>:1360: in _find_and_load
???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
name = 'docs_src.dependencies.tutorial008c'
import_ = <function _gcd_import at 0x105c404a0>
> ???
E ModuleNotFoundError: No module named 'docs_src.dependencies.tutorial008c'
<frozen importlib._bootstrap>:1324: ModuleNotFoundError
=============================== warnings summary ===============================
../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-0.48.0-py3-none-any/starlette/testclient.py:51
/Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-0.48.0-py3-none-any/starlette/testclient.py:51: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.
_PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
40 passed, 1 warning, 1 error in 1.02s