From 0e4f8f65f21b5922db5c9206c313e79a79aa994f Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Mon, 2 Feb 2026 19:51:45 +0000 Subject: [PATCH] Moved fastapi-vue project to fastapi-vue-setup repository, using common version numbering (starting with v0.7.0). --- fastapi-vue/README.md | 47 +++++ fastapi-vue/fastapi_vue/__init__.py | 3 + fastapi-vue/fastapi_vue/hostutil.py | 67 +++++++ fastapi-vue/fastapi_vue/server.py | 78 ++++++++ fastapi-vue/fastapi_vue/staticfiles.py | 267 +++++++++++++++++++++++++ fastapi-vue/pyproject.toml | 23 +++ 6 files changed, 485 insertions(+) create mode 100644 fastapi-vue/README.md create mode 100644 fastapi-vue/fastapi_vue/__init__.py create mode 100644 fastapi-vue/fastapi_vue/hostutil.py create mode 100644 fastapi-vue/fastapi_vue/server.py create mode 100644 fastapi-vue/fastapi_vue/staticfiles.py create mode 100644 fastapi-vue/pyproject.toml diff --git a/fastapi-vue/README.md b/fastapi-vue/README.md new file mode 100644 index 0000000..677fd43 --- /dev/null +++ b/fastapi-vue/README.md @@ -0,0 +1,47 @@ +# fastapi-vue + +Implements Single-Page-App serving at site root with FastAPI, that the standard StaticFiles module cannot handle. This also caches and zstd compresses the files for lightning-fast operation. This is primarily meant for use with Vue frontend, but technically can host any static files in a similar manner. + +## Installation + +Script [fastapi-vue-setup](https://git.zi.fi/LeoVasanko/fastapi-vue-setup) should normally be used to convert or create a project with connection between FastAPI and Vue. The target project will depend on this package to serve its static files. + +```sh +uvx fastapi-vue-setup --help +``` + +Refer to instructions below for further configuration. + +## Usage + +```python +from contextlib import asynccontextmanager +from fastapi import FastAPI +from fastapi_vue import Frontend + +frontend = Frontend( + Path(__file__).with_name("frontend-build"), + spa=True, + cached=["/assets/"], +) + +@asynccontextmanager +async def lifespan(app: FastAPI): + await frontend.load() + yield + +app = FastAPI(lifespan=lifespan) + +# Add API routes here... + +# Final catch-all route for frontend files (keep at end of file) +frontend.route(app, "/") +``` + +## Configuration + +- `directory`: Path to static files directory +- `spa`: Enable SPA mode (serve index.html for unknown routes) +- `cached`: Path prefixes for immutable cache headers (browser won't check for changes) +- `favicon`: Path to serve at `/favicon.ico` (e.g., `"/logo.png"` will be served as `image/png`) +- `zstdlevel`: Compression level (default: 18) diff --git a/fastapi-vue/fastapi_vue/__init__.py b/fastapi-vue/fastapi_vue/__init__.py new file mode 100644 index 0000000..5409b89 --- /dev/null +++ b/fastapi-vue/fastapi_vue/__init__.py @@ -0,0 +1,3 @@ +from .staticfiles import Frontend + +__all__ = ["Frontend"] diff --git a/fastapi-vue/fastapi_vue/hostutil.py b/fastapi-vue/fastapi_vue/hostutil.py new file mode 100644 index 0000000..6b2d893 --- /dev/null +++ b/fastapi-vue/fastapi_vue/hostutil.py @@ -0,0 +1,67 @@ +import contextlib +import ipaddress +from urllib.parse import urlparse + + +def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]: + """Parse an endpoint string into uvicorn bind configurations. + + Args: + value: Endpoint string to parse + default_port: Port to use when not specified + + Returns: + List of dicts with uvicorn bind kwargs (host/port or uds). + + Supported forms: + - None or empty -> [{host: "localhost", port: default_port}] + - port (numeric) -> [{host: "localhost", port: port}] + - :port -> [{host: "0.0.0.0", port}, {host: "::", port}] (all interfaces) + - host:port -> [{host, port}] + - host -> [{host, port: default_port}] + - [ipv6]:port -> [{host: ipv6, port}] + - ipv6 (unbracketed) -> [{host: ipv6, port: default_port}] + - /path or unix:/path -> [{uds: path}] + """ + if not value: + return [{"host": "localhost", "port": default_port}] + + # Port only (numeric) -> localhost:port + if value.isdigit(): + return [{"host": "localhost", "port": int(value)}] + + # Leading colon :port -> bind all interfaces (0.0.0.0 + ::) + if value.startswith(":") and value != ":": + port_part = value[1:] + if not port_part.isdigit(): + raise SystemExit(f"Invalid port in '{value}'") + port = int(port_part) + return [{"host": "0.0.0.0", "port": port}, {"host": "::", "port": port}] # noqa: S104 + + # UNIX domain socket (unix:/path or just /path) + if value.startswith("/"): + return [{"uds": value}] + if value.startswith("unix:"): + uds_path = value[5:] or None + if uds_path is None: + raise SystemExit("unix: path must not be empty") + return [{"uds": uds_path}] + + # Unbracketed IPv6 (cannot safely contain a port) -> detect by multiple colons + if value.count(":") > 1 and not value.startswith("["): + try: + ipaddress.IPv6Address(value) + except ValueError as e: + raise SystemExit(f"Invalid IPv6 address '{value}': {e}") from e + return [{"host": value, "port": default_port}] + + # Use urllib.parse for everything else (host[:port], [ipv6][:port]) + parsed = urlparse(f"//{value}") # // prefix lets urlparse treat it as netloc + host = parsed.hostname or "localhost" + port = parsed.port or default_port + + # Validate IP literals (optional; hostname passes through) + with contextlib.suppress(ValueError): + ipaddress.ip_address(host) + + return [{"host": host, "port": port}] diff --git a/fastapi-vue/fastapi_vue/server.py b/fastapi-vue/fastapi_vue/server.py new file mode 100644 index 0000000..f57fceb --- /dev/null +++ b/fastapi-vue/fastapi_vue/server.py @@ -0,0 +1,78 @@ +import asyncio +import logging +import os +from contextlib import suppress + +import uvicorn +from uvicorn import Config, Server + +from .hostutil import parse_endpoint + +logger = logging.getLogger(__name__) + + +def run( + app: str, + *, + listen: str | list[str] | None = None, + default_port: int = 8000, + reload: bool = False, + workers: int | None = None, + **uvicorn_config, +): + """Run uvicorn server(s) for the given app. + + Args: + app: The ASGI application path (e.g., "myapp.main:app") + listen: Endpoint string(s) (see parse_endpoint for formats). + default_port: Port to use when not specified in listen args. + reload: Enable auto-reload (requires uvicorn.run, single endpoint only). + workers: Number of worker processes (requires uvicorn.run, single endpoint only). + **uvicorn_config: Additional uvicorn config options (overrides all other settings). + """ + if listen is None: + listen = [f"localhost:{default_port}"] + elif isinstance(listen, str): + listen = [listen] + endpoints: list[dict] = [] + for ep in listen: + endpoints.extend(parse_endpoint(ep, default_port)) + + conf: dict[str, object] = {"app": app, "reload": reload, "workers": workers} + proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1") + if proxy: + conf["proxy_headers"] = True + conf["forwarded_allow_ips"] = proxy + conf.update(uvicorn_config) + + with suppress(KeyboardInterrupt): + if reload or workers: + serve_multiprocess(endpoints, **conf) + else: + asyncio.run(serve(endpoints, **conf)) + + +async def serve(endpoints: list[dict], **kwargs) -> None: + """Serve the given endpoints in current process/loop. Does not spawn extra processes.""" + forbidden = {"reload", "workers"} & {k for k, v in kwargs.items() if v} + if forbidden: + logger.warning( + "Options %s have no effect in simple mode (multiple endpoints)", + ", ".join(sorted(forbidden)), + ) + await asyncio.gather(*(Server(Config(**kwargs, **ep)).serve() for ep in endpoints)) + + +def serve_multiprocess(endpoints: list[dict], **kwargs) -> None: + """Serve using uvicorn.run() for reload/workers support. Only first endpoint is used.""" + if len(endpoints) > 1: + eps = [ + ep["uds"] if "uds" in ep else f"{ep['host']}:{ep['port']}" + for ep in endpoints + ] + logger.warning( + "Current mode supports only one endpoint. Listening: %s, skipped: %s", + eps[0], + " ".join(eps[1:]), + ) + uvicorn.run(**kwargs, **endpoints[0]) diff --git a/fastapi-vue/fastapi_vue/staticfiles.py b/fastapi-vue/fastapi_vue/staticfiles.py new file mode 100644 index 0000000..86afa42 --- /dev/null +++ b/fastapi-vue/fastapi_vue/staticfiles.py @@ -0,0 +1,267 @@ +"""FastAPI static file serving with zstd compression and SPA support.""" + +import logging +import mimetypes +import os +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 + +# Dev mode: index files but don't load content, return error responses +_DEVMODE = os.getenv("FASTAPI_VUE_FRONTEND_URL") + +logger = logging.getLogger("uvicorn.error") # Use FastAPI logging style + + +class Frontend: + """Static file server with automatic zstd compression and caching. + + Features: + - Automatic zstd compression for compressible files + - ETag-based caching with configurable cache headers + - SPA (Single Page Application) support + - Favicon handling from hashed assets + - 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 should have immutable cache headers + zstdlevel: Zstd compression level (default: 18) + favicon: Path to favicon for automatic /favicon.ico handling + """ + + def __init__( + self, + directory: Path | str, + *, + index: str = "index.html", + spa: bool = False, + catch_all: bool | None = None, + cached: str | list[str] | None = None, + zstdlevel: int = 18, + favicon: str | None = None, + ) -> None: + 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 + if cached is None: + self.cached_paths = [] + elif isinstance(cached, str): + self.cached_paths = [cached] + else: + self.cached_paths = cached + self.zstdlevel = zstdlevel + self.favicon = favicon + self.devmode = bool(_DEVMODE) + 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): + """Load static files from disk with compression.""" + www: dict[str, tuple[bytes, bytes | None, dict]] = {} + if not self.base.exists(): + raise ValueError(f"Frontend folder {self.base} not found (try uv build)") + 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: + p = PurePosixPath(self.favicon) + base = str(p.with_suffix("")) + ext = p.suffix + hashed_path = next( + (path for path in www if path.startswith(base) and path.endswith(ext)), + None, + ) + if hashed_path: + www["/favicon.ico"] = www[hashed_path] + 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, *, log=True): + """Load or reload static files from disk. + + In dev mode (FASTAPI_VUE_FRONTEND_URL set), only indexes paths without loading content. + """ + if self.devmode: + # 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( + f"{self.base.name}: {len(self.www)} files in {1000 * duration:.1f} ms | " + f"zstd {len(compfiles)} files {1e-6 * raw:.2f}->{1e-6 * comp:.2f} MB ({ratio:.0f} %)" + ) + + def route(self, app: FastAPI, mount_path="/"): + """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")(self.handle) + + def _register_routes(self): + """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) + paths = self._devmode_paths if self.devmode else self.www.keys() + handler = _devmode_respond if self.devmode 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 self.devmode else partial(handler, name=p), + methods=["GET", "HEAD"], + name=f"frontend{p.replace('/', '_')}", + ) + for p in paths + ] + + def _respond(self, request: Request, name: str): + """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): + """SPA catch-all handler with directory redirects and fallback to index.""" + name = path.removesuffix(self.index) + files = self._devmode_paths if self.devmode 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 self.devmode else self._respond)(request, name) + + +def _devmode_respond(request: Request, name=""): + """Return error response directing to Vite server.""" + return JSONResponse( + status_code=409, + content={ + "detail": f"Frontend assets served by Vite in dev mode. Connect via {_DEVMODE} instead." + }, + ) diff --git a/fastapi-vue/pyproject.toml b/fastapi-vue/pyproject.toml new file mode 100644 index 0000000..b710799 --- /dev/null +++ b/fastapi-vue/pyproject.toml @@ -0,0 +1,23 @@ +[project] +name = "fastapi-vue" +dynamic = ["version"] +description = "Serves Vue assets on a FastAPI app. Use fastapi-vue-setup tool to add Vue build to your package." +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "fastapi>=0.115.0", + "zstandard>=0.23.0", + "blake3>=1.0.8", +] + +[project.urls] +Homepage = "https://git.zi.fi/LeoVasanko/fastapi-vue" +Repository = "https://github.com/LeoVasanko/fastapi-vue" + +[build-system] +requires = ["hatchling", "hatch-vcs"] +build-backend = "hatchling.build" + +[tool.hatch.version] +source = "vcs" +raw-options.root = ".."