Browse Source

Add support for `app.frontend("/", directory="dist")` and `router.frontend("/", directory="dist")`

pull/15800/head
Sebastián Ramírez 2 months ago
parent
commit
0c3d09c723
  1. 76
      fastapi/applications.py
  2. 531
      fastapi/routing.py

76
fastapi/applications.py

@ -1,6 +1,7 @@
import os
from collections.abc import Awaitable, Callable, Coroutine, Sequence from collections.abc import Awaitable, Callable, Coroutine, Sequence
from enum import Enum from enum import Enum
from typing import Annotated, Any, TypeVar from typing import Annotated, Any, Literal, TypeVar
from annotated_doc import Doc from annotated_doc import Doc
from fastapi import routing from fastapi import routing
@ -1218,6 +1219,79 @@ class FastAPI(Starlette):
generate_unique_id_function=generate_unique_id_function, 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( def api_route(
self, self,
path: str, path: str,

531
fastapi/routing.py

@ -1,9 +1,12 @@
import contextlib import contextlib
import copy import copy
import email.message import email.message
import errno
import functools import functools
import inspect import inspect
import json import json
import os
import stat
import types import types
from collections.abc import ( from collections.abc import (
AsyncIterator, AsyncIterator,
@ -28,6 +31,7 @@ from enum import Enum, IntEnum
from typing import ( from typing import (
Annotated, Annotated,
Any, Any,
Literal,
Protocol, Protocol,
TypeVar, TypeVar,
cast, cast,
@ -80,22 +84,25 @@ from starlette import routing
from starlette._exception_handler import wrap_app_handling_exceptions from starlette._exception_handler import wrap_app_handling_exceptions
from starlette._utils import get_route_path, is_async_callable from starlette._utils import get_route_path, is_async_callable
from starlette.concurrency import iterate_in_threadpool, run_in_threadpool 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.exceptions import HTTPException
from starlette.requests import Request from starlette.requests import Request
from starlette.responses import ( from starlette.responses import (
JSONResponse, JSONResponse,
PlainTextResponse, PlainTextResponse,
RedirectResponse,
Response, Response,
StreamingResponse, StreamingResponse,
) )
from starlette.routing import ( from starlette.routing import (
BaseRoute, BaseRoute,
Match, Match,
NoMatchFound,
compile_path, compile_path,
get_name, get_name,
) )
from starlette.routing import Mount as Mount # noqa 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.types import AppType, ASGIApp, Lifespan, Receive, Scope, Send
from starlette.websockets import WebSocket from starlette.websockets import WebSocket
from typing_extensions import deprecated from typing_extensions import deprecated
@ -819,6 +826,7 @@ class APIWebSocketRoute(routing.WebSocketRoute):
_FASTAPI_SCOPE_KEY = "fastapi" _FASTAPI_SCOPE_KEY = "fastapi"
_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY = "effective_route_context" _FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY = "effective_route_context"
_FASTAPI_FRONTEND_PATH_KEY = "frontend_path"
_FASTAPI_INCLUDED_ROUTER_KEY = "included_router" _FASTAPI_INCLUDED_ROUTER_KEY = "included_router"
_effective_route_context_var: ContextVar[Any | None] = ContextVar( _effective_route_context_var: ContextVar[Any | None] = ContextVar(
"fastapi_effective_route_context", default=None "fastapi_effective_route_context", default=None
@ -826,12 +834,25 @@ _effective_route_context_var: ContextVar[Any | None] = ContextVar(
_SCOPE_MISSING = object() _SCOPE_MISSING = object()
class _RouteWithPath(Protocol):
path: str
def _get_fastapi_scope(scope: Scope) -> dict[str, Any]: def _get_fastapi_scope(scope: Scope) -> dict[str, Any]:
fastapi_scope = scope.setdefault(_FASTAPI_SCOPE_KEY, {}) fastapi_scope = scope.setdefault(_FASTAPI_SCOPE_KEY, {})
assert isinstance(fastapi_scope, dict) assert isinstance(fastapi_scope, dict)
return fastapi_scope 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: def _get_scope_effective_route_context(scope: Scope) -> Any | None:
return scope.get(_FASTAPI_SCOPE_KEY, {}).get(_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY) 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, dependency_overrides_provider=self.dependency_overrides_provider,
) )
def path_for( def path_for(self, route: _RouteWithPath) -> str:
self, route: APIRoute | routing.Route | routing.WebSocketRoute | routing.Mount
) -> str:
return self.prefix + route.path return self.prefix + route.path
@ -1503,6 +1522,10 @@ class _IncludedRouter(BaseRoute):
default_factory=list default_factory=list
) )
_effective_candidates_version: int | None = None _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"]: def effective_candidates(self) -> list["_EffectiveRouteContext | _IncludedRouter"]:
routes_version = self.original_router._get_routes_version() routes_version = self.original_router._get_routes_version()
@ -1525,6 +1548,28 @@ class _IncludedRouter(BaseRoute):
self._effective_candidates_version = routes_version self._effective_candidates_version = routes_version
return self._effective_candidates 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( def _build_effective_context(
self, route: BaseRoute self, route: BaseRoute
) -> _EffectiveRouteContext | None: ) -> _EffectiveRouteContext | None:
@ -1533,6 +1578,11 @@ class _IncludedRouter(BaseRoute):
original_route=route, original_route=route,
include_context=self.include_context, 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): if isinstance(route, routing.Route):
starlette_route: BaseRoute = routing.Route( starlette_route: BaseRoute = routing.Route(
self.include_context.path_for(route), self.include_context.path_for(route),
@ -1720,6 +1770,279 @@ def _iter_routes_with_context(
yield route, None 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): class APIRouter(routing.Router):
""" """
`APIRouter` class, used to group *path operations*, for example to structure `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.generate_unique_id_function = generate_unique_id_function
self.strict_content_type = strict_content_type self.strict_content_type = strict_content_type
self._routes_version = 0 self._routes_version = 0
self._low_priority_routes: list[BaseRoute] = []
self._frontend_routes: _FrontendRouteGroup | None = None
def _mark_routes_changed(self) -> None: def _mark_routes_changed(self) -> None:
self._routes_version += 1 self._routes_version += 1
@ -2093,6 +2418,150 @@ class APIRouter(routing.Router):
super().add_websocket_route(path, endpoint, name=name) super().add_websocket_route(path, endpoint, name=name)
self._mark_routes_changed() 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: async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
included_router = _get_scope_included_router(scope) included_router = _get_scope_included_router(scope)
if ( if (
@ -2113,6 +2582,60 @@ class APIRouter(routing.Router):
return match, child_scope return match, child_scope
return Match.NONE, {} 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( def route(
self, self,
path: str, path: str,

Loading…
Cancel
Save