_is_gen_callable,
)
from fastapi.dependencies.utils import (
+ SolvedDependency,
_get_body_field,
_get_flat_body_params,
_should_embed_body_fields,
assert isinstance(path, str)
return os.path.normpath(os.path.join(*path.split("/")))
+ async def get_response_for_scope(self, scope: Scope) -> Response:
+ if not self.config_checked:
+ await self.check_config()
+ self.config_checked = True
+ return await self.get_response(self.get_path(scope), scope)
+
async def get_response(self, path: str, scope: Scope) -> Response:
if scope["method"] not in ("GET", "HEAD"):
if await self._lookup_static_resource(path) is not None:
return None
async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
- await self.app(scope, receive, send)
+ response = await self.app.get_response_for_scope(scope)
+ await response(scope, receive, send)
def url_path_for(self, name: str, /, **path_params: Any) -> URLPath:
raise NoMatchFound(name, path_params)
dependant=dependant,
dependency_overrides_provider=dependency_overrides_provider,
embed_body_fields=embed_body_fields,
- ):
- await route.handle(scope, receive, send)
+ ) as solved_result:
+ response = await route.app.get_response_for_scope(scope)
+ if response.background is None:
+ response.background = solved_result.background_tasks
+ response.headers.raw.extend(solved_result.response.headers.raw)
+ await response(scope, receive, send)
return
await route.handle(scope, receive, send)
dependant: Dependant,
dependency_overrides_provider: Any | None,
embed_body_fields: bool,
- ) -> AsyncIterator[None]:
+ ) -> AsyncIterator[SolvedDependency]:
request = Request(scope, receive, send)
previous_inner_astack = scope.get("fastapi_inner_astack", _SCOPE_MISSING)
previous_function_astack = scope.get("fastapi_function_astack", _SCOPE_MISSING)
)
if solved_result.errors:
raise RequestValidationError(solved_result.errors)
- yield
+ yield solved_result
finally:
if previous_inner_astack is _SCOPE_MISSING:
scope.pop("fastapi_inner_astack", None)
import anyio
import pytest
-from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request, WebSocket
+from fastapi import (
+ APIRouter,
+ BackgroundTasks,
+ Depends,
+ FastAPI,
+ HTTPException,
+ Request,
+ WebSocket,
+)
from fastapi.testclient import TestClient
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.responses import PlainTextResponse, Response
assert calls == ["middleware-before", "dependency", "middleware-after"]
+def test_frontend_dependency_response_headers_and_background_tasks(tmp_path: Path):
+ calls: list[str] = []
+
+ def frontend_dependency(
+ response: Response, background_tasks: BackgroundTasks
+ ) -> None:
+ response.headers["X-Frontend-Dependency"] = "applied"
+ response.set_cookie("frontend", "dependency")
+ background_tasks.add_task(calls.append, "background")
+
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app")
+ app = FastAPI(dependencies=[Depends(frontend_dependency)])
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).get("/")
+
+ assert response.status_code == 200
+ assert response.text == "app"
+ assert response.headers["X-Frontend-Dependency"] == "applied"
+ assert response.cookies["frontend"] == "dependency"
+ assert calls == ["background"]
+
+
def test_frontend_dependency_validation_errors_return_422(tmp_path: Path):
def require_token(token: str) -> None:
pass # pragma: no cover