LeoVasanko/fastapi-vue-setup
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cleanup for ALL ruff checks, and re-ruff to target project settings when installing templates, avoiding formatting errors after patching.

Browse Source
This commit is contained in:
LeoVasanko
2026-03-08 15:25:52 +00:00
parent 1aaf040db7
commit 682d70cb23
14 changed files with 405 additions and 317 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fastapi-vue/fastapi_vue/__init__.py
+2
View File
@@ -1,3 +1,5 @@
"""FastAPI Vue integration - serve Vue frontend from FastAPI."""
from .staticfiles import Frontend
__all__ = ["Frontend"]
fastapi-vue/fastapi_vue/hostutil.py
+68 -33
View File
@@ -1,8 +1,60 @@
"""Parse endpoint strings for uvicorn server configuration."""
import contextlib
import ipaddress
from urllib.parse import urlparse
def _parse_all_interfaces(value: str) -> list[dict] | None:
"""Parse ':port' format to bind all interfaces."""
if not (value.startswith(":") and value != ":"):
return None
port_part = value[1:]
if not port_part.isdigit():
msg = f"Invalid port in '{value}'"
raise SystemExit(msg)
port = int(port_part)
return [{"host": "0.0.0.0", "port": port}, {"host": "::", "port": port}] # noqa: S104
def _parse_unix_socket(value: str) -> list[dict] | None:
"""Parse UNIX domain socket paths."""
if value.startswith("/"):
return [{"uds": value}]
if value.startswith("unix:"):
uds_path = value[5:] or None
if uds_path is None:
msg = "unix: path must not be empty"
raise SystemExit(msg)
return [{"uds": uds_path}]
return None
def _parse_unbracketed_ipv6(value: str, default_port: int) -> list[dict] | None:
"""Parse unbracketed IPv6 addresses."""
if value.count(":") <= 1 or value.startswith("["):
return None
try:
ipaddress.IPv6Address(value)
except ValueError as e:
msg = f"Invalid IPv6 address '{value}': {e}"
raise SystemExit(msg) from e
return [{"host": value, "port": default_port}]
def _parse_host_port(value: str, default_port: int) -> list[dict]:
"""Parse host[:port] or [ipv6][:port] using urlparse."""
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}]
def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
"""Parse an endpoint string into uvicorn bind configurations.
@@ -23,6 +75,7 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
- [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}]
@@ -31,51 +84,33 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
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
# Try specialized parsers in order
result = _parse_all_interfaces(value)
if result is not None:
return result
# 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}]
result = _parse_unix_socket(value)
if result is not None:
return result
# 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}]
result = _parse_unbracketed_ipv6(value, default_port)
if result is not None:
return result
# 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}]
# Fallback: host[:port], [ipv6][:port]
return _parse_host_port(value, default_port)
def parse_endpoints(
listen: str | list[str] | None = None, default_port: int = 8000
listen: str | list[str] | None = None,
default_port: int = 8000,
) -> list[dict]:
"""Parse listen strings into a list of endpoint dicts.
Args:
listen: Endpoint string(s) (see parse_endpoint for formats).
default_port: Port to use when not specified in listen args.
"""
if listen is None:
listen = [f"localhost:{default_port}"]
fastapi-vue/fastapi_vue/server.py
+11 -9
View File
@@ -1,7 +1,10 @@
"""Uvicorn server runner with multi-endpoint support."""
import asyncio
import logging
import os
from contextlib import suppress
from typing import Any
import uvicorn
from uvicorn import Config, Server
@@ -18,8 +21,8 @@ def run(
default_port: int = 8000,
reload: bool = False,
workers: int | None = None,
**uvicorn_config,
):
**uvicorn_config: Any, # noqa: ANN401
) -> None:
"""Run uvicorn server(s) for the given app.
Args:
@@ -29,10 +32,12 @@ def run(
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).
"""
endpoints = parse_endpoints(listen, default_port)
if not endpoints:
raise ValueError("No endpoints to serve; check listen configuration")
msg = "No endpoints to serve; check listen configuration"
raise ValueError(msg)
conf: dict[str, object] = {"app": app, "reload": reload, "workers": workers}
proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1")
@@ -48,7 +53,7 @@ def run(
asyncio.run(serve(endpoints, **conf))
async def serve(endpoints: list[dict], **kwargs) -> None:
async def serve(endpoints: list[dict], **kwargs: Any) -> None: # noqa: ANN401
"""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:
@@ -59,13 +64,10 @@ async def serve(endpoints: list[dict], **kwargs) -> None:
await asyncio.gather(*(Server(Config(**kwargs, **ep)).serve() for ep in endpoints))
def serve_multiprocess(endpoints: list[dict], **kwargs) -> None:
def serve_multiprocess(endpoints: list[dict], **kwargs: Any) -> None: # noqa: ANN401
"""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
]
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],
fastapi-vue/fastapi_vue/staticfiles.py
+36 -29
View File
@@ -25,7 +25,7 @@ __all__ = ["Frontend"]
class Assets:
"""Default cached value to /assets/"""
"""Default cached value to /assets/."""
@staticmethod
def parse(cached: str | list[str] | Assets) -> list[str]:
@@ -37,7 +37,8 @@ class Assets:
case list():
return cached
case _:
raise ValueError(f"Invalid cached value: {cached!r}")
msg = f"Invalid cached value: {cached!r}"
raise ValueError(msg)
class Frontend:
@@ -55,27 +56,29 @@ class Frontend:
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: May use wildcards of full path. E.g. /assets/logo*.png matches logo.hash.png created by Vite
favicon: Wildcard path to favicon. E.g. /assets/logo*.png matches Vite output
zstdlevel: Zstd compression level (default: 18)
"""
def __init__(
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 = Assets(),
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)
self.cached_paths = Assets.parse(cached if cached is not None else Assets())
self.zstdlevel = zstdlevel
self.favicon = favicon
self._app: FastAPI | None = None
@@ -107,11 +110,12 @@ class Frontend:
paths.add("/favicon.ico")
return paths
def _load(self):
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():
raise ValueError(f"Frontend folder {self.base} not found (try uv build)")
msg = f"Frontend folder {self.base} not found (try uv build)"
raise ValueError(msg)
paths = [PurePath()]
while paths:
current = self.base / paths.pop(0)
@@ -133,21 +137,18 @@ class Frontend:
headers = {
"etag": f'"{etag}"',
"last-modified": format_date_time(mtime),
"cache-control": (
"max-age=31536000, immutable" if cached else "no-cache"
),
"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:
if 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 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["/"] = (
@@ -161,7 +162,7 @@ class Frontend:
)
return www
async def load(self, *, debug: bool | None = None, log: bool = True):
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)
@@ -187,13 +188,19 @@ class Frontend:
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} %)"
"%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="/"):
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.
@@ -204,6 +211,7 @@ class Frontend:
Args:
app: FastAPI application instance
mount_path: Path where the frontend should be mounted (default: "/")
"""
self._app = app
self._mount_path = mount_path.rstrip("/")
@@ -214,7 +222,7 @@ class Frontend:
path = self._mount_path + "{path:path}"
app.api_route(path, methods=["GET", "HEAD"], name="frontend")(self.handle)
def _register_routes(self):
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
@@ -240,18 +248,19 @@ class Frontend:
for p in paths
]
def _respond(self, request: Request, name: str):
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"}
content=zstd,
headers={**headers, "content-encoding": "zstd"},
)
return Response(content=data, headers=headers)
def handle(self, request: Request, path: str):
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)
@@ -271,11 +280,9 @@ class Frontend:
return (_devmode_respond if debug else self._respond)(request, name)
def _devmode_respond(request: Request, name=""):
def _devmode_respond(_request: Request, _name: str = "") -> JSONResponse:
"""Return error response directing to Vite server."""
return JSONResponse(
status_code=409,
content={
"detail": "[devmode] Not serving frontend files here. Should you connect to Vite instead?"
},
content={"detail": "[devmode] Use Vite devserver instead."},
)