LeoVasanko/paskia
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Update fastapi-vue-setup, make use of its access logging facility.

Browse Source
This commit is contained in:
LeoVasanko
2026-09-05 14:36:25 +00:00
parent 3912b5473e
commit 8c2809a879
10 changed files with 204 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/devserver.py
+3
View File
@@ -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",
scripts/fastapi-vue/build-frontend.py → scripts/fastapi-vue/buildhook.py
+7 -3
View File
@@ -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")
scripts/fastapi-vue/buildutil.py
+98 -58
View File
@@ -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
scripts/fastapi-vue/devutil.py
+78 -36
View File
@@ -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