From 8c2809a879eb752932ca477bbe625060ae63e490 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Sat, 5 Sep 2026 14:35:49 +0000 Subject: [PATCH] Update fastapi-vue-setup, make use of its access logging facility. --- frontend/vite-plugin-fastapi.js | 2 + paskia/__main__.py | 9 +- paskia/fastapi/logging.py | 249 +----------------- paskia/fastapi/mainapp.py | 13 +- paskia/fastapi/wsutil.py | 12 +- pyproject.toml | 2 +- scripts/devserver.py | 3 + .../{build-frontend.py => buildhook.py} | 10 +- scripts/fastapi-vue/buildutil.py | 156 +++++++---- scripts/fastapi-vue/devutil.py | 114 +++++--- 10 files changed, 204 insertions(+), 366 deletions(-) rename scripts/fastapi-vue/{build-frontend.py => buildhook.py} (50%) diff --git a/frontend/vite-plugin-fastapi.js b/frontend/vite-plugin-fastapi.js index ffef157..e7e5945 100644 --- a/frontend/vite-plugin-fastapi.js +++ b/frontend/vite-plugin-fastapi.js @@ -5,6 +5,7 @@ * Configures Vite for FastAPI backend integration: * - Proxies /api/* requests to the FastAPI backend * - Builds to the Python module's frontend-build directory + * - Disables Vite's screen clearing on startup * * Options: * paths - Array of paths to proxy (default: ["/api"]) @@ -26,6 +27,7 @@ export default function fastapiVue({ paths = ["/api"] } = {}) { return { name: "vite-plugin-fastapi-paskia", config: () => ({ + clearScreen: false, server: { proxy }, build: { outDir: "../paskia/frontend-build", diff --git a/paskia/__main__.py b/paskia/__main__.py index a66b1b0..4c4a15f 100644 --- a/paskia/__main__.py +++ b/paskia/__main__.py @@ -166,14 +166,15 @@ def main(): os.environ["PASKIA_CONFIG"] = msgspec.json.encode(runtime).decode() # Run the server (spawns processes in dev mode) - dev = {"reload": True, "reload_dirs": ["paskia"]} if DEVMODE else {} + # tracerite, access logging and log config are handled by fastapi_vue.server; + # we print our own startup config box, so disable the built-in one. server.run( "paskia.fastapi.mainapp:app", listen=config.listen, default_port=DEFAULT_PORT, - log_level="warning", - access_log=False, - **dev, + server_header=False, + startup_box=None, + reload=Path(__file__).parent if DEVMODE else False, ) diff --git a/paskia/fastapi/logging.py b/paskia/fastapi/logging.py index 60a19db..6b74657 100644 --- a/paskia/fastapi/logging.py +++ b/paskia/fastapi/logging.py @@ -1,34 +1,19 @@ -"""Custom access logging middleware for FastAPI/Uvicorn.""" +"""Authorization-related logging. + +HTTP/WebSocket access logging is handled by fastapi_vue's ASGI middleware +(installed via fastapi_vue.server.run); request handlers can pass extra +details to the access log line via request.state.log_extra. +""" import logging -import sys -import time -from ipaddress import IPv6Address from typing import TYPE_CHECKING -from starlette.middleware.base import BaseHTTPMiddleware - if TYPE_CHECKING: from paskia.db.structs import SessionContext -from starlette.requests import Request -from starlette.responses import Response logger = logging.getLogger("paskia.access") _RESET = "\033[0m" -_STATUS_INFO = "\033[32m" # 1xx (green) -_STATUS_OK = "\033[1;92m" # 2xx (bright green) -_STATUS_REDIRECT = "\033[32m" # 3xx (green) -_STATUS_CLIENT_ERR = "\033[0;31m" # 4xx (red) -_STATUS_SERVER_ERR = "\033[1;91m" # 5xx (bold bright red) -_METHOD_READ = "\033[0;34m" # GET, HEAD, OPTIONS (blue) -_METHOD_WRITE = "\033[1;94m" # POST, PUT, DELETE, PATCH (bold bright blue) -_HOST = "\033[38;5;242m" # hostname (dark grey) -_PATH = "\033[38;5;250m" # path (white) -_TIMING = "\033[38;5;242m" # timing/devmode (dark grey) -_WS_OPEN = "\033[38;5;226m" # WebSocket connect (brightest yellow from 6x6x6 cube) -_WS_CLOSE = "\033[38;5;142m" # WebSocket disconnect (significantly dimmer yellow) -_WS_STATUS = "\033[38;5;242m" # WebSocket close status (dark grey) _AUTHZ_DENIED = "\033[0;31m" # Permission denied (red) _AUTHZ_USER = "\033[1;34m" # User info (light blue) _AUTHZ_ORG = "\033[34m" # User info (blue) @@ -37,191 +22,6 @@ _AUTHZ_MISSING = "\033[1;31m" # Missing scope (bold red) _AUTHZ_GRANTED = "\033[0;32m" # Granted scope (green) -def format_ipv6_network(ip: str) -> str: - """Format IPv6 address to show only network part (first 64 bits). - - Special addresses are returned as-is for clarity: - - ::1 (loopback) - - :: (unspecified) - - ::ffff:x.x.x.x (IPv4-mapped, returns just the IPv4 part) - - fe80:: (link-local, returned as-is since interface-specific) - """ - try: - # Strip brackets that some proxies add around IPv6 - ip = ip.strip("[]") - # Strip zone ID (e.g., fe80::1%eth0) - if "%" in ip: - ip = ip.split("%")[0] - addr = IPv6Address(ip) - - # Special cases - return as-is or with minimal processing - if addr.is_loopback: # ::1 - return "::1" - if addr.is_unspecified: # :: - return "::" - if addr.ipv4_mapped: # ::ffff:x.x.x.x - return str(addr.ipv4_mapped) - if addr.is_link_local: # fe80::/10 - interface-specific, keep full - return str(addr) - - # Regular addresses: truncate to /64 network prefix - network_int = int(addr) >> 64 - # Format as IPv6 with trailing :: - # Split into 4 groups of 16 bits - groups = [] - for _ in range(4): - groups.insert(0, format(network_int & 0xFFFF, "x")) - network_int >>= 16 - # Compress consecutive zero groups - result = ":".join(groups) + "::" - # Simplify leading zeros in groups and compress, then strip trailing :: - return str(IPv6Address(result + "0")).removesuffix("::") - except Exception: - return ip - - -def format_client_ip(ip: str) -> str: - """Format client IP, compressing IPv6 to network part only.""" - if not ip or ip == "-": - return "-" - # Strip brackets for detection (some proxies add them) - stripped = ip.strip("[]") - if ":" in stripped: - return format_ipv6_network(ip) - return ip - - -def status_color(status: int) -> str: - """Return color code based on HTTP status.""" - if status < 200: - return _STATUS_INFO - if status < 300: - return _STATUS_OK - if status < 400: - return _STATUS_REDIRECT - if status < 500: - return _STATUS_CLIENT_ERR - return _STATUS_SERVER_ERR - - -def method_color(method: str) -> str: - """Return color code based on HTTP method.""" - if method in ("GET", "HEAD", "OPTIONS"): - return _METHOD_READ - return _METHOD_WRITE - - -def format_access_log( - client: str, - status: int, - method: str, - host: str, - path: str, - duration_ms: float, - extra: str = "", -) -> str: - """Format access log line with colors and aligned fields.""" - # Format components with fixed widths for alignment - ip = format_client_ip(client).ljust(19) # IPv6 network max 19 chars - timing = f"{duration_ms:.0f}ms" - method_padded = method.ljust(7) # Longest method is OPTIONS (7) - - status_str = f"{status_color(status)}{status}{_RESET}" - timing_str = f"{_TIMING}{timing}{_RESET}" - method_str = f"{method_color(method)}{method_padded}{_RESET}" - host_str = f"{_HOST}{host}{_RESET}" - path_str = f"{_PATH}{path}{_RESET}" - - # Format: "IP STATUS METHOD host path [extra] TIMING" - extra_str = f" {_TIMING}{extra}{_RESET}" if extra else "" - return ( - f"{ip} {status_str} {method_str} {host_str}{path_str}{extra_str} {timing_str}" - ) - - -# WebSocket connection counter (mod 100) -_ws_counter = 0 - - -def _next_ws_id() -> int: - """Get next WebSocket connection ID (0-99).""" - global _ws_counter - ws_id = _ws_counter - _ws_counter = (_ws_counter + 1) % 100 - return ws_id - - -def log_ws_open(ws) -> int: - """Log WebSocket connection open. Returns connection ID for use in close.""" - ws_id = _next_ws_id() - - client = ws.client.host if ws.client else "-" - host = ws.headers.get("host", "-") - path = ws.url.path - origin = ws.headers.get("origin") - - ip = format_client_ip(client).ljust(19) - # ID right-aligned like status codes (3 chars), emoji formatted like method - id_str = f"{_WS_OPEN}{str(ws_id).rjust(3)}{_RESET}" - # Emoji (2 display width) + 6 spaces = 8 display chars, but within color for alignment - emoji_str = f"{_METHOD_READ}🔌 {_RESET}" - - # Determine if origin should be shown (omit when same as host) - # Origin header includes scheme (e.g., "https://example.com"), compare host part - origin_host = origin.split("://", 1)[-1] if origin else None - show_origin = origin_host and origin_host != host - - host_str = f"{_HOST}{host}{_RESET}" - path_str = f"{_PATH}{path}{_RESET}" - origin_str = f" {_RESET}from {_HOST}{origin_host}{_RESET}" if show_origin else "" - - logger.info(f"{ip} {id_str} {emoji_str}{host_str}{path_str}{origin_str}") - return ws_id - - -# WebSocket close codes to human-readable status -WS_CLOSE_CODES = { - 1000: "ok", - 1001: "going away", - 1002: "protocol error", - 1003: "unsupported", - 1005: "no status", - 1006: "abnormal", - 1007: "invalid data", - 1008: "policy violation", - 1009: "too large", - 1010: "extension required", - 1011: "server error", - 1012: "restarting", - 1013: "try again", - 1014: "bad gateway", - 1015: "tls error", -} - - -def log_ws_close(ws_id: int, close_code: int | None, duration: float) -> None: - """Log WebSocket connection close with duration and status.""" - # ID right-aligned like status codes (3 chars), "closed" formatted like method - id_str = f"{_WS_CLOSE}{str(ws_id).rjust(3)}{_RESET}" - # Pad within the dim color to keep full width in color (8 display chars) - closed_str = f"{_TIMING}closed {_RESET}" - timing = f"{duration * 1000:.0f}ms" - - # Convert close code to status text - if close_code is None: - code = "----" - status = "unknown" - else: - code = str(close_code) - status = WS_CLOSE_CODES.get(close_code, f"code {close_code}") - - # Status code and text in normal color, not dim - status_str = f"{code} {status}" - timing_str = f"{_TIMING}{timing}{_RESET}" - - logger.info(f"{' ' * 19} {id_str} {closed_str}{status_str} {timing_str}") - - def log_permission_denied( ctx: SessionContext, required: list[str], missing: list[str], *, require_all: bool ) -> None: @@ -240,40 +40,3 @@ def log_permission_denied( f"{_AUTHZ_ORG}({ctx.org.display_name} {ctx.role.display_name}){_RESET} " f"{_AUTHZ_NEEDS}needs{n}:{_RESET} {scopes}" ) - - -class AccessLogMiddleware(BaseHTTPMiddleware): - """Middleware that logs HTTP requests with custom format.""" - - async def dispatch(self, request: Request, call_next) -> Response: - start = time.perf_counter() - response = await call_next(request) - duration_ms = (time.perf_counter() - start) * 1000 - - client = request.client.host if request.client else "-" - host = request.headers.get("host", "-") - method = request.method - path = request.url.path - if request.url.query: - path = f"{path}?{request.url.query}" - status = response.status_code - - extra = getattr(request.state, "log_extra", "") - - line = format_access_log( - client, status, method, host, path, duration_ms, extra=extra - ) - logger.info(line) - - return response - - -def configure_access_logging(): - """Configure the access logger to output to stderr.""" - handler = logging.StreamHandler(sys.stderr) - handler.setFormatter(logging.Formatter("%(message)s")) - logger.addHandler(handler) - logger.setLevel(logging.INFO) - logger.propagate = False - # Suppress watchfiles "X changes detected" INFO messages (keep WARNING for reload notification) - logging.getLogger("watchfiles.main").setLevel(logging.WARNING) diff --git a/paskia/fastapi/mainapp.py b/paskia/fastapi/mainapp.py index d502c8f..55494c1 100644 --- a/paskia/fastapi/mainapp.py +++ b/paskia/fastapi/mainapp.py @@ -18,18 +18,14 @@ from paskia.fastapi.admin.adminapp import adminapp # Import frontend instance from paskia.fastapi.front import frontend -from paskia.fastapi.logging import AccessLogMiddleware, configure_access_logging from paskia.fastapi.session import AUTH_COOKIE from paskia.util import hostutil, passphrase, vitedev from paskia.util.constants import DEVMODE from paskia.util.runtime import RuntimeConfig # Configure custom logging -configure_access_logging() configure_kanta_logging() -_access_logger = logging.getLogger("paskia.access") - # Path to examples/index.html when running from source tree _EXAMPLES_DIR = Path(__file__).parent.parent.parent / "examples" @@ -61,11 +57,6 @@ async def lifespan(app: FastAPI): # pragma: no cover - startup path if runtime.save: db.update_config(runtime.config) - # Restore uvicorn info logging (suppressed during startup in dev mode) - # Keep uvicorn.error at WARNING to suppress WebSocket "connection open/closed" messages - if app.debug: - logging.getLogger("uvicorn").setLevel(logging.INFO) - logging.getLogger("uvicorn.error").setLevel(logging.WARNING) await frontend.load() await start_background() yield @@ -82,8 +73,8 @@ app = FastAPI( debug=DEVMODE, ) -# Custom access logging (uvicorn's access_log is disabled) -app.add_middleware(AccessLogMiddleware) +# WebSocket and HTTP access logging is handled by fastapi_vue's ASGI middleware; +# extra details are passed via request.state.log_extra (ASGI scope state). # Apply redirections to auth-host if configured (deny access to restricted endpoints, remove /auth/) app.middleware("http")(auth_host.redirect_middleware) diff --git a/paskia/fastapi/wsutil.py b/paskia/fastapi/wsutil.py index d2ebec4..d547915 100644 --- a/paskia/fastapi/wsutil.py +++ b/paskia/fastapi/wsutil.py @@ -3,7 +3,6 @@ Shared WebSocket utilities for FastAPI endpoints. """ import logging -import time from functools import wraps import base64url @@ -11,7 +10,6 @@ from fastapi import WebSocket, WebSocketDisconnect from webauthn.helpers.exceptions import InvalidAuthenticationResponse from paskia.fastapi import authz -from paskia.fastapi.logging import log_ws_close, log_ws_open from paskia.globals import passkey from paskia.util import pow @@ -21,15 +19,11 @@ def websocket_error_handler(func): @wraps(func) async def wrapper(ws: WebSocket, *args, **kwargs): - start = time.perf_counter() - ws_id = log_ws_open(ws) - close_code = None - try: await ws.accept() return await func(ws, *args, **kwargs) - except WebSocketDisconnect as e: - close_code = e.code + except WebSocketDisconnect: + pass except authz.AuthException as e: await ws.send_json( { @@ -42,8 +36,6 @@ def websocket_error_handler(func): except Exception: logging.exception("Internal Server Error") await ws.send_json({"status": 500, "detail": "Internal Server Error"}) - finally: - log_ws_close(ws_id, close_code, time.perf_counter() - start) return wrapper diff --git a/pyproject.toml b/pyproject.toml index 83a349a..395306b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,7 +21,7 @@ dependencies = [ "pyjwt[crypto]>=2.11.0", "jsondiff>=2.2.1", "msgspec>=0.20.0", - "fastapi-vue>=1.1.0", + "fastapi-vue~=1.4.2", "ua-parser[regex]>=1.0.1", "kanta>=0.7.0", ] diff --git a/scripts/devserver.py b/scripts/devserver.py index ac5fda2..c30cf40 100755 --- a/scripts/devserver.py +++ b/scripts/devserver.py @@ -11,6 +11,8 @@ from contextlib import suppress from pathlib import Path from urllib.parse import urlparse +import tracerite + # Import utilities from scripts/fastapi-vue (not a package, so we adjust sys.path) sys.path.insert(0, str(Path(__file__).with_name("fastapi-vue"))) from devutil import ( # noqa: E402 @@ -193,6 +195,7 @@ async def run_devserver(args: argparse.Namespace, remaining: list[str]) -> None: def main(): + tracerite.load() parser = argparse.ArgumentParser(add_help=False) parser.add_argument( "-l", diff --git a/scripts/fastapi-vue/build-frontend.py b/scripts/fastapi-vue/buildhook.py similarity index 50% rename from scripts/fastapi-vue/build-frontend.py rename to scripts/fastapi-vue/buildhook.py index 152c48c..407e4bf 100644 --- a/scripts/fastapi-vue/build-frontend.py +++ b/scripts/fastapi-vue/buildhook.py @@ -1,15 +1,19 @@ +# ruff: noqa: INP001 """Hatch build hook for building Vue frontend during package build.""" import sys from pathlib import Path -from hatchling.builders.hooks.plugin.interface import BuildHookInterface # type: ignore +from hatchling.builders.hooks.plugin.interface import BuildHookInterface sys.path.insert(0, str(Path(__file__).parent)) from buildutil import build -class CustomBuildHook(BuildHookInterface): - def initialize(self, version, build_data): +class CustomBuildHook(BuildHookInterface): # type: ignore[misc] + """Hatch build hook that builds Vue frontend during package build.""" + + def initialize(self, version: str, build_data: dict) -> None: # type: ignore[override] + """Build frontend before package is built.""" super().initialize(version, build_data) build("frontend") diff --git a/scripts/fastapi-vue/buildutil.py b/scripts/fastapi-vue/buildutil.py index c2b641e..3150423 100644 --- a/scripts/fastapi-vue/buildutil.py +++ b/scripts/fastapi-vue/buildutil.py @@ -1,3 +1,4 @@ +# ruff: noqa: INP001 """Utilities used at build time and in devserver script. No dependencies.""" import logging @@ -7,6 +8,8 @@ import shutil import subprocess from pathlib import Path +MIN_NODE_VERSION = 20 + class _PrefixFormatter(logging.Formatter): """Formatter that adds prefix based on log level.""" @@ -30,82 +33,119 @@ def _check_node_version(node_path: str) -> None: Raises RuntimeError if version is too old or cannot be determined. """ try: - result = subprocess.run( - [node_path, "--version"], capture_output=True, text=True, check=True + result = subprocess.run( # noqa: S603 + [node_path, "--version"], + capture_output=True, + text=True, + check=True, ) version_str = result.stdout.strip() # Parse version like "v20.10.0" or "v18.17.1" match = re.match(r"v(\d+)", version_str) if match: major_version = int(match.group(1)) - if major_version >= 20: + if major_version >= MIN_NODE_VERSION: return - raise RuntimeError( - f"Node.js {version_str} found, but v20+ required (install with nvm)" - ) + msg = f"Node.js {version_str} found, but v{MIN_NODE_VERSION}+ required" + raise RuntimeError(msg) except subprocess.CalledProcessError, FileNotFoundError, ValueError: pass - raise RuntimeError("Could not determine Node.js version") + msg = "Could not determine Node.js version" + raise RuntimeError(msg) + + +def _validate_npm_runtime(tool: str) -> bool: + """Validate npm runtime by checking Node.js version. Returns True if valid.""" + node_path = shutil.which("node", path=str(Path(tool).parent)) + if node_path is None: + return False + try: + _check_node_version(node_path) + except RuntimeError: + return False + return True + + +def _find_runtime_from_env(options: list[str]) -> tuple[str, str] | None: + """Find runtime specified by JS_RUNTIME environment variable.""" + js_runtime_env = os.environ.get("JS_RUNTIME") + if not js_runtime_env: + return None + + js_runtime = js_runtime_env + js_path = Path(js_runtime) + runtime_name = js_path.name + + # Map node to npm + if runtime_name == "node": + runtime_name = "npm" + js_runtime = str(js_path.parent / "npm") if js_path.parent.name else "npm" + + for option in options: + if option != runtime_name and not runtime_name.startswith(option): + continue + + tool = shutil.which(js_runtime) + if tool is None: + msg = f"JS_RUNTIME={js_runtime_env}: {option} not found" + raise RuntimeError(msg) + + if option == "npm": + node_path = shutil.which("node", path=str(Path(tool).parent)) + if node_path is None: + msg = f"JS_RUNTIME={js_runtime_env}: node not found" + raise RuntimeError(msg) + _check_node_version(node_path) + + return tool, option + + msg = f"JS_RUNTIME={js_runtime_env} not recognized" + raise RuntimeError(msg) + + +def _auto_detect_runtime(options: list[str]) -> tuple[str, str]: + """Auto-detect JavaScript runtime from available options.""" + node_version_error: RuntimeError | None = None + + for option in options: + tool = shutil.which(option) + if not tool: + continue + + if option == "npm" and not _validate_npm_runtime(tool): + try: + node_path = shutil.which("node", path=str(Path(tool).parent)) + if node_path: + _check_node_version(node_path) + except RuntimeError as e: + node_version_error = e + continue + + return tool, option + + if node_version_error: + raise node_version_error + msg = "Node.js (v20+), Deno or Bun is required but none was found" + raise RuntimeError(msg) def find_js_runtime() -> tuple[str, str]: """Find a JavaScript runtime from JS_RUNTIME env or auto-detect. Returns (tool_path, tool_name) where tool_name is "deno", "npm", or "bun". - Raises JSRuntimeError if no suitable runtime is found. + Raises RuntimeError if no suitable runtime is found. """ options = ["npm", "deno", "bun"] - node_version_error: RuntimeError | None = None # Check for JS_RUNTIME environment variable - if js_runtime_env := os.environ.get("JS_RUNTIME"): - js_runtime = js_runtime_env - js_path = Path(js_runtime) - runtime_name = js_path.name - # Map node to npm - if runtime_name == "node": - runtime_name = "npm" - js_runtime = str(js_path.parent / "npm") if js_path.parent.name else "npm" - for option in options: - if option == runtime_name or runtime_name.startswith(option): - tool = shutil.which(js_runtime) - if tool is None: - raise RuntimeError( - f"JS_RUNTIME={js_runtime_env}: {option} not found" - ) - # Check Node.js version if using npm - if option == "npm": - node_path = shutil.which("node", path=str(Path(tool).parent)) - if node_path is None: - raise RuntimeError( - f"JS_RUNTIME={js_runtime_env}: node not found" - ) - _check_node_version(node_path) # Raises on failure - return tool, option - raise RuntimeError(f"JS_RUNTIME={js_runtime_env} not recognized") + if result := _find_runtime_from_env(options): + return result # Auto-detect - for option in options: - if tool := shutil.which(option): - # Check Node.js version if using npm - if option == "npm": - node_path = shutil.which("node", path=str(Path(tool).parent)) - if node_path is None: - continue - try: - _check_node_version(node_path) - except RuntimeError as e: - node_version_error = e - continue # Try next runtime - return tool, option - - # No runtime found - provide helpful error - if node_version_error: - raise node_version_error - raise RuntimeError("Node.js (v20+), Deno or Bun is required but none was found") + return _auto_detect_runtime(options) -def find_build_tool(): +def find_build_tool() -> tuple[list[str], list[str]]: """Find JavaScript runtime and construct install/build commands. Returns (install_cmd, build_cmd) tuples of command lists. @@ -143,7 +183,7 @@ def find_dev_tool() -> list[str]: if name == "bun": logger.warning( - "Bun has a bug in WS proxying (https://github.com/oven-sh/bun/issues/9882). Consider using npm instead." + "Bun has a WS proxy bug (github.com/oven-sh/bun/issues/9882). Consider npm.", ) return [tool, *dev_args[name]] @@ -176,16 +216,16 @@ def build(folder: str = "frontend") -> None: install_cmd, build_cmd = find_build_tool() except RuntimeError as e: logger.warning(e) - raise SystemExit(1) + raise SystemExit(1) from None - def run(cmd): + def run(cmd: list[str]) -> None: display_cmd = [Path(cmd[0]).stem, *cmd[1:]] logger.info("### %s", " ".join(display_cmd)) - subprocess.run(cmd, check=True, cwd=folder) + subprocess.run(cmd, check=True, cwd=folder) # noqa: S603 try: run(install_cmd) logger.info("") run(build_cmd) except subprocess.CalledProcessError: - raise SystemExit(1) + raise SystemExit(1) from None diff --git a/scripts/fastapi-vue/devutil.py b/scripts/fastapi-vue/devutil.py index bb8e0de..72b34b6 100644 --- a/scripts/fastapi-vue/devutil.py +++ b/scripts/fastapi-vue/devutil.py @@ -1,27 +1,33 @@ +# ruff: noqa: INP001 """Utilities meant for devserver script, used only in source repository with dev deps.""" import asyncio import subprocess import sys -from collections.abc import Coroutine from contextlib import suppress from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any, Self +from urllib.parse import urlsplit -import httpx from buildutil import find_dev_tool, find_install_tool, logger from fastapi_vue.hostutil import parse_endpoint +if TYPE_CHECKING: + from collections.abc import Coroutine + class ProcessGroup: """Manage async subprocesses with automatic cleanup, like TaskGroup for processes.""" - def __init__(self): + def __init__(self) -> None: + """Initialize empty process tracking.""" self._procs: list[asyncio.subprocess.Process] = [] self._cmds: dict[int, str] = {} # pid -> command name async def spawn( - self, *cmd: str, cwd: str | None = None + self, + *cmd: str, + cwd: str | None = None, ) -> asyncio.subprocess.Process: """Spawn a subprocess and track it.""" cmd_name = Path(cmd[0]).stem @@ -32,7 +38,8 @@ class ProcessGroup: return proc async def wait( - self, *waitables: asyncio.subprocess.Process | Coroutine[Any, Any, Any] + self, + *waitables: asyncio.subprocess.Process | Coroutine[Any, Any, Any], ) -> None: """Wait for processes/coroutines to complete, raise SystemExit on failure.""" @@ -52,14 +59,15 @@ class ProcessGroup: logger.warning("%s failed with exit status %d", e.cmd, e.returncode) raise SystemExit(1) from None - async def __aenter__(self): + async def __aenter__(self) -> Self: + """Enter the async context manager.""" return self - async def __aexit__(self, exc_type, *_): + async def __aexit__(self, exc_type: type[BaseException] | None, *_: object) -> None: """Wait for one process to exit, terminate others, then wait for all.""" await self._cleanup(immediate=exc_type is not None) - async def _cleanup(self, immediate: bool = False): + async def _cleanup(self, *, immediate: bool = False) -> None: running = [p for p in self._procs if p.returncode is None] if not running: return @@ -87,7 +95,7 @@ class ProcessGroup: asyncio.wait_for( asyncio.gather(*[p.wait() for p in still_running]), timeout=10, - ) + ), ) except TimeoutError: for p in self._procs: @@ -97,43 +105,71 @@ class ProcessGroup: await p.wait() +async def http_get_server(url: str, timeout: float) -> str | None: # noqa: ASYNC109 + """GET url with plain asyncio streams, return the response Server header. + + Returns an empty string when the server responds without a Server header, + and None when the server is unreachable or doesn't answer in time. + """ + parts = urlsplit(url) + host = parts.hostname or "localhost" + port = parts.port or (443 if parts.scheme == "https" else 80) + path = parts.path or "/" + if parts.query: + path += f"?{parts.query}" + try: + async with asyncio.timeout(timeout): + reader, writer = await asyncio.open_connection(host, port) + try: + writer.write(f"GET {path} HTTP/1.0\r\nHost: {host}\r\n\r\n".encode()) + await writer.drain() + data = await reader.readuntil(b"\r\n\r\n") + finally: + writer.close() + except OSError, EOFError, ValueError, TimeoutError: + return None + for line in data.decode("latin-1").split("\r\n"): + if line.lower().startswith("server:"): + return line.split(":", 1)[1].strip() + return "" + + async def check_ports_free(*urls: str) -> None: """Verify URLs are not responding (ports are free). Raise SystemExit if any respond.""" - async def check(client: httpx.AsyncClient, url: str) -> None: - with suppress(httpx.RequestError): - res = await client.get(url, timeout=0.1) - server = res.headers.get("server", "server") - logger.warning("Conflicting %s already running at %s", server, url) + async def check(url: str) -> None: + server = await http_get_server(url, timeout=0.1) + if server is not None: + logger.warning( + "Conflicting %s already running at %s", server or "server", url + ) raise SystemExit(1) - async with httpx.AsyncClient() as client: - await asyncio.gather(*[check(client, url) for url in urls]) + await asyncio.gather(*[check(url) for url in urls]) -async def ready(url: str, path: str = "") -> None: +async def ready(url: str, path: str = "", max_attempts: int = 50) -> None: """Wait for the server to be ready by polling an endpoint. + Use empty path to disable the check and make this return immediately. Raises SystemExit(1) if server doesn't start in time. """ - max_attempts = 50 - full_url = f"{url}{path}" + if not path: + return - async with httpx.AsyncClient() as client: - for attempt in range(max_attempts): - try: - await client.get(full_url, timeout=1.0) - logger.info("✓ Backend ready!") - return - except httpx.RequestError: - if attempt == max_attempts - 1: - logger.warning("Backend didn't start in time") - raise SystemExit(1) - await asyncio.sleep(0.1) + for attempt in range(max_attempts): + if await http_get_server(f"{url}{path}", timeout=1.0) is not None: + logger.info("✓ Backend ready!") + return + if attempt == max_attempts - 1: + logger.warning("Backend didn't start in time") + raise SystemExit(1) + await asyncio.sleep(0.1) def setup_vite( - endpoint: str, default_port: int = 5173 + endpoint: str, + default_port: int = 5173, ) -> tuple[str, list[str], list[str]]: """Parse frontend endpoint and build commands. @@ -159,7 +195,9 @@ def setup_vite( def setup_fastapi( - endpoint: str, module: str, default_port: int = 8000 + endpoint: str, + module: str, + default_port: int = 8000, ) -> tuple[str, list[str]]: """Parse backend endpoint and build uvicorn command. @@ -174,7 +212,7 @@ def setup_fastapi( host = endpoints[0]["host"] port = endpoints[0]["port"] - reload_dir = module.split(".")[0] # Don't reload on frontend changes + reload_dir = module.split(".", maxsplit=1)[0] # Don't reload on frontend changes cmd = [ sys.executable, @@ -191,7 +229,9 @@ def setup_fastapi( def setup_cli( - cli: str, endpoint: str, default_port: int = 8000 + cli: str, + endpoint: str, + default_port: int = 8000, ) -> tuple[str, list[str]]: """Parse backend endpoint and build CLI command. @@ -207,5 +247,7 @@ def setup_cli( host = endpoints[0]["host"] port = endpoints[0]["port"] - cmd = [cli, f"--listen={host}:{port}"] + # Run the package as a module with the current interpreter, instead of + # relying on a PATH-installed CLI entry point. + cmd = [sys.executable, "-m", cli, f"--listen={host}:{port}"] return f"http://{host}:{port}", cmd