← oracle_full

fastapi_15800

resolved RESOLVED UNSUBMITTED PASS · None tool calls · 0 s · fastapi/fastapi

Task input

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

## Pull Request

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

<!--
Please start with a GitHub Discussion.

Once a team member asks you to open a PR, create it and link the discussion here.

Obvious typo fixes can be made in a PR without starting a discussion.
-->

Discussion: <!-- Link to the GitHub Discussion -->

## Description

<!-- Write the description of your PR here -->

## AI Disclaimer

Codex with GPT 5.5, long design iterations, then many iterations tweaking and improving the implementation. Then some refactors and docs manually written as well.

<!-- If using AI, write here the prompt and model used -->

<details>
<summary>AI transcript</summary>

<!-- Paste here the entire AI transcript -->

</details>

## Checklist

- [ ] This PR is an obvious typo fix, or it links to a GitHub Discussion for the proposed code change.
- [ ] I added tests for the change.
- [ ] The new or updated tests fail on the main branch and pass on this PR.
- [ ] Coverage stays at 100%.
- [ ] The documentation explains the change if needed.

Tool calls (0)

#ToolArgumentsResult
No trace captured.

Patch

--- a/docs_src/frontend/tutorial001_py310.py
+++ b/docs_src/frontend/tutorial001_py310.py
@@ -0,0 +1,5 @@
+from fastapi import FastAPI
+
+app = FastAPI()
+
+app.frontend("/", directory="dist")
--- a/docs_src/frontend/tutorial002_py310.py
+++ b/docs_src/frontend/tutorial002_py310.py
@@ -0,0 +1,5 @@
+from fastapi import FastAPI
+
+app = FastAPI()
+
+app.frontend("/", directory="dist", fallback="index.html")
--- a/docs_src/frontend/tutorial003_py310.py
+++ b/docs_src/frontend/tutorial003_py310.py
@@ -0,0 +1,5 @@
+from fastapi import FastAPI
+
+app = FastAPI()
+
+app.frontend("/", directory="dist", fallback="404.html")
--- a/docs_src/frontend/tutorial004_py310.py
+++ b/docs_src/frontend/tutorial004_py310.py
@@ -0,0 +1,7 @@
+from fastapi import APIRouter, FastAPI
+
+app = FastAPI()
+router = APIRouter()
+
+router.frontend("/", directory="dist", fallback="index.html")
+app.include_router(router, prefix="/app")
--- a/docs_src/frontend/tutorial005_py310.py
+++ b/docs_src/frontend/tutorial005_py310.py
@@ -0,0 +1,5 @@
+from fastapi import FastAPI
+
+app = FastAPI()
+
+app.frontend("/", directory="dist", fallback=None)
--- a/docs_src/frontend/tutorial006_py310.py
+++ b/docs_src/frontend/tutorial006_py310.py
@@ -0,0 +1,5 @@
+from fastapi import FastAPI
+
+app = FastAPI()
+
+app.frontend("/", directory="dist", check_dir=False)
--- 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 @@ def add_api_route(
             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,
--- 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 typing import (
     Annotated,
     Any,
+    Literal,
     Protocol,
     TypeVar,
     cast,
@@ -80,22 +84,25 @@
 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,19 +826,33 @@ def matches(self, scope: Scope) -> tuple[Match, Scope]:
 
 _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)
 
@@ -1305,9 +1326,7 @@ def combine(
             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 @@ def effective_candidates(self) -> list["_EffectiveRouteContext | _IncludedRouter
         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 @@ def _build_effective_context(
                 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,294 @@ 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)
+
+
+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(sc

Test output

show
............................................................             [100%]
=============================== warnings summary ===============================
../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53
  /Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.
    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
60 passed, 1 warning in 1.04s