Moved fastapi-vue project to fastapi-vue-setup repository, using common version numbering (starting with v0.7.0).
This commit is contained in:
@@ -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)
|
||||
@@ -0,0 +1,3 @@
|
||||
from .staticfiles import Frontend
|
||||
|
||||
__all__ = ["Frontend"]
|
||||
@@ -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}]
|
||||
@@ -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])
|
||||
@@ -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."
|
||||
},
|
||||
)
|
||||
@@ -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 = ".."
|
||||
Reference in New Issue
Block a user