LeoVasanko
3a9c5d1674
The root logger entry patch_log_config adds now uses INFO in dev and WARNING in production (Python's default), so third-party library INFO noise stays silent in production while subloggers remain free to define their own level overrides. A user-supplied root level still wins (setdefault). staticfiles now logs via its own module logger instead of uvicorn.error: the startup stats line shows in dev (root INFO) and is hidden in production, and it no longer passes through the uvicorn-quiet filter that silently ate it. The kanta logger/handler/formatter block is removed: kanta configures its own event loggers at import time as of its logging rework, and its diagnostics follow the root logger like any other library.
297 lines
11 KiB
Python
297 lines
11 KiB
Python
"""FastAPI static file serving with zstd compression and SPA support."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import logging
|
|
import mimetypes
|
|
import time
|
|
from base64 import urlsafe_b64encode
|
|
from functools import partial
|
|
from pathlib import Path, PurePath, PurePosixPath
|
|
from wsgiref.handlers import format_date_time
|
|
|
|
from blake3 import blake3
|
|
from fastapi import FastAPI, Request, Response
|
|
from fastapi.concurrency import run_in_threadpool
|
|
from fastapi.responses import JSONResponse, RedirectResponse
|
|
from starlette.exceptions import HTTPException
|
|
from starlette.routing import Route
|
|
from zstandard import ZstdCompressor
|
|
|
|
from .environ import env
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
__all__ = ["Frontend"]
|
|
|
|
|
|
class Assets:
|
|
"""Default cached value to /assets/."""
|
|
|
|
@staticmethod
|
|
def parse(cached: str | list[str] | Assets) -> list[str]:
|
|
match cached:
|
|
case Assets():
|
|
return ["/assets/"]
|
|
case str():
|
|
return [cached]
|
|
case list():
|
|
return cached
|
|
case _:
|
|
msg = f"Invalid cached value: {cached!r}"
|
|
raise ValueError(msg)
|
|
|
|
|
|
class Frontend:
|
|
"""Static file server with automatic zstd compression and caching.
|
|
|
|
Features:
|
|
- Automatic zstd compression for compressible files
|
|
- ETag-based caching of immutable assets
|
|
- SPA (Single Page Application) support
|
|
- /favicon.ico with correct MIME type (image/png etc)
|
|
- Dev mode: indexes files but returns error directing to Vite server
|
|
|
|
Args:
|
|
directory: Path to the directory containing static files
|
|
index: Name of the index file (default: "index.html")
|
|
spa: Enable SPA mode - serve index.html for unknown routes (default: False)
|
|
cached: Path prefixes that are immutable (default: "/assets/")
|
|
favicon: Wildcard path to favicon. E.g. /assets/logo*.png matches Vite output
|
|
zstdlevel: Zstd compression level (default: 18)
|
|
|
|
"""
|
|
|
|
def __init__( # noqa: PLR0913
|
|
self,
|
|
directory: Path | str,
|
|
*,
|
|
index: str = "index.html",
|
|
spa: bool = False,
|
|
catch_all: bool | None = None,
|
|
cached: str | list[str] | Assets | None = None,
|
|
favicon: str | None = None,
|
|
zstdlevel: int = 18,
|
|
) -> None:
|
|
"""Initialize Frontend with given configuration."""
|
|
self.www: dict[str, tuple[bytes, bytes | None, dict]] = {}
|
|
self.base: Path = Path(directory)
|
|
self.index = index
|
|
self.spa = spa
|
|
self._catch_all = spa if catch_all is None else catch_all
|
|
self.cached_paths = Assets.parse(cached if cached is not None else Assets())
|
|
self.zstdlevel = zstdlevel
|
|
self.favicon = favicon
|
|
self._app: FastAPI | None = None
|
|
self._mount_path: str = ""
|
|
self._ridx: int = 0
|
|
self._routes: list[Route] = []
|
|
|
|
def _index_only(self) -> set[str]:
|
|
"""Index file paths without loading content (for dev mode)."""
|
|
paths: set[str] = set()
|
|
if not self.base.exists():
|
|
return paths
|
|
queue = [PurePath()]
|
|
while queue:
|
|
current = self.base / queue.pop(0)
|
|
for p in current.iterdir():
|
|
rel = p.relative_to(self.base)
|
|
if p.is_dir():
|
|
queue.append(rel)
|
|
continue
|
|
name = "/" + rel.as_posix()
|
|
name = name.removesuffix(self.index)
|
|
paths.add(name)
|
|
if self.favicon:
|
|
p = PurePosixPath(self.favicon)
|
|
base = str(p.with_suffix(""))
|
|
ext = p.suffix
|
|
if any(path.startswith(base) and path.endswith(ext) for path in paths):
|
|
paths.add("/favicon.ico")
|
|
return paths
|
|
|
|
def _load(self) -> dict[str, tuple[bytes, bytes | None, dict]]:
|
|
"""Load static files from disk with compression."""
|
|
www: dict[str, tuple[bytes, bytes | None, dict]] = {}
|
|
if not self.base.exists():
|
|
logger.error(
|
|
"Missing %s - no frontend (try uv build)",
|
|
self.base,
|
|
)
|
|
else:
|
|
paths = [PurePath()]
|
|
while paths:
|
|
current = self.base / paths.pop(0)
|
|
for p in current.iterdir():
|
|
rel = p.relative_to(self.base)
|
|
if p.is_dir():
|
|
paths.append(rel)
|
|
continue
|
|
# Read file
|
|
name = "/" + rel.as_posix()
|
|
mime = mimetypes.guess_type(name)[0] or "application/octet-stream"
|
|
name = name.removesuffix(self.index)
|
|
data = p.read_bytes()
|
|
etag = urlsafe_b64encode(blake3(data).digest(9)).decode()
|
|
if mime.startswith("text/"):
|
|
mime += "; charset=UTF-8"
|
|
mtime = p.stat().st_mtime
|
|
cached = any(name.startswith(prefix) for prefix in self.cached_paths)
|
|
headers = {
|
|
"etag": f'"{etag}"',
|
|
"last-modified": format_date_time(mtime),
|
|
"cache-control": ("max-age=31536000, immutable" if cached else "no-cache"),
|
|
"content-type": mime,
|
|
}
|
|
zstd = ZstdCompressor(self.zstdlevel).compress(data)
|
|
if len(zstd) >= len(data):
|
|
zstd = None
|
|
www[name] = data, zstd, headers
|
|
if self.favicon and (m := fnmatch.filter(www, self.favicon)):
|
|
data, zstd, headers = www[m[0]]
|
|
if "immutable" in headers.get("cache-control", ""):
|
|
headers = {**headers, "cache-control": "max-age=86400"}
|
|
www["/favicon.ico"] = data, zstd, headers
|
|
if not www:
|
|
msg = "Frontend files missing, check your installation.\n"
|
|
www["/"] = (
|
|
msg.encode(),
|
|
None,
|
|
{
|
|
"etag": "error",
|
|
"content-type": "text/plain",
|
|
"cache-control": "no-store",
|
|
},
|
|
)
|
|
return www
|
|
|
|
async def load(self, *, debug: bool | None = None, log: bool = True) -> None:
|
|
"""Load or reload static files from disk.
|
|
|
|
In debug mode, returns 409 instead of files (avoid accidental use of stale builds)
|
|
If debug is None, uses app.debug (app passed to frontend.route)
|
|
"""
|
|
if debug is None:
|
|
debug = getattr(self._app, "debug", False)
|
|
if debug:
|
|
# Dev mode: just index paths, no content loading
|
|
self._devmode_paths = await run_in_threadpool(self._index_only)
|
|
self._register_routes()
|
|
return
|
|
|
|
start = time.perf_counter()
|
|
self.www = await run_in_threadpool(self._load)
|
|
self._register_routes()
|
|
duration = time.perf_counter() - start
|
|
if not log:
|
|
return
|
|
compfiles = [(len(d), len(z)) for d, z, _ in self.www.values() if z]
|
|
raw = sum(v[0] for v in compfiles)
|
|
comp = sum(v[1] for v in compfiles)
|
|
ratio = comp / raw * 100 if raw else 100.0
|
|
if log and self.www:
|
|
logger.info(
|
|
"%s: %d files in %.1f ms | zstd %d files %.2f->%.2f MB (%.0f %%)",
|
|
self.base.name,
|
|
len(self.www),
|
|
1000 * duration,
|
|
len(compfiles),
|
|
1e-6 * raw,
|
|
1e-6 * comp,
|
|
ratio,
|
|
)
|
|
if self.favicon and "/favicon.ico" not in self.www:
|
|
logger.warning("Favicon not found: %s", self.favicon)
|
|
|
|
def route(self, app: FastAPI, mount_path: str = "/") -> None:
|
|
"""Register frontend routes with a FastAPI app.
|
|
|
|
In SPA/catch-all mode, this must only be called only after all other routes.
|
|
|
|
The calling position determines routing priority, although in regular mode the
|
|
routes are actually added only after load() is called.
|
|
|
|
Args:
|
|
app: FastAPI application instance
|
|
mount_path: Path where the frontend should be mounted (default: "/")
|
|
|
|
"""
|
|
self._app = app
|
|
self._mount_path = mount_path.rstrip("/")
|
|
self._ridx = len(app.routes)
|
|
|
|
if self._catch_all:
|
|
# Register catch-all immediately (works without load)
|
|
path = self._mount_path + "{path:path}"
|
|
app.api_route(path, methods=["GET", "HEAD"], name="frontend", response_model=None)(
|
|
self.handle
|
|
)
|
|
|
|
def _register_routes(self) -> None:
|
|
"""Register individual routes for each loaded file (non-catch_all mode)."""
|
|
if self._app is None or self._catch_all:
|
|
return
|
|
|
|
# Remove previously registered routes (for reload support)
|
|
for route in list(self._routes):
|
|
if route in self._app.routes:
|
|
self._app.routes.remove(route)
|
|
self._routes.clear()
|
|
|
|
# Get paths and select handler based on mode (checked once, not per request)
|
|
debug = getattr(self._app, "debug", False)
|
|
paths = self._devmode_paths if debug else self.www.keys()
|
|
handler = _devmode_respond if debug else self._respond
|
|
# Insert at the position where route() was called
|
|
self._app.routes[self._ridx : self._ridx] = self._routes = [
|
|
Route(
|
|
self._mount_path + p,
|
|
endpoint=handler if debug else partial(handler, name=p),
|
|
methods=["GET", "HEAD"],
|
|
name=f"frontend{p.replace('/', '_')}",
|
|
)
|
|
for p in paths
|
|
]
|
|
|
|
def _respond(self, request: Request, name: str) -> Response:
|
|
"""Serve a static file with ETag and compression support."""
|
|
data, zstd, headers = self.www[name]
|
|
if request.headers.get("if-none-match") == headers["etag"]:
|
|
return Response(status_code=304, headers=headers)
|
|
if zstd and "zstd" in request.headers.get("accept-encoding", ""):
|
|
return Response(
|
|
content=zstd,
|
|
headers={**headers, "content-encoding": "zstd"},
|
|
)
|
|
return Response(content=data, headers=headers)
|
|
|
|
def handle(self, request: Request, path: str) -> Response | RedirectResponse:
|
|
"""SPA catch-all handler with directory redirects and fallback to index."""
|
|
name = path.removesuffix(self.index)
|
|
debug = getattr(self._app, "debug", False)
|
|
files = self._devmode_paths if debug else self.www
|
|
|
|
if name not in files:
|
|
# Friendly redirect for directories missing trailing slash
|
|
if name and f"{name}/" in files:
|
|
return RedirectResponse(request.url.path + "/")
|
|
# SPA support: serve / for unknown paths if the browser wants HTML
|
|
if self.spa and "text/html" in request.headers.get("accept", ""):
|
|
name = "/"
|
|
# 404 for everything else
|
|
if name not in files:
|
|
raise HTTPException(status_code=404)
|
|
|
|
return (_devmode_respond if debug else self._respond)(request, name)
|
|
|
|
|
|
def _devmode_respond(_request: Request, _name: str = "") -> JSONResponse:
|
|
"""Return error response directing to Vite server."""
|
|
at = f" at {env.vite_url}" if env.vite_url else ""
|
|
return JSONResponse(
|
|
status_code=409,
|
|
content={"detail": f"[devmode] Use Vite devserver{at} instead."},
|
|
)
|