diff --git a/fastapi/applications.py b/fastapi/applications.py index c7c551e4e..56e1a3e60 100644 --- a/fastapi/applications.py +++ b/fastapi/applications.py @@ -1,6 +1,7 @@ +import os from collections.abc import Awaitable, Callable, Coroutine, Sequence from enum import Enum -from typing import Annotated, Any, TypeVar +from typing import Annotated, Any, Literal, TypeVar from annotated_doc import Doc from fastapi import routing @@ -1218,6 +1219,79 @@ class FastAPI(Starlette): generate_unique_id_function=generate_unique_id_function, ) + 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 FastAPI + + app = FastAPI() + app.frontend("/", directory="dist") + ``` + """ + self.router.frontend( + path, + directory=directory, + fallback=fallback, + check_dir=check_dir, + ) + def api_route( self, path: str, diff --git a/fastapi/routing.py b/fastapi/routing.py index 4a55fda8a..87a458967 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -1,9 +1,12 @@ 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, @@ -28,6 +31,7 @@ from enum import Enum, IntEnum from typing import ( Annotated, Any, + Literal, Protocol, TypeVar, cast, @@ -80,22 +84,25 @@ from starlette import routing 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 @@ -819,6 +826,7 @@ class APIWebSocketRoute(routing.WebSocketRoute): _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 @@ -826,12 +834,25 @@ _effective_route_context_var: ContextVar[Any | None] = ContextVar( _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) @@ -1305,9 +1326,7 @@ class _RouterIncludeContext: 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 @@ -1503,6 +1522,10 @@ class _IncludedRouter(BaseRoute): 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() @@ -1525,6 +1548,28 @@ class _IncludedRouter(BaseRoute): 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: @@ -1533,6 +1578,11 @@ class _IncludedRouter(BaseRoute): 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), @@ -1720,6 +1770,279 @@ def _iter_routes_with_context( 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) + + +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 + 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}'" + ) + + 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}'" + ) + 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 @@ -2032,6 +2355,8 @@ class APIRouter(routing.Router): 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 @@ -2093,6 +2418,150 @@ class APIRouter(routing.Router): 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 ( @@ -2113,6 +2582,60 @@ class APIRouter(routing.Router): 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,