From c106ce1a44ddef37e5dd7e1d186c325548812807 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Mon, 2 Feb 2026 16:27:34 +0000 Subject: [PATCH] Add missing files. --- template/scripts/fastapi-vue/buildutil.py | 191 +++++++++++++++++++++ template/scripts/fastapi-vue/devutil.py | 197 ++++++++++++++++++++++ 2 files changed, 388 insertions(+) create mode 100644 template/scripts/fastapi-vue/buildutil.py create mode 100644 template/scripts/fastapi-vue/devutil.py diff --git a/template/scripts/fastapi-vue/buildutil.py b/template/scripts/fastapi-vue/buildutil.py new file mode 100644 index 0000000..8b8612d --- /dev/null +++ b/template/scripts/fastapi-vue/buildutil.py @@ -0,0 +1,191 @@ +"""Utilities used at build time and in devserver script. No dependencies.""" + +import logging +import os +import re +import shutil +import subprocess +from pathlib import Path + + +class _PrefixFormatter(logging.Formatter): + """Formatter that adds prefix based on log level.""" + + def format(self, record: logging.LogRecord) -> str: + if record.levelno >= logging.WARNING: + return f"⚠️ {record.getMessage()}" + return record.getMessage() + + +_handler = logging.StreamHandler() +_handler.setFormatter(_PrefixFormatter()) +logger = logging.getLogger("fastapi-vue") +logger.addHandler(_handler) +logger.setLevel(logging.INFO) + + +def _check_node_version(node_path: str) -> None: + """Check if Node.js version is >= 20. + + 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 + ) + 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: + return + raise RuntimeError( + f"Node.js {version_str} found, but v20+ required (install with nvm)" + ) + except (subprocess.CalledProcessError, FileNotFoundError, ValueError): + pass + raise RuntimeError("Could not determine Node.js version") + + +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. + """ + 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") + + # 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") + + +def find_build_tool(): + """Find JavaScript runtime and construct install/build commands. + + Returns (install_cmd, build_cmd) tuples of command lists. + Raises RuntimeError if no runtime is found. + """ + install = { + "deno": ("install", "--allow-scripts=npm:vue-demi"), + "npm": ("install",), + "bun": ("--bun", "install"), + } + # Run vite directly for deno to avoid npm-run-all2/run-p issues + build = { + "deno": ("run", "-A", "npm:vite", "build"), + "npm": ("run", "build"), + "bun": ("--bun", "run", "build"), + } + + tool, name = find_js_runtime() + return [tool, *install[name]], [tool, *build[name]] + + +def find_dev_tool() -> list[str]: + """Find JavaScript runtime and construct dev command. + + Returns dev_cmd (without vite-specific args). + Raises RuntimeError if no runtime is found. + """ + dev_args = { + "deno": ("run", "dev", "--"), + "npm": ("--silent", "run", "dev", "--"), + "bun": ("run", "dev", "--"), + } + + tool, name = find_js_runtime() + + if name == "bun": + logger.warning( + "Bun has a bug in WS proxying (https://github.com/oven-sh/bun/issues/9882). Consider using npm instead." + ) + + return [tool, *dev_args[name]] + + +def find_install_tool() -> list[str]: + """Find JavaScript runtime and construct install command. + + Returns install_cmd. + Raises RuntimeError if no runtime is found. + """ + install_args = { + "deno": ("install", "--quiet", "--allow-scripts=npm:vue-demi"), + "npm": ("install", "--silent"), + "bun": ("install", "--silent"), + } + + tool, name = find_js_runtime() + return [tool, *install_args[name]] + + +def build(folder: str = "frontend") -> None: + """Build the frontend in the specified folder. + + Raises SystemExit(1) on failure. + """ + logger.info(">>> Building %s", folder) + + try: + install_cmd, build_cmd = find_build_tool() + except RuntimeError as e: + logger.warning(e) + raise SystemExit(1) + + def run(cmd): + display_cmd = [Path(cmd[0]).name, *cmd[1:]] + logger.info("### %s", " ".join(display_cmd)) + subprocess.run(cmd, check=True, cwd=folder) + + try: + run(install_cmd) + logger.info("") + run(build_cmd) + except subprocess.CalledProcessError: + raise SystemExit(1) diff --git a/template/scripts/fastapi-vue/devutil.py b/template/scripts/fastapi-vue/devutil.py new file mode 100644 index 0000000..5d3ed37 --- /dev/null +++ b/template/scripts/fastapi-vue/devutil.py @@ -0,0 +1,197 @@ +"""Utilities meant for devserver script, used only in source repository with dev deps.""" + +import asyncio +import subprocess +from collections.abc import Coroutine +from pathlib import Path +from typing import Any + +import httpx +from buildutil import find_dev_tool, find_install_tool, logger +from fastapi_vue.hostutil import parse_endpoint + +DEFAULT_VITE_PORT = 5173 +DEFAULT_BACKEND_PORT = 5180 + + +class ProcessGroup: + """Manage async subprocesses with automatic cleanup, like TaskGroup for processes.""" + + def __init__(self): + self._procs: list[asyncio.subprocess.Process] = [] + self._cmds: dict[int, str] = {} # pid -> command name + + async def spawn( + self, *cmd: str, cwd: str | None = None + ) -> asyncio.subprocess.Process: + """Spawn a subprocess and track it.""" + cmd_name = Path(cmd[0]).name + logger.info(">>> %s", " ".join([cmd_name, *cmd[1:]])) + proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd) + self._procs.append(proc) + self._cmds[proc.pid] = cmd_name + return proc + + async def wait( + self, *waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]" + ) -> None: + """Wait for processes/coroutines to complete, raise SystemExit on failure.""" + + async def wait_proc(proc: asyncio.subprocess.Process) -> None: + returncode = await proc.wait() + if returncode != 0: + cmd_name = self._cmds.get(proc.pid, "unknown") + raise subprocess.CalledProcessError(returncode, cmd_name) + + tasks = [ + wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w + for w in waitables + ] + try: + await asyncio.gather(*tasks) + except subprocess.CalledProcessError as e: + logger.warning("%s failed with exit status %d", e.cmd, e.returncode) + raise SystemExit(1) from None + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, *_): + """Wait for one process to exit, terminate others, then wait for all.""" + cleanup_task = asyncio.create_task( + self._cleanup(immediate=exc_type is not None) + ) + try: + await asyncio.shield(cleanup_task) + except asyncio.CancelledError: + # Shield was cancelled but cleanup_task continues - wait for it + await cleanup_task + + async def _cleanup(self, immediate: bool = False): + running = [p for p in self._procs if p.returncode is None] + if not running: + return + + if not immediate: + # Wait for any one process to exit + await asyncio.wait( + [asyncio.create_task(p.wait()) for p in running], + return_when=asyncio.FIRST_COMPLETED, + ) + + # Terminate remaining processes + for p in self._procs: + if p.returncode is None: + try: + p.terminate() + except ProcessLookupError: + pass + + # Wait for all to finish (with overall timeout) + still_running = [p for p in self._procs if p.returncode is None] + if still_running: + try: + await asyncio.wait_for( + asyncio.gather(*[p.wait() for p in still_running]), + timeout=10, + ) + except TimeoutError: + for p in self._procs: + if p.returncode is None: + try: + p.kill() + except ProcessLookupError: + pass + await p.wait() + + +async def check_ports_free(*urls: str) -> None: + """Verify URLs are not responding (ports are free). Raise SystemExit if any respond.""" + async with httpx.AsyncClient() as client: + for url in urls: + try: + res = await client.get(url, timeout=0.1) + logger.warning( + "Conflicting %s already running at %s", + res.headers.get("server", "server"), + url, + ) + raise SystemExit(1) + except httpx.RequestError: + pass # Expected - port is free + + +async def ready(url: str, path: str = "") -> None: + """Wait for the server to be ready by polling an endpoint. + + Raises SystemExit(1) if server doesn't start in time. + """ + max_attempts = 50 + full_url = f"{url}{path}" + + 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) + + +def setup_vite(endpoint: str) -> tuple[str, list[str], list[str]]: + """Parse frontend endpoint and build commands. + + Returns (url, install_cmd, dev_cmd). + Raises SystemExit(1) on invalid config. + """ + endpoints = parse_endpoint(endpoint, DEFAULT_VITE_PORT) + + if "uds" in endpoints[0]: + logger.warning("Unix sockets not supported with vite devserver") + raise SystemExit(1) + + port = endpoints[0]["port"] + host = endpoints[0]["host"] + + install_cmd = find_install_tool() + dev_cmd = find_dev_tool() + if host != "localhost": + dev_cmd.append("--host" if len(endpoints) > 1 else f"--host={host}") + if port != 5173: + dev_cmd.append(f"--port={port}") + + return f"http://{host}:{port}", install_cmd, dev_cmd + + +def setup_fastapi( + endpoint: str, module: str, default_port: int = DEFAULT_BACKEND_PORT +) -> tuple[str, list[str]]: + """Parse backend endpoint and build fastapi dev command. + + Returns (url, cmd). + Raises SystemExit(1) on invalid config. + """ + endpoints = parse_endpoint(endpoint, default_port) + + if "uds" in endpoints[0]: + logger.warning("Unix sockets not supported with vite devserver") + raise SystemExit(1) + + host = endpoints[0]["host"] + port = endpoints[0]["port"] + + cmd = [ + "fastapi", + "dev", + "--entrypoint", + module, + "--host", + host, + "--port", + str(port), + ] + return f"http://{host}:{port}", cmd