import contextlib
import copy
import email.message
+import errno
import functools
import inspect
import json
+import os
+import stat
import types
from collections.abc import (
AsyncIterator,
from typing import (
Annotated,
Any,
+ Literal,
Protocol,
TypeVar,
cast,
from starlette._exception_handler import wrap_app_handling_exceptions
from starlette._utils import get_route_path, is_async_callable
from starlette.concurrency import iterate_in_threadpool, run_in_threadpool
-from starlette.datastructures import FormData, URLPath
+from starlette.datastructures import URL, FormData, URLPath
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.responses import (
JSONResponse,
PlainTextResponse,
+ RedirectResponse,
Response,
StreamingResponse,
)
from starlette.routing import (
BaseRoute,
Match,
+ NoMatchFound,
compile_path,
get_name,
)
from starlette.routing import Mount as Mount # noqa
+from starlette.staticfiles import StaticFiles
from starlette.types import AppType, ASGIApp, Lifespan, Receive, Scope, Send
from starlette.websockets import WebSocket
from typing_extensions import deprecated
_FASTAPI_SCOPE_KEY = "fastapi"
_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY = "effective_route_context"
+_FASTAPI_FRONTEND_PATH_KEY = "frontend_path"
_FASTAPI_INCLUDED_ROUTER_KEY = "included_router"
_effective_route_context_var: ContextVar[Any | None] = ContextVar(
"fastapi_effective_route_context", default=None
_SCOPE_MISSING = object()
+class _RouteWithPath(Protocol):
+ path: str
+
+
def _get_fastapi_scope(scope: Scope) -> dict[str, Any]:
fastapi_scope = scope.setdefault(_FASTAPI_SCOPE_KEY, {})
assert isinstance(fastapi_scope, dict)
return fastapi_scope
+def _update_scope(scope: Scope, child_scope: Scope) -> None:
+ fastapi_child_scope = child_scope.get(_FASTAPI_SCOPE_KEY)
+ for key, value in child_scope.items():
+ if key != _FASTAPI_SCOPE_KEY:
+ scope[key] = value
+ if isinstance(fastapi_child_scope, dict):
+ _get_fastapi_scope(scope).update(fastapi_child_scope)
+
+
def _get_scope_effective_route_context(scope: Scope) -> Any | None:
return scope.get(_FASTAPI_SCOPE_KEY, {}).get(_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY)
dependency_overrides_provider=self.dependency_overrides_provider,
)
- def path_for(
- self, route: APIRoute | routing.Route | routing.WebSocketRoute | routing.Mount
- ) -> str:
+ def path_for(self, route: _RouteWithPath) -> str:
return self.prefix + route.path
default_factory=list
)
_effective_candidates_version: int | None = None
+ _effective_low_priority_routes: list["_EffectiveRouteContext"] = field(
+ default_factory=list
+ )
+ _effective_low_priority_routes_version: int | None = None
def effective_candidates(self) -> list["_EffectiveRouteContext | _IncludedRouter"]:
routes_version = self.original_router._get_routes_version()
self._effective_candidates_version = routes_version
return self._effective_candidates
+ def effective_low_priority_routes(self) -> list["_EffectiveRouteContext"]:
+ routes_version = self.original_router._get_routes_version()
+ if routes_version == self._effective_low_priority_routes_version:
+ return self._effective_low_priority_routes
+ self._effective_low_priority_routes = []
+ for route in self.original_router._low_priority_routes:
+ route_context = self._build_effective_context(route)
+ if route_context is not None:
+ self._effective_low_priority_routes.append(route_context)
+ for route in self.original_router.routes:
+ if isinstance(route, _IncludedRouter):
+ child_context = self.include_context.combine(route.include_context)
+ child_branch = _IncludedRouter(
+ original_router=route.original_router,
+ include_context=child_context,
+ )
+ self._effective_low_priority_routes.extend(
+ child_branch.effective_low_priority_routes()
+ )
+ self._effective_low_priority_routes_version = routes_version
+ return self._effective_low_priority_routes
+
def _build_effective_context(
self, route: BaseRoute
) -> _EffectiveRouteContext | None:
original_route=route,
include_context=self.include_context,
)
+ if isinstance(route, _FrontendRouteGroup):
+ return _EffectiveRouteContext(
+ original_route=route,
+ starlette_route=route.with_prefix(self.include_context.prefix),
+ )
if isinstance(route, routing.Route):
starlette_route: BaseRoute = routing.Route(
self.include_context.path_for(route),
yield route, None
+def _normalize_frontend_path(path: str) -> str:
+ if not path:
+ raise AssertionError("A frontend path cannot be empty")
+ if not path.startswith("/"):
+ raise AssertionError("A frontend path must start with '/'")
+ if path != "/":
+ path = path.rstrip("/")
+ return path
+
+
+def _join_frontend_paths(prefix: str, path: str) -> str:
+ if not prefix:
+ return path
+ if path == "/":
+ return prefix
+ return prefix + path
+
+
+def _frontend_path_specificity(path: str) -> int:
+ if path == "/":
+ return 0
+ return len(path)
+
+
+def _get_resolved_absolute_path(path: str | os.PathLike[str]) -> str:
+ return os.path.realpath(os.fspath(path))
+
+
+class _FrontendStaticFiles(StaticFiles):
+ def __init__(
+ self,
+ *,
+ directory: str | os.PathLike[str],
+ fallback: Literal["auto", "index.html", "404.html"] | None,
+ check_dir: bool = True,
+ ) -> None:
+ self.fallback = fallback
+ if check_dir and not os.path.isdir(directory):
+ raise RuntimeError(
+ f"Frontend directory {directory!r} does not exist. "
+ f"Resolved absolute path: {_get_resolved_absolute_path(directory)!r}"
+ )
+ super().__init__(
+ directory=directory,
+ html=True,
+ check_dir=check_dir,
+ follow_symlink=False,
+ )
+ if check_dir and fallback in {"index.html", "404.html"}:
+ self._check_fallback_file(fallback)
+
+ def _check_fallback_file(self, fallback: str) -> None:
+ _, stat_result = self.lookup_path(fallback)
+ if stat_result is None or not stat.S_ISREG(stat_result.st_mode):
+ raise RuntimeError(
+ f"Frontend fallback file '{fallback}' does not exist in "
+ f"directory '{self.directory}'. Resolved absolute directory: "
+ f"'{self._get_resolved_directory()}'"
+ )
+
+ def _get_resolved_directory(self) -> str:
+ assert self.directory is not None
+ return _get_resolved_absolute_path(self.directory)
+
+ def get_path(self, scope: Scope) -> str:
+ path = _get_fastapi_scope(scope).get(_FASTAPI_FRONTEND_PATH_KEY, "")
+ assert isinstance(path, str)
+ return os.path.normpath(os.path.join(*path.split("/")))
+
+ async def get_response(self, path: str, scope: Scope) -> Response:
+ if scope["method"] not in ("GET", "HEAD"):
+ raise HTTPException(status_code=405)
+
+ try:
+ full_path, stat_result = await run_in_threadpool(self.lookup_path, path)
+ except PermissionError:
+ raise HTTPException(status_code=401) from None
+ except OSError as exc:
+ if exc.errno == errno.ENAMETOOLONG:
+ raise HTTPException(status_code=404) from None
+ raise exc
+ except ValueError:
+ raise HTTPException(status_code=404) from None
+
+ if stat_result and stat.S_ISREG(stat_result.st_mode):
+ return self.file_response(full_path, stat_result, scope)
+
+ if stat_result and stat.S_ISDIR(stat_result.st_mode):
+ index_path = os.path.join(path, "index.html")
+ full_path, stat_result = await run_in_threadpool(
+ self.lookup_path, index_path
+ )
+ if stat_result is not None and stat.S_ISREG(stat_result.st_mode):
+ if not scope["path"].endswith("/"):
+ url = URL(scope=scope)
+ url = url.replace(path=url.path + "/")
+ return RedirectResponse(url=url)
+ return self.file_response(full_path, stat_result, scope)
+
+ if self.fallback == "404.html" or (
+ self.fallback == "auto" and self._fallback_file_exists("404.html")
+ ):
+ return await self._fallback_response("404.html", scope, status_code=404)
+
+ if (
+ self.fallback == "index.html"
+ or (self.fallback == "auto" and self._fallback_file_exists("index.html"))
+ ) and _is_frontend_navigation_request(scope):
+ return await self._fallback_response("index.html", scope, status_code=200)
+
+ raise HTTPException(status_code=404)
+
+ def _fallback_file_exists(self, fallback: str) -> bool:
+ _, stat_result = self.lookup_path(fallback)
+ return stat_result is not None and stat.S_ISREG(stat_result.st_mode)
+
+ async def _fallback_response(
+ self, fallback: str, scope: Scope, *, status_code: int
+ ) -> Response:
+ full_path, stat_result = await run_in_threadpool(self.lookup_path, fallback)
+ if stat_result is None or not stat.S_ISREG(stat_result.st_mode):
+ raise RuntimeError(
+ f"Frontend fallback file '{fallback}' does not exist in "
+ f"directory '{self.directory}'. Resolved absolute directory: "
+ f"'{self._get_resolved_directory()}'"
+ )
+ return self.file_response(
+ full_path, stat_result, scope, status_code=status_code
+ )
+
+
+def _iter_accept_media_types(accept: str) -> Iterator[tuple[str, float]]:
+ for raw_value in accept.split(","):
+ message = email.message.Message()
+ message["content-type"] = raw_value.strip()
+ q = message.get_param("q")
+ quality = 1.0
+ if isinstance(q, str):
+ try:
+ quality = float(q)
+ except ValueError:
+ pass
+ yield (
+ f"{message.get_content_maintype()}/{message.get_content_subtype()}",
+ quality,
+ )
+
+
+def _is_frontend_navigation_request(scope: Scope) -> bool:
+ route_path = get_route_path(scope)
+ final_segment = route_path.rsplit("/", 1)[-1]
+ if os.path.splitext(final_segment)[1]:
+ return False
+ request = Request(scope)
+ wildcard_accepted = False
+ html_rejected = False
+ for media_type, quality in _iter_accept_media_types(
+ request.headers.get("accept", "")
+ ):
+ if media_type in {"text/html", "application/xhtml+xml"}:
+ if quality == 0:
+ html_rejected = True
+ else:
+ return True
+ elif media_type == "*/*" and quality != 0:
+ wildcard_accepted = True
+ return wildcard_accepted and not html_rejected
+
+
+class _FrontendRoute(BaseRoute):
+ def __init__(
+ self,
+ path: str,
+ *,
+ directory: str | os.PathLike[str],
+ fallback: Literal["auto", "index.html", "404.html"] | None = "auto",
+ check_dir: bool = True,
+ ) -> None:
+ if fallback not in {"auto", "index.html", "404.html", None}:
+ raise AssertionError(
+ "fallback must be 'auto', 'index.html', '404.html', or None"
+ )
+ self.path = _normalize_frontend_path(path)
+ self.methods = {"GET", "HEAD"}
+ self.app = _FrontendStaticFiles(
+ directory=directory, fallback=fallback, check_dir=check_dir
+ )
+
+ def with_path(self, path: str) -> "_FrontendRoute":
+ route = copy.copy(self)
+ route.path = _normalize_frontend_path(path)
+ return route
+
+ def matches(self, scope: Scope) -> tuple[Match, Scope]:
+ if scope["type"] != "http":
+ return Match.NONE, {}
+ frontend_path = self._get_frontend_path(get_route_path(scope))
+ if frontend_path is None:
+ return Match.NONE, {}
+ child_scope = {_FASTAPI_SCOPE_KEY: {_FASTAPI_FRONTEND_PATH_KEY: frontend_path}}
+ if scope["method"] not in self.methods:
+ return Match.PARTIAL, child_scope
+ return Match.FULL, child_scope
+
+ def _get_frontend_path(self, route_path: str) -> str | None:
+ if self.path == "/":
+ return route_path.lstrip("/")
+ if route_path == self.path:
+ return ""
+ prefix = self.path + "/"
+ if route_path.startswith(prefix):
+ return route_path[len(prefix) :]
+ return None
+
+ async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
+ await self.app(scope, receive, send)
+
+ def url_path_for(self, name: str, /, **path_params: Any) -> URLPath:
+ raise NoMatchFound(name, path_params)
+
+
+class _FrontendRouteGroup(BaseRoute):
+ def __init__(self) -> None:
+ self.routes: list[_FrontendRoute] = []
+
+ def add_frontend_route(
+ self,
+ path: str,
+ *,
+ directory: str | os.PathLike[str],
+ fallback: Literal["auto", "index.html", "404.html"] | None = "auto",
+ check_dir: bool = True,
+ ) -> None:
+ self.routes.append(
+ _FrontendRoute(
+ path,
+ directory=directory,
+ fallback=fallback,
+ check_dir=check_dir,
+ )
+ )
+
+ def with_prefix(self, prefix: str) -> "_FrontendRouteGroup":
+ route_group = copy.copy(self)
+ route_group.routes = [
+ route.with_path(_join_frontend_paths(prefix, route.path))
+ for route in self.routes
+ ]
+ return route_group
+
+ def matches(self, scope: Scope) -> tuple[Match, Scope]:
+ match, child_scope, _ = self._match(scope)
+ return match, child_scope
+
+ def _match(self, scope: Scope) -> tuple[Match, Scope, _FrontendRoute | None]:
+ full: tuple[Scope, _FrontendRoute] | None = None
+ partial: tuple[Scope, _FrontendRoute] | None = None
+ for route in self.routes:
+ match, child_scope = route.matches(scope)
+ if match == Match.FULL:
+ if full is None or _frontend_path_specificity(
+ route.path
+ ) > _frontend_path_specificity(full[1].path):
+ full = (child_scope, route)
+ elif match == Match.PARTIAL:
+ if partial is None or _frontend_path_specificity(
+ route.path
+ ) > _frontend_path_specificity(partial[1].path):
+ partial = (child_scope, route)
+ if full is not None:
+ child_scope, route = full
+ return Match.FULL, child_scope, route
+ if partial is not None:
+ child_scope, route = partial
+ return Match.PARTIAL, child_scope, route
+ return Match.NONE, {}, None
+
+ async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
+ match, child_scope, route = self._match(scope)
+ if match == Match.NONE or route is None:
+ raise HTTPException(status_code=404)
+ _update_scope(scope, child_scope)
+ await route.handle(scope, receive, send)
+
+ def url_path_for(self, name: str, /, **path_params: Any) -> URLPath:
+ raise NoMatchFound(name, path_params)
+
+
class APIRouter(routing.Router):
"""
`APIRouter` class, used to group *path operations*, for example to structure
self.generate_unique_id_function = generate_unique_id_function
self.strict_content_type = strict_content_type
self._routes_version = 0
+ self._low_priority_routes: list[BaseRoute] = []
+ self._frontend_routes: _FrontendRouteGroup | None = None
def _mark_routes_changed(self) -> None:
self._routes_version += 1
super().add_websocket_route(path, endpoint, name=name)
self._mark_routes_changed()
+ def frontend(
+ self,
+ path: Annotated[
+ str,
+ Doc(
+ """
+ The URL path prefix where the frontend build should be served.
+ """
+ ),
+ ],
+ *,
+ directory: Annotated[
+ str | os.PathLike[str],
+ Doc(
+ """
+ The directory containing the static frontend build output.
+ """
+ ),
+ ],
+ fallback: Annotated[
+ Literal["auto", "index.html", "404.html"] | None,
+ Doc(
+ """
+ The fallback file behavior for missing frontend paths.
+ """
+ ),
+ ] = "auto",
+ check_dir: Annotated[
+ bool,
+ Doc(
+ """
+ Check that the frontend directory exists when the app is created.
+ """
+ ),
+ ] = True,
+ ) -> None:
+ """
+ Serve a static frontend build as low-priority routes.
+
+ Use this for frontend tools that build static files into a directory,
+ such as `dist`. **FastAPI** path operations are checked first, and
+ the frontend files are checked only if no normal route matched.
+
+ A typical project could look like this:
+
+ ```text
+ .
+ ├── pyproject.toml
+ ├── app
+ │ ├── __init__.py
+ │ └── main.py
+ └── dist
+ ├── index.html
+ └── assets
+ └── app.js
+ ```
+
+ Then in `app/main.py`:
+
+ ```python
+ from fastapi import APIRouter, FastAPI
+
+ app = FastAPI()
+ router = APIRouter()
+ router.frontend("/", directory="dist")
+ app.include_router(router)
+ ```
+ """
+ normalized_path = _normalize_frontend_path(path)
+ if self._frontend_routes is None:
+ self._frontend_routes = _FrontendRouteGroup()
+ self._low_priority_routes.append(self._frontend_routes)
+ self._frontend_routes.add_frontend_route(
+ _join_frontend_paths(self.prefix, normalized_path),
+ directory=directory,
+ fallback=fallback,
+ check_dir=check_dir,
+ )
+ self._mark_routes_changed()
+
+ async def app(self, scope: Scope, receive: Receive, send: Send) -> None:
+ assert scope["type"] in ("http", "websocket", "lifespan")
+
+ if "router" not in scope:
+ scope["router"] = self
+
+ if scope["type"] == "lifespan":
+ await self.lifespan(scope, receive, send)
+ return
+
+ partial: tuple[BaseRoute, Scope] | None = None
+ for route in self.routes:
+ match, child_scope = route.matches(scope)
+ if match == Match.FULL:
+ scope.update(child_scope)
+ await route.handle(scope, receive, send)
+ return
+ if match == Match.PARTIAL and partial is None:
+ partial = (route, child_scope)
+
+ if partial is not None:
+ route, child_scope = partial
+ scope.update(child_scope)
+ await route.handle(scope, receive, send)
+ return
+
+ route_path = get_route_path(scope)
+ if scope["type"] == "http" and self.redirect_slashes and route_path != "/":
+ redirect_scope = dict(scope)
+ if route_path.endswith("/"):
+ redirect_scope["path"] = redirect_scope["path"].rstrip("/")
+ else:
+ redirect_scope["path"] = redirect_scope["path"] + "/"
+
+ for route in self.routes:
+ match, _ = route.matches(redirect_scope)
+ if match != Match.NONE:
+ redirect_url = URL(scope=redirect_scope)
+ response = RedirectResponse(url=str(redirect_url))
+ await response(scope, receive, send)
+ return
+
+ (
+ low_priority_match,
+ low_priority_scope,
+ low_priority_route,
+ low_priority_context,
+ ) = self._match_low_priority(scope)
+ if low_priority_match != Match.NONE and low_priority_route is not None:
+ _update_scope(scope, low_priority_scope)
+ if low_priority_context is not None:
+ _get_fastapi_scope(scope)[_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY] = (
+ low_priority_context
+ )
+ original_route = low_priority_context.original_route
+ if isinstance(original_route, APIRoute):
+ scope["route"] = original_route
+ await original_route.handle(scope, receive, send)
+ return
+ await low_priority_route.handle(scope, receive, send)
+ return
+
+ await self.default(scope, receive, send)
+
async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
included_router = _get_scope_included_router(scope)
if (
return match, child_scope
return Match.NONE, {}
+ def _iter_low_priority_routes(
+ self,
+ ) -> Iterator[BaseRoute | _EffectiveRouteContext]:
+ yield from self._low_priority_routes
+ for route in self.routes:
+ if isinstance(route, _IncludedRouter):
+ yield from route.effective_low_priority_routes()
+
+ def _match_low_priority(
+ self, scope: Scope
+ ) -> tuple[Match, Scope, BaseRoute | None, _EffectiveRouteContext | None]:
+ full: tuple[Scope, BaseRoute, _EffectiveRouteContext | None] | None = None
+ partial: tuple[Scope, BaseRoute, _EffectiveRouteContext | None] | None = None
+ for candidate in self._iter_low_priority_routes():
+ route: BaseRoute
+ if isinstance(candidate, _EffectiveRouteContext):
+ route_context: _EffectiveRouteContext | None = candidate
+ original_route = candidate.original_route
+ if isinstance(original_route, APIRoute):
+ fastapi_scope = _get_fastapi_scope(scope)
+ previous_context = fastapi_scope.get(
+ _FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY, _SCOPE_MISSING
+ )
+ fastapi_scope[_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY] = route_context
+ try:
+ match, child_scope = original_route.matches(scope)
+ finally:
+ _restore_fastapi_scope_key(
+ scope,
+ _FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY,
+ previous_context,
+ )
+ route = original_route
+ else:
+ match, child_scope = candidate.matches(scope)
+ route = candidate.starlette_route or original_route
+ else:
+ route_context = None
+ match, child_scope = candidate.matches(scope)
+ route = candidate
+ if match == Match.FULL:
+ if full is None:
+ full = (child_scope, route, route_context)
+ elif match == Match.PARTIAL:
+ if partial is None:
+ partial = (child_scope, route, route_context)
+ if full is not None:
+ child_scope, route, route_context = full
+ return Match.FULL, child_scope, route, route_context
+ if partial is not None:
+ child_scope, route, route_context = partial
+ return Match.PARTIAL, child_scope, route, route_context
+ return Match.NONE, {}, None, None
+
def route(
self,
path: str,
--- /dev/null
+import errno
+import os
+import runpy
+from pathlib import Path
+
+import anyio
+import pytest
+from fastapi import APIRouter, FastAPI, HTTPException, Request, WebSocket
+from fastapi.testclient import TestClient
+from starlette.exceptions import HTTPException as StarletteHTTPException
+from starlette.responses import PlainTextResponse, Response
+from starlette.routing import BaseRoute, Match, NoMatchFound, Route
+
+
+def write_file(path: Path, content: str) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(content)
+
+
+def test_frontend_exact_prefix_path_serves_index(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app")
+ app = FastAPI()
+ app.frontend("/app", directory=dist)
+
+ response = TestClient(app).get("/app")
+
+ assert response.status_code == 200
+ assert response.text == "app"
+
+
+def test_apirouter_frontend_with_router_prefix_and_frontend_subpath(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "asset.txt", "asset")
+ router = APIRouter(prefix="/internal")
+ router.frontend("/ui", directory=dist)
+ app = FastAPI()
+ app.include_router(router, prefix="/prefix")
+
+ response = TestClient(app).get("/prefix/internal/ui/asset.txt")
+
+ assert response.status_code == 200
+ assert response.text == "asset"
+
+
+def test_frontend_fallback_rejects_invalid_fallback(tmp_path: Path):
+ dist = tmp_path / "dist"
+ dist.mkdir()
+ app = FastAPI()
+
+ with pytest.raises(AssertionError, match="fallback"):
+ app.frontend("/", directory=dist, fallback="invalid") # type: ignore[arg-type] # ty: ignore[invalid-argument-type]
+
+
+def test_index_fallback_ignores_invalid_q_value(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app shell")
+ app = FastAPI()
+ app.frontend("/", directory=dist, fallback="index.html")
+
+ response = TestClient(app).get(
+ "/dashboard/settings", headers={"accept": "text/html; q=wat"}
+ )
+
+ assert response.status_code == 200
+ assert response.text == "app shell"
+
+
+def test_frontend_static_files_lookup_errors(monkeypatch, tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app")
+ app = FastAPI()
+ app.frontend("/", directory=dist)
+ frontend_routes = app.router._frontend_routes
+ assert frontend_routes is not None
+ static_files = frontend_routes.routes[0].app
+
+ def raise_permission_error(path: str):
+ raise PermissionError
+
+ monkeypatch.setattr(static_files, "lookup_path", raise_permission_error)
+ response = TestClient(app).get("/asset.txt")
+ assert response.status_code == 401
+
+ def raise_value_error(path: str):
+ raise ValueError
+
+ monkeypatch.setattr(static_files, "lookup_path", raise_value_error)
+ response = TestClient(app).get("/asset.txt")
+ assert response.status_code == 404
+
+ def raise_name_too_long(path: str):
+ raise OSError(errno.ENAMETOOLONG, "name too long")
+
+ monkeypatch.setattr(static_files, "lookup_path", raise_name_too_long)
+ response = TestClient(app).get("/asset.txt")
+ assert response.status_code == 404
+
+ def raise_os_error(path: str):
+ raise OSError(5, "other")
+
+ monkeypatch.setattr(static_files, "lookup_path", raise_os_error)
+ with pytest.raises(OSError):
+ TestClient(app).get("/asset.txt")
+
+
+def test_frontend_route_group_helpers(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app")
+ app = FastAPI()
+ app.frontend("/", directory=dist)
+ route_group = app.router._frontend_routes
+ assert route_group is not None
+
+ match, child_scope = route_group.matches({"type": "websocket", "path": "/"})
+ assert match == Match.NONE
+ assert child_scope == {}
+
+ with pytest.raises(StarletteHTTPException) as exc_info:
+ anyio.run(
+ route_group.with_prefix("/app").handle,
+ {"type": "http", "path": "/missing", "method": "GET"},
+ None,
+ None,
+ )
+ assert exc_info.value.status_code == 404
+
+ with pytest.raises(NoMatchFound):
+ route_group.url_path_for("frontend")
+ with pytest.raises(NoMatchFound):
+ route_group.routes[0].url_path_for("frontend")
+
+
+def test_included_low_priority_routes_cache_is_reused():
+ async def low_priority_endpoint(request: Request):
+ return PlainTextResponse("low")
+
+ router = APIRouter()
+ router._low_priority_routes.append(Route("/low", low_priority_endpoint))
+ router._mark_routes_changed()
+ app = FastAPI()
+ app.include_router(router, prefix="/prefix")
+ included_router = next(
+ route
+ for route in app.router.routes
+ if hasattr(route, "effective_low_priority_routes")
+ )
+
+ first = included_router.effective_low_priority_routes() # ty: ignore[call-non-callable]
+ second = included_router.effective_low_priority_routes() # ty: ignore[call-non-callable]
+ response = TestClient(app).get("/prefix/low")
+
+ assert first is second
+ assert response.status_code == 200
+ assert response.text == "low"
+
+
+def test_low_priority_api_route_handles_with_context():
+ app = FastAPI()
+
+ async def endpoint(request: Request) -> Response:
+ return PlainTextResponse(request.scope["path_params"]["item_id"])
+
+ route = app.router.route_class("/low/{item_id}", endpoint=endpoint, methods=["GET"])
+ app.router._low_priority_routes.append(route)
+ app.router._mark_routes_changed()
+
+ response = TestClient(app).get("/low/abc")
+
+ assert response.status_code == 200
+ assert response.text == "abc"
+
+
+def test_included_low_priority_api_route_handles_with_context():
+ router = APIRouter()
+
+ async def endpoint(request: Request) -> Response:
+ return PlainTextResponse(request.scope["path_params"]["item_id"])
+
+ route = router.route_class("/low/{item_id}", endpoint=endpoint, methods=["GET"])
+ router._low_priority_routes.append(route)
+ router._mark_routes_changed()
+ app = FastAPI()
+ app.include_router(router, prefix="/prefix")
+
+ response = TestClient(app).get("/prefix/low/abc")
+
+ assert response.status_code == 200
+ assert response.text == "abc"
+
+
+def test_normal_route_partial_match_returns_before_frontend(tmp_path: Path):
+ class PartialRoute(BaseRoute):
+ def matches(self, scope):
+ return Match.PARTIAL, {}
+
+ async def handle(self, scope, receive, send):
+ response = PlainTextResponse("partial", status_code=405)
+ await response(scope, receive, send)
+
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "frontend")
+ app = FastAPI()
+ app.router.routes.append(PartialRoute())
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).get("/anything")
+
+ assert response.status_code == 405
+ assert response.text == "partial"
+
+
+def test_normal_route_partial_match_wins_before_frontend(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "api", "frontend")
+ app = FastAPI()
+
+ @app.get("/api")
+ def read_api():
+ return {"source": "api"}
+
+ app.frontend("/", directory=dist)
+
+ client = TestClient(app)
+
+ response = client.get("/api")
+ assert response.status_code == 200
+ assert response.json() == {"source": "api"}
+
+ response = client.post("/api")
+ assert response.status_code == 405
+
+
+def test_basic_file_serving(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "assets" / "app.js", "console.log('ok')")
+ app = FastAPI()
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).get("/assets/app.js")
+
+ assert response.status_code == 200
+ assert response.text == "console.log('ok')"
+ assert "etag" in response.headers
+ assert "last-modified" in response.headers
+
+
+def test_existing_api_route_wins_over_frontend(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "api" / "users", "frontend")
+ app = FastAPI()
+
+ @app.get("/api/users")
+ def read_users():
+ return {"source": "api"}
+
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).get("/api/users")
+
+ assert response.status_code == 200
+ assert response.json() == {"source": "api"}
+
+
+def test_api_route_404_is_not_replaced_by_frontend_fallback(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "frontend")
+ app = FastAPI()
+
+ @app.get("/api/users")
+ def read_users():
+ raise HTTPException(status_code=404, detail="api missing")
+
+ app.frontend("/", directory=dist, fallback="index.html")
+
+ response = TestClient(app).get("/api/users", headers={"accept": "text/html"})
+
+ assert response.status_code == 404
+ assert response.json() == {"detail": "api missing"}
+
+
+def test_index_fallback_for_navigation_request(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app shell")
+ app = FastAPI()
+ app.frontend("/", directory=dist, fallback="index.html")
+
+ response = TestClient(app).get(
+ "/dashboard/settings", headers={"accept": "text/html"}
+ )
+
+ assert response.status_code == 200
+ assert response.text == "app shell"
+
+
+def test_index_fallback_parses_accept_parameters(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app shell")
+ app = FastAPI()
+ app.frontend("/", directory=dist, fallback="index.html")
+
+ response = TestClient(app).get(
+ "/dashboard/settings", headers={"accept": "text/html; q=0.8"}
+ )
+
+ assert response.status_code == 200
+ assert response.text == "app shell"
+
+
+def test_index_fallback_ignores_q_zero_accept(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app shell")
+ app = FastAPI()
+ app.frontend("/", directory=dist, fallback="index.html")
+
+ response = TestClient(app).get(
+ "/dashboard/settings", headers={"accept": "text/html; q=0.0"}
+ )
+
+ assert response.status_code == 404
+
+
+def test_index_fallback_respects_explicit_html_rejection_with_wildcard(
+ tmp_path: Path,
+):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app shell")
+ app = FastAPI()
+ app.frontend("/", directory=dist, fallback="index.html")
+
+ response = TestClient(app).get(
+ "/dashboard/settings",
+ headers={"accept": "text/html; q=0, */*; q=1"},
+ )
+
+ assert response.status_code == 404
+
+
+def test_index_fallback_respects_explicit_xhtml_rejection_with_wildcard(
+ tmp_path: Path,
+):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app shell")
+ app = FastAPI()
+ app.frontend("/", directory=dist, fallback="index.html")
+
+ response = TestClient(app).get(
+ "/dashboard/settings",
+ headers={"accept": "application/xhtml+xml; q=0, */*; q=1"},
+ )
+
+ assert response.status_code == 404
+
+
+@pytest.mark.parametrize(
+ ("path", "accept"),
+ [
+ ("/assets/missing.js", "*/*"),
+ ("/assets/missing.css", "text/css"),
+ ("/assets/missing.png", "image/png"),
+ ("/api/missing", "application/json"),
+ ("/users/jane.doe", "text/html"),
+ ],
+)
+def test_index_fallback_does_not_handle_asset_like_or_non_html_requests(
+ tmp_path: Path, path: str, accept: str
+):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app shell")
+ app = FastAPI()
+ app.frontend("/", directory=dist, fallback="index.html")
+
+ response = TestClient(app).get(path, headers={"accept": accept})
+
+ assert response.status_code == 404
+ assert response.text != "app shell"
+
+
+def test_404_fallback_handles_missing_assets(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "404.html", "missing")
+ app = FastAPI()
+ app.frontend("/", directory=dist, fallback="404.html")
+
+ response = TestClient(app).get("/assets/missing.js")
+
+ assert response.status_code == 404
+ assert response.text == "missing"
+
+
+def test_auto_fallback_prefers_404_over_index(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app shell")
+ write_file(dist / "404.html", "missing")
+ app = FastAPI()
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).get("/dashboard", headers={"accept": "text/html"})
+
+ assert response.status_code == 404
+ assert response.text == "missing"
+
+
+def test_auto_fallback_uses_index_when_404_is_missing(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app shell")
+ app = FastAPI()
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).get("/dashboard", headers={"accept": "text/html"})
+
+ assert response.status_code == 200
+ assert response.text == "app shell"
+
+
+def test_auto_fallback_returns_normal_404_without_fallback_files(tmp_path: Path):
+ dist = tmp_path / "dist"
+ dist.mkdir()
+ app = FastAPI()
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).get("/dashboard", headers={"accept": "text/html"})
+
+ assert response.status_code == 404
+ assert response.json() == {"detail": "Not Found"}
+
+
+def test_no_fallback_returns_normal_404(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app shell")
+ app = FastAPI()
+ app.frontend("/", directory=dist, fallback=None)
+
+ response = TestClient(app).get("/dashboard", headers={"accept": "text/html"})
+
+ assert response.status_code == 404
+ assert response.json() == {"detail": "Not Found"}
+
+
+def test_directory_index_and_redirect(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "about" / "index.html", "about")
+ app = FastAPI()
+ app.frontend("/", directory=dist)
+ client = TestClient(app)
+
+ redirect = client.get("/about", follow_redirects=False)
+ response = client.get("/about/")
+
+ assert redirect.status_code == 307
+ assert redirect.headers["location"] == "http://testserver/about/"
+ assert response.status_code == 200
+ assert response.text == "about"
+
+
+def test_path_validation_and_trailing_slash_normalization(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "asset.txt", "ok")
+ app = FastAPI()
+
+ with pytest.raises(AssertionError):
+ app.frontend("", directory=dist)
+ with pytest.raises(AssertionError):
+ app.frontend("app", directory=dist)
+
+ app.frontend("/app/", directory=dist)
+ response = TestClient(app).get("/app/asset.txt")
+
+ assert response.status_code == 200
+ assert response.text == "ok"
+
+
+def test_frontend_path_matching_uses_segment_boundaries(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app")
+ app = FastAPI()
+ app.frontend("/app", directory=dist, fallback="index.html")
+
+ response = TestClient(app).get("/application", headers={"accept": "text/html"})
+
+ assert response.status_code == 404
+
+
+def test_multiple_frontends_use_longest_matching_prefix(tmp_path: Path):
+ site = tmp_path / "site"
+ admin = tmp_path / "admin"
+ write_file(site / "index.html", "site")
+ write_file(admin / "index.html", "admin")
+ app = FastAPI()
+ app.frontend("/", directory=site, fallback="index.html")
+ app.frontend("/admin", directory=admin, fallback="index.html")
+
+ response = TestClient(app).get("/admin/settings", headers={"accept": "text/html"})
+
+ assert response.status_code == 200
+ assert response.text == "admin"
+
+
+def test_apirouter_frontend_uses_include_prefix(tmp_path: Path):
+ dist = tmp_path / "admin"
+ write_file(dist / "index.html", "admin")
+ router = APIRouter()
+ router.frontend("/", directory=dist, fallback="index.html")
+ app = FastAPI()
+ app.include_router(router, prefix="/admin")
+
+ response = TestClient(app).get("/admin/settings", headers={"accept": "text/html"})
+
+ assert response.status_code == 200
+ assert response.text == "admin"
+
+
+def test_global_priority_across_included_routers(tmp_path: Path):
+ dist = tmp_path / "site"
+ write_file(dist / "index.html", "site")
+ site_router = APIRouter()
+ site_router.frontend("/", directory=dist, fallback="index.html")
+ api_router = APIRouter()
+
+ @api_router.get("/api/users")
+ def read_users():
+ return {"source": "api"}
+
+ app = FastAPI()
+ app.include_router(site_router)
+ app.include_router(api_router)
+
+ response = TestClient(app).get("/api/users", headers={"accept": "text/html"})
+
+ assert response.status_code == 200
+ assert response.json() == {"source": "api"}
+
+
+def test_nested_apirouter_frontend_uses_all_include_prefixes(tmp_path: Path):
+ dist = tmp_path / "admin"
+ write_file(dist / "index.html", "admin")
+ child_router = APIRouter()
+ child_router.frontend("/", directory=dist, fallback="index.html")
+ parent_router = APIRouter()
+ parent_router.include_router(child_router, prefix="/child")
+ app = FastAPI()
+ app.include_router(parent_router, prefix="/parent")
+
+ response = TestClient(app).get(
+ "/parent/child/settings", headers={"accept": "text/html"}
+ )
+
+ assert response.status_code == 200
+ assert response.text == "admin"
+
+
+def test_low_priority_cache_updates_after_route_added_to_included_router(
+ tmp_path: Path,
+):
+ dist = tmp_path / "site"
+ write_file(dist / "index.html", "site")
+ router = APIRouter()
+ router.frontend("/", directory=dist, fallback="index.html")
+ app = FastAPI()
+ app.include_router(router, prefix="/app")
+ client = TestClient(app)
+
+ frontend_response = client.get("/app/dashboard", headers={"accept": "text/html"})
+
+ @router.get("/dashboard")
+ def read_dashboard():
+ return {"source": "api"}
+
+ api_response = client.get("/app/dashboard", headers={"accept": "text/html"})
+
+ assert frontend_response.status_code == 200
+ assert frontend_response.text == "site"
+ assert api_response.status_code == 200
+ assert api_response.json() == {"source": "api"}
+
+
+def test_normal_route_slash_redirect_wins_before_frontend_redirect(tmp_path: Path):
+ dist = tmp_path / "site"
+ write_file(dist / "api" / "index.html", "frontend")
+ app = FastAPI()
+
+ @app.get("/api/")
+ def read_api():
+ return {"source": "api"}
+
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).get("/api", follow_redirects=False)
+
+ assert response.status_code == 307
+ assert response.headers["location"] == "http://testserver/api/"
+
+ followed = TestClient(app).get("/api/")
+ assert followed.status_code == 200
+ assert followed.json() == {"source": "api"}
+
+
+def test_frontend_respects_root_path(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "assets" / "app.js", "console.log('ok')")
+ app = FastAPI()
+ app.frontend("/app", directory=dist)
+
+ response = TestClient(app, root_path="/proxy").get("/app/assets/app.js")
+
+ assert response.status_code == 200
+ assert response.text == "console.log('ok')"
+
+
+def test_websocket_route_wins_over_frontend(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "ws", "frontend")
+ app = FastAPI()
+
+ @app.websocket("/ws")
+ async def websocket_endpoint(websocket: WebSocket):
+ await websocket.accept()
+ await websocket.send_text("websocket")
+ await websocket.close()
+
+ app.frontend("/", directory=dist)
+
+ with TestClient(app).websocket_connect("/ws") as websocket:
+ data = websocket.receive_text()
+
+ assert data == "websocket"
+
+
+def test_head_requests_work(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "asset.txt", "ok")
+ app = FastAPI()
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).head("/asset.txt")
+
+ assert response.status_code == 200
+ assert response.text == ""
+ assert response.headers["content-length"] == "2"
+
+
+def test_unsupported_methods_return_405(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "asset.txt", "ok")
+ app = FastAPI()
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).post("/asset.txt")
+
+ assert response.status_code == 405
+
+
+@pytest.mark.parametrize(
+ "path",
+ [
+ "/../secret.txt",
+ "/%2e%2e/secret.txt",
+ "/..%2fsecret.txt",
+ "/%5c..%5csecret.txt",
+ "/..%5csecret.txt",
+ ],
+)
+def test_path_traversal_cannot_escape_directory(tmp_path: Path, path: str):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app")
+ write_file(tmp_path / "secret.txt", "secret")
+ app = FastAPI()
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).get(path)
+
+ assert response.status_code == 404
+ assert response.text != "secret"
+
+
+def test_symlink_outside_directory_is_not_served(tmp_path: Path):
+ dist = tmp_path / "dist"
+ dist.mkdir()
+ outside = tmp_path / "secret.txt"
+ outside.write_text("secret")
+ link = dist / "secret.txt"
+ try:
+ os.symlink(outside, link)
+ except (OSError, NotImplementedError): # pragma: no cover
+ pytest.skip("symlinks are not supported")
+ app = FastAPI()
+ app.frontend("/", directory=dist)
+
+ response = TestClient(app).get("/secret.txt")
+
+ assert response.status_code == 404
+ assert response.text != "secret"
+
+
+def test_check_dir_true_fails_early_for_missing_directory(monkeypatch, tmp_path: Path):
+ app = FastAPI()
+ monkeypatch.chdir(tmp_path)
+
+ with pytest.raises(RuntimeError, match="does not exist") as exc_info:
+ app.frontend("/", directory="missing")
+
+ message = str(exc_info.value)
+ assert "'missing'" in message
+ assert str(tmp_path / "missing") in message
+
+
+def test_check_dir_false_allows_missing_directory_and_fails_on_request(tmp_path: Path):
+ app = FastAPI()
+ app.frontend("/", directory=tmp_path / "missing", check_dir=False)
+
+ with pytest.raises(RuntimeError, match="does not exist"):
+ TestClient(app).get("/asset.txt")
+
+
+def test_explicit_fallback_files_fail_clearly_when_missing(monkeypatch, tmp_path: Path):
+ dist = tmp_path / "dist"
+ dist.mkdir()
+ monkeypatch.chdir(tmp_path)
+ app = FastAPI()
+
+ with pytest.raises(RuntimeError, match="index.html") as exc_info:
+ app.frontend("/", directory="dist", fallback="index.html")
+
+ message = str(exc_info.value)
+ assert "directory 'dist'" in message
+ assert str(dist) in message
+
+ app = FastAPI()
+ app.frontend("/", directory="dist", fallback="404.html", check_dir=False)
+
+ with pytest.raises(RuntimeError, match="404.html") as exc_info:
+ TestClient(app).get("/missing.js")
+
+ message = str(exc_info.value)
+ assert "directory 'dist'" in message
+ assert str(dist) in message
+
+
+def test_frontend_routes_are_not_in_openapi(tmp_path: Path):
+ dist = tmp_path / "dist"
+ write_file(dist / "index.html", "app")
+ app = FastAPI()
+
+ @app.get("/api")
+ def read_api():
+ return {"ok": True}
+
+ app.frontend("/", directory=dist, fallback="index.html")
+
+ schema = TestClient(app).get("/openapi.json").json()
+
+ assert set(schema["paths"]) == {"/api"}
+
+ response = TestClient(app).get("/api")
+ assert response.status_code == 200
+ assert response.json() == {"ok": True}
+
+
+@pytest.mark.parametrize(
+ ("example", "files", "path", "status_code", "body"),
+ [
+ (
+ "tutorial001_py310.py",
+ {"asset.txt": "asset"},
+ "/asset.txt",
+ 200,
+ "asset",
+ ),
+ (
+ "tutorial002_py310.py",
+ {"index.html": "index"},
+ "/dashboard",
+ 200,
+ "index",
+ ),
+ (
+ "tutorial003_py310.py",
+ {"404.html": "missing"},
+ "/missing",
+ 404,
+ "missing",
+ ),
+ (
+ "tutorial004_py310.py",
+ {"index.html": "index"},
+ "/app/dashboard",
+ 200,
+ "index",
+ ),
+ (
+ "tutorial005_py310.py",
+ {"index.html": "index"},
+ "/dashboard",
+ 404,
+ '{"detail":"Not Found"}',
+ ),
+ (
+ "tutorial006_py310.py",
+ {"asset.txt": "asset"},
+ "/asset.txt",
+ 200,
+ "asset",
+ ),
+ ],
+)
+def test_docs_frontend_examples(
+ tmp_path: Path,
+ monkeypatch,
+ example: str,
+ files: dict[str, str],
+ path: str,
+ status_code: int,
+ body: str,
+):
+ dist = tmp_path / "dist"
+ for file, content in files.items():
+ write_file(dist / file, content)
+ monkeypatch.chdir(tmp_path)
+
+ namespace = runpy.run_path(
+ str(Path(__file__).parents[1] / "docs_src" / "frontend" / example)
+ )
+
+ app = namespace["app"]
+ assert isinstance(app, FastAPI)
+ response = TestClient(app).get(path, headers={"accept": "text/html"})
+ assert response.status_code == status_code
+ assert response.text == body
+
+
+def test_low_priority_routes_can_store_non_frontend_routes():
+ async def low_priority_endpoint(request):
+ return PlainTextResponse("low")
+
+ app = FastAPI()
+ app.router._low_priority_routes.append(Route("/low", low_priority_endpoint))
+ app.router._mark_routes_changed()
+
+ response = TestClient(app).get("/low")
+
+ assert response.status_code == 200
+ assert response.text == "low"
+
+
+def test_included_low_priority_routes_can_store_non_frontend_routes():
+ async def low_priority_endpoint(request):
+ return PlainTextResponse("low")
+
+ router = APIRouter()
+ router._low_priority_routes.append(Route("/low", low_priority_endpoint))
+ router._mark_routes_changed()
+ app = FastAPI()
+ app.include_router(router, prefix="/prefix")
+
+ response = TestClient(app).get("/prefix/low")
+
+ assert response.status_code == 200
+ assert response.text == "low"
projects / thirdparty / fastapi / fastapi.git / commitdiff
