Proper error messages on npm and port check failures, avoid leaking asyncio tasks.

- Terminates instantly when vital process dies (e.g. backend doesn't start)
- Derived of TaskGroup, augmenting its functionality with similar semantics

README.md

@@ -1,135 +1,96 @@
 # fastapi-vue-setup
 
-Tool to create or patch FastAPI project with a Vue frontend, with integrated build and development systems. The Python package will not need any JS runtime because it includes a prebuilt Vue frontend in it. For development (Vite and FastAPI auto reloads) and building the package one of npm, deno or bun is required (npm recommended due to bugs in deno and bun).
+Create or patch a FastAPI + Vue project with an integrated dev/build workflow.
 
-## Features
+- Development: one command runs Vite + FastAPI (reloads)
+- Production: `uv build` bakes the built Vue assets into the Python package (no Node/JS runtime needed to *run* the installed package)
 
-- **No JavaScript**: Your Python package can be installed and used without any JS runtime
+## Quick start
-- **Integrated build system**: Vue frontend builds into Python package during `uv build`
-- **Development server**: Single command runs Vite + FastAPI with hot-reload
-- **Optimized static serving**: Caching, zstd compression and SPA support
 
-## Installation
+Install [UV](https://docs.astral.sh/uv/) and any JS runtime (node, deno, or bun).
 
-Install [uv](https://docs.astral.sh/uv/getting-started/installation/), then:
+This README uses `my-app` as the example project name:
 
-```sh
+- project directory: `my-app/`
-uv tool install fastapi-vue-setup
+- Python module: `my_app`
-fastapi-vue-setup --help
+- env prefix: `MY_APP`
-```
+- CLI command: `my-app`
 
-Or run directly:
+Create a new project in `./my-app`:
 
 ```sh
 uvx fastapi-vue-setup my-app
 ```
 
-## Usage
+Once in your source tree, you will typically use `.` for the path. If there is an existing project, `fastapi-vue-setup` will do its best to find and patch a backend module and create or patch a Vue project in `frontend/`. The integration can be upgraded by running a new version of `fastapi-vue-setup` on it, preserving earlier default ports and user customizations.
 
-The script may be used to create an all new project folder or to patch or update an existing project to use this framework. It autodetects the project folder given and performs the appropriate actions.
+## In your project
 
-### Create a new project
+ℹ️ Everything below is meant to be run within your project source tree.
+
+The setup creates a CLI entry for your package, so that it becomes a command to run, not a Python module nor `fastapi myapp...`. The CLI main can be customized, although --listen should be kept for devserver compatibility.
+
+You can choose the JS runtime with environment `JS_RUNTIME` (e.g. `node`, `deno`, `bun`, or path to one). This is used by the build and the devserver scripts. By default any available runtime on the system is chosen.
+
+### Development server (Vite + FastAPI)
 
 ```sh
-fastapi-vue-setup my-app
+uv run scripts/devserver.py [args]
 ```
 
-This will:
+ℹ️ Arguments are forwarded to the main CLI, except that `--listen` controls where Vite listens, and `--backend` is passed to main CLI as `--listen`.
 
-1. Run `uv init my-app`
+### Production
-2. Run `create-vue frontend` (interactive - choose your Vue options)
-3. Patch the project with FastAPI integration
-4. Install dependencies via `uv add`
 
-### Patch an existing project
+Build the Python package (this compiles the Vue frontend) and run the production server:
-
-You should have your pyproject.toml at the current working directory, and Vue with its package.json under `frontend/`. If either one is missing, new applications will be initialised. Otherwise we only patch what can be patched without breaking your existing projects.
 
 ```sh
-fastapi-vue-setup .
+uv build && uv run my-app [args]
 ```
 
-### CLI Options
+Once happy with it, publish the package
 
-```
+```sh
-fastapi-vue-setup [project-dir] [options]
+uv build && uv publish
-
-Options:
-  --module-name NAME    Python module name (auto-detected from pyproject.toml)
-  --dry-run             Preview changes without modifying files
 ```
 
-## Port Configuration
+Afterwards, you can easily run it anywhere, no JS runtimes required:
 
-The tool uses three distinct ports:
+```sh
-
+uvx my-app [args]
-| Port | Purpose               | Used by                        |
-| ---- | --------------------- | ------------------------------ |
-| 5173 | Vite dev server (HMR) | `npm run dev` via devserver.py |
-| 5180 | FastAPI in dev mode   | uvicorn via devserver.py       |
-| 5080 | Production server     | `uv run my-app`                |
-
-In development, you access the app at `http://localhost:5173`. Vite proxies `/api/*` requests to FastAPI at port 5180. Configurable with `devserver.py` arguments.
-
-In production, FastAPI serves both the API and static files at `http://localhost:5080`. Configurable by `host:port` argument.
-
-## Vite Plugin Configuration
-
-The `fastapiVue()` plugin in `vite.config.ts` accepts options to customize proxy behavior:
-
-```js
-import fastapiVue from "./vite-plugin-fastapi.js";
-
-export default defineConfig({
-  plugins: [
-    vue(),
-    // Default: proxies only /api
-    fastapiVue(),
-
-    // Or specify custom paths to proxy to backend
-    fastapiVue({ paths: ["/api", "/auth", "/ws"] }),
-  ],
-});
 ```
 
-The plugin reads environment `FASTAPI_VUE_BACKEND_URL` (default: `http://localhost:5180`) to determine where to proxy requests. This is set automatically by `devserver.py`.
+ℹ️ Instead of `uvx` you may consider `uv tool install`, oldskool `pip install` or whatever best suits you.
 
-## Project Structure
+### Vite plugin
 
-After patching, your project will have:
+The generated Vite plugin lives in `frontend/vite-plugin-fastapi.js` and defaults to proxying `/api`.
+
+It reads `MY_APP_BACKEND_URL` to know where to proxy; if unset it falls back to your configured default backend port.
+
+## Project layout (typical)
 
 ```
 my-app/
-├── frontend/                # Vue application
+├── frontend/                    # Vue app (Vite)
 │   ├── src/
-│   ├── vite.config.ts       # Builds to ../my_app/frontend-build
+│   ├── vite-plugin-fastapi.js
 │   └── package.json
 ├── my_app/                      # Python package
-│   ├── __init__.py
+│   ├── __main__.py              # CLI entrypoint
-│   ├── __main__.py          # CLI entry point
+│   ├── app.py                   # FastAPI app
-│   ├── app.py               # FastAPI application
+│   └── frontend-build/          # built assets (included in distributions)
-│   └── frontend-build/      # Built frontend (gitignored)
+├── pyproject.toml
-├── scripts/
+└── scripts/
-│   ├── devserver.py         # Development server
+    ├── devserver.py             # Run Vite and FastAPI together in dev mode
-│   └── fastapi-vue/         # Build utilities
+    └── fastapi-vue/             # Dev utilities (only on the source tree)
-└── pyproject.toml           # Project configuration
+        ├── buildhook.py
+        ├── buildutil.py
+        └── devutil.py
 ```
 
-The script finds your existing fastpi app module and other files and patches them with minimal changes to enable the Vue-FastAPI interconnection.
+## The fastapi-vue runtime module
 
-## Development Workflow
+The backend runs the FastAPI app and serves the frontend build using the companion package in [fastapi-vue/README.md](fastapi-vue/README.md). Your project will depend on Fastapi and this lightweight module.
 
-```bash
+ℹ️ Development functionality is in `scripts/fastapi-vue/` directly in your source tree, and is not to be confused with this runtime module. Only the runtime is installed with your package.
-# Start dev server (runs both Vite and FastAPI)
-uv run scripts/devserver.py
-
-# Build for production
-uv build
-
-# Run production server
-uv run my-app
-```
-
-# Frontend serving
-
-Your FastAPI app will use [fastapi-vue](https://git.zi.fi/LeoVasanko/fastapi-vue) to serve the frontend file. Refer to that package's documentation for further configuration.

fastapi-vue/README.md

@@ -0,0 +1,68 @@
+# fastapi-vue
+
+Runtime helpers for FastAPI + Vite/Vue projects.
+
+## Overview
+
+This package provides:
+
+- `fastapi_vue.Frontend`: serves built SPA assets (with SPA support, caching, and optional zstd)
+- `fastapi_vue.server.run`: a small Uvicorn runner with convenient `listen` endpoint parsing
+
+## Quickstart
+
+Serve built frontend assets from `frontend-build/`:
+
+```python
+from pathlib import Path
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from fastapi_vue import Frontend
+
+frontend = Frontend(Path(__file__).with_name("frontend-build"), spa=True)
+
+@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, "/")
+```
+
+## Frontend
+
+`Frontend` serves a directory with:
+
+- RAM caching, with zstd compression when smaller than original
+- Browser caching: ETag + Last-Modified, Immutable assets
+- Favicon mapping (serve PNG or other images there instead)
+- SPA routing (serve browsers index.html at all paths not otherwise handled)
+
+Dev-mode behavior with `FastAPI(debug=True)`: requests error HTTP 409 with a message telling you to use the Vite dev server instead. Avoids accidentally using outdated `frontend-build` during development.
+
+- `directory`: Path on local filesystem
+- `index`: Index file name (default: `index.html`)
+- `spa`: Serve index at any path (default: `False`)
+- `catch_all`: Register a single catch-all handler instead of a route to each file; default for SPA
+- `cached`: Path prefixes treated as immutable (default: `/assets/`)
+- `favicon`: Optional path or glob (e.g. `/assets/logo*.png`)
+- `zstdlevel`: Compression level (default: 18)
+
+ℹ️ Even when your page has a meta tag giving favicon location, browsers still try loading `/favicon.ico` whenever looking at something else. We find it more convenient to simply serve the image where the browser expects it, with correct MIME type. This also allows having a default favicon for your application that can be easily overriden at the reverse proxy (Caddy, Nginx) to serve the company branding if needed in deployment.
+
+## Server runner
+
+When you need more flexibility than `fastapi` CLI can provide (e.g. CLI arguments to your own program), you may use this convenience to run FastAPI app with Uvicorn startup on given `listen` endpoints. Runs in the same process if possible but delegates to `uvicorn.run()` for auto-reloads and multiple workers. This would typically be called from your CLI main, which can set its own env variables to pass information to the FastAPI instances that run (Python imports only work in same-process mode).
+
+```python
+from fastapi_vue import server
+
+server.run("my_app.app:app", listen=["localhost:8000"])
+```
+
+- As a deployment option, environment `FORWARDED_ALLOW_IPS` controls `X-Forwarded` trusted IPs (default: `127.0.0.1,::1`).

fastapi-vue/fastapi_vue/__init__.py

@@ -0,0 +1,5 @@
+"""FastAPI Vue integration - serve Vue frontend from FastAPI."""
+
+from .staticfiles import Frontend
+
+__all__ = ["Frontend"]

fastapi-vue/fastapi_vue/accesslog.py

@@ -0,0 +1,418 @@
+"""HTTP/WebSocket access logging ASGI middleware."""
+
+from __future__ import annotations
+
+import http
+import itertools
+import logging
+import time
+from ipaddress import IPv6Address
+from typing import TYPE_CHECKING, cast
+
+from .termwidth import pad_display
+
+if TYPE_CHECKING:
+    from uvicorn._types import (
+        ASGI3Application,
+        ASGIReceiveCallable,
+        ASGIReceiveEvent,
+        ASGISendCallable,
+        ASGISendEvent,
+        Scope,
+        WWWScope,
+    )
+
+logger = logging.getLogger("fastapi_vue.access")
+
+# Terminal color codes
+_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)
+_WS_CLOSE = "\033[38;5;142m"  # WebSocket disconnect (dimmer yellow)
+
+
+def _format_duration(duration: float) -> str:
+    ms = int(duration * 1000)
+    if ms < 2000:
+        return f"{ms}ms"
+
+    total_seconds = ms // 1000
+    if total_seconds < 60:
+        return f"{total_seconds}s"
+
+    if total_seconds < 3600:
+        minutes, seconds = divmod(total_seconds, 60)
+        return f"{minutes}m{seconds}s"
+
+    hours, remainder = divmod(total_seconds, 3600)
+    minutes = remainder // 60
+    return f"{hours}h{minutes}m"
+
+
+def _status_color(status: int) -> str:
+    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 _METHOD_READ if method in ("GET", "HEAD", "OPTIONS") else _METHOD_WRITE
+
+
+def _format_extra_timing(extra: str = "", duration: float | None = None) -> tuple[str, str]:
+    timing = _format_duration(duration) if duration is not None else ""
+    return (f"{extra} " if extra else "", f"{_TIMING}{timing}{_RESET}" if timing else "")
+
+
+def _format_ipv6_network(ip: str) -> str:
+    try:
+        ip = ip.strip("[]")
+        if "%" in ip:
+            ip = ip.split("%")[0]
+        addr = IPv6Address(ip)
+
+        if addr.is_loopback:
+            return "::1"
+        if addr.is_unspecified:
+            return "::"
+        if addr.ipv4_mapped:
+            return str(addr.ipv4_mapped)
+        if addr.is_link_local:
+            return str(addr)
+
+        network_int = int(addr) >> 64
+        groups: list[str] = []
+        for _ in range(4):
+            groups.insert(0, format(network_int & 0xFFFF, "x"))
+            network_int >>= 16
+        result = ":".join(groups) + "::"
+        return str(IPv6Address(result + "0")).removesuffix("::")
+    except ValueError:
+        return ip
+
+
+def _format_client_ip(ip: str) -> str:
+    if not ip or ip == "-":
+        return "-"
+    stripped = ip.strip("[]")
+    if ":" in stripped:
+        return _format_ipv6_network(ip)
+    return ip
+
+
+def _header(scope: WWWScope, name: str) -> str | None:
+    name_bytes = name.lower().encode("latin-1")
+    for key, value in scope["headers"]:
+        if key.lower() == name_bytes:
+            return value.decode("latin-1")
+    return None
+
+
+def _client_host(scope: WWWScope) -> str:
+    client = scope["client"]
+    return client[0] if client else "-"
+
+
+def _path(scope: WWWScope) -> str:
+    path = scope["path"]
+    query = scope["query_string"]
+    if query:
+        return f"{path}?{query.decode('latin-1')}"
+    return path
+
+
+# WebSocket connection counter (mod 100)
+_ws_counter = itertools.count()
+
+
+def _next_ws_id() -> str:
+    return f"{next(_ws_counter) % 100:02d}"
+
+
+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 _http_access_log_extra(
+    scope: WWWScope,
+    status: int,
+    duration: float,
+    extra: str = "",
+    method: str | None = None,
+) -> dict[str, object]:
+    client_addr = _client_host(scope)
+    full_path = _path(scope)
+    method = method if method is not None else cast("str", scope.get("method", "-"))
+    method = cast("str", scope.get("state", {}).get("access_log_method") or method)
+
+    try:
+        status_phrase = http.HTTPStatus(status).phrase
+    except ValueError:
+        status_phrase = ""
+
+    extra, timing = _format_extra_timing(extra, duration)
+
+    return {
+        "client": _format_client_ip(client_addr).ljust(19),
+        "status": f"{_status_color(status)}{str(status).rjust(3)}{_RESET}",
+        "method": (
+            f"{_METHOD_READ}{pad_display('🔌', 7)}{_RESET}"
+            if method == "🔌"
+            else f"{_method_color(method)}{pad_display(method, 7)}{_RESET}"
+        ),
+        "host": f"{_HOST}{_header(scope, 'host') or '-'}{_RESET}",
+        "path": f"{_PATH}{full_path}{_RESET}",
+        "extra": extra,
+        "timing": timing,
+        "client_addr": client_addr,
+        "status_code": f"{status} {status_phrase}",
+        "request_line": f"{method} {full_path} HTTP/{scope.get('http_version', '-')}",
+        "http_version": scope.get("http_version", "-"),
+        "full_path": full_path,
+    }
+
+
+def _ws_open_extra(
+    scope: WWWScope,
+    ws_id: str,
+    origin: str | None,
+    extra: str = "",
+) -> dict[str, object]:
+    client_addr = _client_host(scope)
+    path = scope.get("path", "")
+    full_path = _path(scope)
+
+    origin_host = origin.split("://", 1)[-1] if origin else None
+    extra, timing = _format_extra_timing(extra)
+
+    host = _header(scope, "host")
+    path = f"{_PATH}{path}{_RESET}"
+    if origin_host and origin_host != host:
+        path += f" {_RESET}from {_HOST}{origin_host}{_RESET}"
+    return {
+        "client": _format_client_ip(client_addr).ljust(19),
+        "status": f"{_WS_OPEN} {ws_id}{_RESET}",
+        "method": f"{_METHOD_READ}{pad_display('🔌', 7)}{_RESET}",
+        "host": f"{_HOST}{host}{_RESET}" if host else "",
+        "path": path,
+        "extra": extra,
+        "timing": timing,
+        "client_addr": client_addr,
+        "status_code": "",
+        "request_line": f"WebSocket {path}",
+        "http_version": scope.get("http_version", "-"),
+        "full_path": full_path,
+    }
+
+
+def _ws_close_extra(
+    scope: WWWScope,
+    ws_id: str,
+    close_code: int | None,
+    duration: float,
+    extra: str = "",
+) -> dict[str, object]:
+    client_addr = _client_host(scope)
+    path = scope.get("path", "-")
+    full_path = _path(scope)
+
+    if close_code is None:
+        code, status_text = "----", "unknown"
+    else:
+        code = str(close_code)
+        status_text = WS_CLOSE_CODES.get(close_code, f"code {close_code}")
+
+    extra, timing = _format_extra_timing(extra, duration)
+
+    return {
+        "client": " " * 19,
+        "status": f"{_WS_CLOSE} {ws_id}{_RESET}",
+        "method": f"{_TIMING}{pad_display('closed', 7)}{_RESET}",
+        "host": "",
+        "path": f"{code} {status_text}",
+        "extra": extra,
+        "timing": timing,
+        "client_addr": client_addr,
+        "status_code": f"{code} {status_text}",
+        "request_line": f"WebSocket {path}",
+        "http_version": scope.get("http_version", "-"),
+        "full_path": full_path,
+    }
+
+
+def _ws_reject_extra(
+    scope: WWWScope,
+    close_code: int | None,
+    duration: float,
+    extra: str = "",
+) -> dict[str, object]:
+    """Open-format line for a connection closed before accept.
+
+    Replaces the open line (which never happened), so the client IP, host and
+    path stay visible.  No connection id is printed (ids are only assigned on
+    accept); the status column shows a dim ``--`` and the close reason rides
+    in the extra column.
+    """
+    if close_code is None:
+        code, status_text = "----", "unknown"
+    else:
+        code = str(close_code)
+        status_text = WS_CLOSE_CODES.get(close_code, f"code {close_code}")
+
+    fields = _ws_open_extra(scope, "--", _header(scope, "origin"))
+    reason = f"closed {code} {status_text}"
+    extra, timing = _format_extra_timing(f"{reason} {extra}" if extra else reason, duration)
+    fields.update(
+        {
+            "status": f"{_WS_CLOSE} --{_RESET}",
+            "extra": extra,
+            "timing": timing,
+            "status_code": f"{code} {status_text}",
+        }
+    )
+    return fields
+
+
+def _assemble_access_log(fields: dict[str, object]) -> str:
+    return (
+        f"{fields['client']} {fields['status']} {fields['method']}"
+        f"{fields['host']}{fields['path']}{fields['extra']}{fields['timing']}"
+    )
+
+
+class AccessLogMiddleware:
+    """ASGI middleware logging HTTP and WebSocket access with colored fields."""
+
+    def __init__(self, app: ASGI3Application) -> None:
+        """Store the wrapped app."""
+        self.app = app
+
+    async def __call__(
+        self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    ) -> None:
+        """Dispatch by scope type to HTTP/WebSocket access logging."""
+        if scope["type"] == "http":
+            return await self._handle_http(scope, receive, send)
+        if scope["type"] == "websocket":
+            return await self._handle_websocket(scope, receive, send)
+        return await self.app(scope, receive, send)
+
+    async def _handle_http(
+        self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    ) -> None:
+        start = time.perf_counter()
+        www_scope = cast("WWWScope", scope)
+
+        async def wrapped_send(message: ASGISendEvent) -> None:
+            if message["type"] == "http.response.start":
+                fields = _http_access_log_extra(
+                    www_scope,
+                    status=message["status"],
+                    duration=time.perf_counter() - start,
+                    extra=www_scope.get("state", {}).get("log_extra", ""),
+                )
+                logger.info(
+                    '%s - "%s" %s',
+                    fields["client_addr"],
+                    fields["request_line"],
+                    fields["status_code"],
+                    extra=fields,
+                )
+            await send(message)
+
+        return await self.app(scope, receive, wrapped_send)
+
+    async def _handle_websocket(
+        self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    ) -> None:
+        start = time.perf_counter()
+        ws_id: str | None = None  # assigned on accept; rejects print "--"
+        accepted = False
+        closed = False
+
+        www_scope = cast("WWWScope", scope)
+        origin = _header(www_scope, "origin")
+
+        def _extra() -> str:
+            return www_scope.get("state", {}).get("log_extra", "")
+
+        def _close_fields(message: ASGIReceiveEvent | ASGISendEvent) -> dict[str, object]:
+            if accepted:
+                assert ws_id is not None  # noqa: S101  # guaranteed once accepted
+                return _ws_close_extra(
+                    www_scope,
+                    ws_id,
+                    message.get("code"),
+                    time.perf_counter() - start,
+                    _extra(),
+                )
+            return _ws_reject_extra(
+                www_scope,
+                message.get("code"),
+                time.perf_counter() - start,
+                _extra(),
+            )
+
+        async def wrapped_send(message: ASGISendEvent) -> None:
+            nonlocal accepted, closed, ws_id
+            if message["type"] == "websocket.accept" and not accepted:
+                accepted = True
+                ws_id = _next_ws_id()
+                fields = _ws_open_extra(www_scope, ws_id, origin, _extra())
+                logger.info(_assemble_access_log(fields), extra=fields)
+            elif message["type"] == "websocket.http.response.start" and not closed:
+                closed = True
+                fields = _http_access_log_extra(
+                    www_scope,
+                    status=message["status"],
+                    duration=time.perf_counter() - start,
+                    extra=_extra(),
+                    method="🔌",
+                )
+                logger.info(_assemble_access_log(fields), extra=fields)
+            elif message["type"] == "websocket.close" and not closed:
+                closed = True
+                fields = _close_fields(message)
+                logger.info(_assemble_access_log(fields), extra=fields)
+            await send(message)
+
+        async def wrapped_receive() -> ASGIReceiveEvent:
+            nonlocal closed
+            message = await receive()
+            if message["type"] == "websocket.disconnect" and not closed:
+                closed = True
+                fields = _close_fields(message)
+                logger.info(_assemble_access_log(fields), extra=fields)
+            return message
+
+        return await self.app(scope, wrapped_receive, wrapped_send)

fastapi-vue/fastapi_vue/hostutil.py

@@ -0,0 +1,119 @@
+"""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.
+
+    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).
+        Two entries may be returned for IPv4 and IPv6 (all interaces).
+
+    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)}]
+
+    # Try specialized parsers in order
+    result = _parse_all_interfaces(value)
+    if result is not None:
+        return result
+
+    result = _parse_unix_socket(value)
+    if result is not None:
+        return result
+
+    result = _parse_unbracketed_ipv6(value, default_port)
+    if result is not None:
+        return result
+
+    # 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,
+) -> 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}"]
+    elif isinstance(listen, str):
+        listen = [listen]
+    return [ep for s in listen for ep in parse_endpoint(s, default_port)]

fastapi-vue/fastapi_vue/logging.py

@@ -0,0 +1,405 @@
+"""Logging integration: tracerite loading and colored access log formatting.
+
+The access log middleware supplies colored fields (``client``, ``status``,
+``method``, ``host``, ``path``, ``extra``, ``timing``) via ``extra=``.  When
+colors are disabled the ANSI escape codes are stripped from the assembled
+output so the same formatting code path produces plain text.
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+import os
+import re
+import sys
+from contextlib import suppress
+from copy import deepcopy
+from typing import TYPE_CHECKING, Literal
+
+import tracerite
+from starlette.middleware.errors import ServerErrorMiddleware
+from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse, Response
+from uvicorn.config import Config
+from uvicorn.lifespan.on import LifespanOn
+
+if TYPE_CHECKING:
+    from typing import Any
+
+    from starlette.requests import Request
+    from uvicorn._types import LifespanScope
+    from uvicorn.lifespan.on import LifespanSendMessage
+
+from .accesslog import AccessLogMiddleware
+
+ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m")
+
+ACCESS_LOG_FMT = "%(client)s %(status)s %(method)s %(host)s%(path)s %(extra)s%(timing)s"
+
+ACCESS_LOGGER = "fastapi_vue.access"
+
+
+def strip_ansi(text: str) -> str:
+    """Remove ANSI escape codes from text."""
+    return ANSI_ESCAPE_RE.sub("", text)
+
+
+def use_color(stream: io.TextIOBase = sys.stderr) -> bool:
+    """Test if the stream supports color codes."""
+    if os.environ.get("NO_COLOR"):  # Non empty means no (no-color.org)
+        return False
+    if os.environ.get("FORCE_COLOR", "") not in {"", "0"}:  # force-color.org, node
+        return True
+    if hasattr(stream, "isatty") and stream.isatty():
+        return True
+    with suppress(KeyError, ValueError, OSError):  # Journald does color (-ocat)
+        dev, ino = map(int, os.environ["JOURNAL_STREAM"].split(":", 1))
+        st = os.fstat(stream.fileno())
+        return st.st_dev == dev and st.st_ino == ino
+    return False
+
+
+_LEVEL_EMOJI = {
+    logging.DEBUG: "🐛",
+    logging.INFO: "🔷",
+    logging.WARNING: "❗",
+    logging.ERROR: "🛑",
+    logging.CRITICAL: "🚨",
+}
+
+
+def _level_prefix(record: logging.LogRecord) -> str:
+    emoji = _LEVEL_EMOJI.get(record.levelno)
+    return f"{emoji} " if emoji else f"{record.levelname}: "
+
+
+class Formatter(logging.Formatter):
+    """Formatter for both access records and ordinary log messages.
+
+    Records with the middleware's access fields (``client`` etc.) are
+    formatted from those; anything else gets an emoji level prefix
+    (``LEVEL: `` fallback for unknown levels) in place of uvicorn's
+    ``levelprefix``.
+
+    Instantiation always loads tracerite, and with ``access=True`` also
+    installs the access-log middleware: ``dictConfig`` builds formatters while
+    uvicorn applies ``log_config``, which happens before the app is loaded —
+    including in reload/worker subprocesses that re-import the config without
+    calling ``fastapi_vue.server.run()`` again.  Patching the server error
+    middleware here likewise propagates it to those subprocesses.
+    """
+
+    def __init__(
+        self,
+        fmt: str | None = None,
+        datefmt: str | None = None,
+        style: Literal["%", "{", "$"] = "%",
+        use_colors: bool | None = None,  # noqa: FBT001  # mirrors logging.Formatter
+        *,
+        access: bool = False,
+    ) -> None:
+        """Load tracerite, optionally install the access log, detect color support."""
+        tracerite.load()
+        tracerite.load_suppressions(
+            extra={"starlette.routing": "until", "fastapi.routing": "until"}
+        )
+        patch_lifespan_logging()
+        patch_server_error_middleware()
+        if access:
+            install_access_log()
+        if use_colors in (True, False):
+            self.use_colors = use_colors
+        else:
+            self.use_colors = use_color(sys.stdout)
+        super().__init__(fmt=fmt, datefmt=datefmt, style=style)
+
+    def formatMessage(self, record: logging.LogRecord) -> str:  # noqa: N802
+        """Format access records via middleware fields, others with an emoji prefix."""
+        if "client" not in record.__dict__:
+            return _level_prefix(record) + record.getMessage()
+        formatted = super().formatMessage(record)
+        if not self.use_colors:
+            formatted = strip_ansi(formatted)
+        return formatted
+
+
+class WebSocketChatterFilter(logging.Filter):
+    """Drop stock uvicorn WebSocket handshake/chatter records.
+
+    Stock uvicorn logs WS handshakes (``'%s - "WebSocket %s" ...'``) and the
+    websockets library's "connection open/closed" chatter to ``uvicorn.error``,
+    ungated by ``access_log``.  Our middleware logs WebSockets itself.
+    """
+
+    _PREFIXES = ('%s - "WebSocket ', "connection open", "connection closed", "connection rejected")
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        """Keep records not matching stock WebSocket chatter prefixes."""
+        msg = record.msg
+        if not isinstance(msg, str):
+            return True
+        return not msg.startswith(self._PREFIXES)
+
+
+class UvicornQuietFilter(logging.Filter):
+    """Silence uvicorn's routine chatter (startup/shutdown lines, etc.).
+
+    Handler-side, not a logger level: uvicorn's ``configure_logging``
+    re-applies ``log_level`` to its loggers after ``dictConfig``, which would
+    override a level lifted in the config dict.
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        """Drop uvicorn records below WARNING."""
+        return not (record.name.startswith("uvicorn") and record.levelno < logging.WARNING)
+
+
+_installed = False
+
+
+def install_access_log() -> None:
+    """Wrap apps loaded by uvicorn with AccessLogMiddleware, once per process.
+
+    The guard is deliberately module-level: reload/worker subprocesses
+    re-import this module, resetting it so the patch is re-applied there.
+    """
+    global _installed  # noqa: PLW0603  # deliberately module-level, see docstring
+    if _installed:
+        return
+    _installed = True
+
+    original_load = Config.load
+
+    def load(self):  # noqa: ANN001, ANN202
+        original_load(self)
+        if not isinstance(self.loaded_app, AccessLogMiddleware):
+            self.loaded_app = AccessLogMiddleware(self.loaded_app)
+
+    Config.load = load  # type: ignore[method-assign]
+
+
+_lifespan_patched = False
+
+
+def patch_lifespan_logging() -> None:
+    """Patch uvicorn's LifespanOn to log lifespan failures with exc_info.
+
+    Starlette formats lifespan exceptions into a plain-text ASGI message,
+    which uvicorn logs as-is without exc_info, while the exc_info-carrying
+    log in ``LifespanOn.main()`` is skipped when a failure message was sent.
+    This suppresses the text message and always logs the exception with
+    exc_info, so tracerite (or any exc_info-aware handler) renders the
+    traceback. Monkeypatches uvicorn internals; written against uvicorn 0.52.
+    """
+    global _lifespan_patched  # noqa: PLW0603  # once-per-process, resets in subprocesses
+    if _lifespan_patched:
+        return
+    _lifespan_patched = True
+
+    original_send = LifespanOn.send
+
+    async def send(self: LifespanOn, message: LifespanSendMessage) -> None:
+        # Drop the pre-formatted traceback text; main() logs the exception itself.
+        if message["type"] in ("lifespan.startup.failed", "lifespan.shutdown.failed"):
+            message = dict(message)  # type: ignore[assignment]
+            message.pop("message", None)
+        await original_send(self, message)
+
+    async def main(self: LifespanOn) -> None:
+        """Mirror upstream LifespanOn.main, but always log failures with exc_info."""
+        try:
+            app = self.config.loaded_app
+            scope: LifespanScope = {
+                "type": "lifespan",
+                "asgi": {"version": self.config.asgi_version, "spec_version": "2.0"},
+                "state": self.state,
+            }
+            await app(scope, self.receive, self.send)
+        except BaseException:
+            self.asgi = None
+            self.error_occurred = True
+            if self.startup_failed or self.shutdown_failed or self.config.lifespan != "auto":
+                phase = "shutdown" if self.shutdown_failed else "startup"
+                self.logger.exception("Uncaught exception during application %s", phase)
+            else:
+                self.logger.info("ASGI 'lifespan' protocol appears unsupported.")
+        finally:
+            self.startup_event.set()
+            self.shutdown_event.set()
+
+    LifespanOn.send = send  # type: ignore[method-assign]
+    LifespanOn.main = main  # type: ignore[method-assign]
+
+
+_server_error_patched = False
+
+DEBUG_INGRESS = """This page is shown for your guidance because the application is \
+running in debug mode and has crashed handling this request."""
+
+
+def _generate_html(exc: Exception) -> str:
+    return tracerite.html_page(
+        exc,
+        title="FastAPI debugger",
+        heading="500 Server Error",
+        ingress=DEBUG_INGRESS,
+    )
+
+
+def _generate_plain_text(exc: Exception) -> str:
+    buffer = io.StringIO()
+    tracerite.tty_traceback(exc, file=buffer)
+    return buffer.getvalue()
+
+
+def _generate_json(exc: Exception) -> dict[str, Any]:
+    chain = tracerite.extract_chain(exc)
+    return {"detail": "Internal Server Error", "traceback": chain}
+
+
+def patch_server_error_middleware() -> None:
+    """Patch Starlette's ServerErrorMiddleware to format debug errors with tracerite.
+
+    Starlette's debug responses use its own static HTML traceback template.
+    This replaces ``debug_response`` with tracerite renderers (source
+    context, locals, chained exceptions), adds ``accept: application/json``
+    handling, and returns a JSON body also for non-debug errors when
+    requested. Only apps running with ``debug=True`` produce traceback
+    responses. Monkeypatches Starlette internals; written against
+    starlette 1.6.
+    """
+    global _server_error_patched  # noqa: PLW0603  # once-per-process, resets in subprocesses
+    if _server_error_patched:
+        return
+    _server_error_patched = True
+
+    def debug_response(
+        self: ServerErrorMiddleware,  # noqa: ARG001
+        request: Request,
+        exc: Exception,
+    ) -> Response:
+        accept = request.headers.get("accept", "")
+        if "text/html" in accept:
+            return HTMLResponse(_generate_html(exc), status_code=500)
+        if "application/json" in accept:
+            return JSONResponse(_generate_json(exc), status_code=500)
+        return PlainTextResponse(_generate_plain_text(exc), status_code=500)
+
+    def error_response(
+        self: ServerErrorMiddleware,  # noqa: ARG001
+        request: Request,
+        exc: Exception,  # noqa: ARG001  # signature mirrors Starlette's
+    ) -> Response:
+        if "application/json" in request.headers.get("accept", ""):
+            return JSONResponse({"detail": "Internal Server Error"}, status_code=500)
+        return PlainTextResponse("Internal Server Error", status_code=500)
+
+    ServerErrorMiddleware.debug_response = debug_response  # type: ignore[method-assign]
+    ServerErrorMiddleware.error_response = error_response  # type: ignore[method-assign]
+
+
+def patch_log_config(log_config, *, access_log: bool = True):  # noqa: ANN001, ANN201
+    """Patch a uvicorn log_config dict for our logging, best-effort.
+
+    Users presumably base their config on uvicorn's default dict, but any
+    shape is tolerated: pieces that do not fit the config's structure are
+    silently skipped.  Non-dict configs (e.g. an ini file path) pass through
+    untouched.
+
+    Always adds an unreferenced NullHandler whose Formatter instantiation
+    loads tracerite in every process uvicorn applies the config in, filters
+    on the default handler dropping stock uvicorn's WebSocket chatter and
+    routine INFO lines, an emoji-level-prefix Formatter in place of
+    uvicorn's stock ``default`` formatter (a user-supplied one wins), a root
+    logger entry so ``logging.info()`` et al. print through the default
+    handler, and a no-prefix ``kanta`` logger entry (likewise).  The
+    ``watchfiles.main`` logger is lifted to WARNING so its INFO "N changes
+    detected" line is dropped while the WARNING "Reloading..." line (logged
+    to ``uvicorn.error``) still shows; a user-supplied level wins.
+    With ``access_log``, additionally rewires the ``access`` formatter to
+    our Formatter and attaches its handler to our ``fastapi_vue.access``
+    logger.  We must not
+    attach handlers to ``uvicorn.access``: uvicorn gates its own
+    protocol-level access logging on ``uvicorn.access.hasHandlers()``.
+    """
+    if not isinstance(log_config, dict):
+        return log_config
+    config = deepcopy(log_config)
+
+    with suppress(Exception):
+        config["formatters"]["fastapi_vue"] = {"()": "fastapi_vue.logging.Formatter"}
+        config["handlers"]["fastapi_vue"] = {
+            "class": "logging.NullHandler",
+            "formatter": "fastapi_vue",
+        }
+
+    with suppress(Exception):
+        filters = config.setdefault("filters", {})
+        filters["ws_chatter"] = {"()": "fastapi_vue.logging.WebSocketChatterFilter"}
+        filters["uvicorn_quiet"] = {"()": "fastapi_vue.logging.UvicornQuietFilter"}
+        handler_filters = config["handlers"]["default"].setdefault("filters", [])
+        for name in ("ws_chatter", "uvicorn_quiet"):
+            if name not in handler_filters:
+                handler_filters.append(name)
+
+    # Emoji level prefixes for ordinary logs, replacing uvicorn's stock
+    # default formatter; a user-supplied default formatter is left alone.
+    with suppress(Exception):
+        default = config["formatters"]["default"]
+        if default.get("()") in (None, "uvicorn.logging.DefaultFormatter"):
+            config["formatters"]["default"] = {
+                "()": "fastapi_vue.logging.Formatter",
+                "fmt": "%(message)s",
+                "use_colors": None,
+            }
+
+    # uvicorn's default config leaves the root logger handlerless, eating
+    # logging.info() et al.; route root through uvicorn's default handler.
+    with suppress(Exception):
+        root = config.setdefault("root", {})
+        root.setdefault("level", "INFO")
+        root_handlers = root.setdefault("handlers", [])
+        if "default" not in root_handlers:
+            root_handlers.append("default")
+
+    # watchfiles logs "N changes detected" to its own logger at INFO; only the
+    # WARNING "Reloading..." line (uvicorn.error) should show.
+    with suppress(Exception):
+        config.setdefault("loggers", {}).setdefault("watchfiles.main", {}).setdefault(
+            "level", "WARNING"
+        )
+
+    # kanta-style output (diffs, colored headers) prints without prefixes,
+    # like our access log.  A user-supplied "kanta" logger entry wins.
+    with suppress(Exception):
+        config["formatters"].setdefault("plain", {"fmt": "%(message)s"})
+        config["handlers"].setdefault(
+            "plain",
+            {
+                "class": "logging.StreamHandler",
+                "formatter": "plain",
+                "stream": "ext://sys.stderr",
+            },
+        )
+        config.setdefault("loggers", {}).setdefault(
+            "kanta",
+            {"handlers": ["plain"], "level": "INFO", "propagate": False},
+        )
+
+    if access_log:
+        with suppress(Exception):
+            config["formatters"]["access"] = {
+                "()": "fastapi_vue.logging.Formatter",
+                "fmt": ACCESS_LOG_FMT,
+                "use_colors": None,
+                "access": True,
+            }
+        with suppress(Exception):
+            if "access" in config["handlers"]:
+                config.setdefault("loggers", {})[ACCESS_LOGGER] = {
+                    "handlers": ["access"],
+                    "level": "INFO",
+                    "propagate": False,
+                }
+
+    return config

fastapi-vue/fastapi_vue/server.py

@@ -0,0 +1,166 @@
+"""Uvicorn server runner with multi-endpoint support."""
+
+import asyncio
+import importlib.metadata
+import logging
+import os
+from contextlib import suppress
+from pathlib import Path
+from typing import Any
+
+import tracerite
+import uvicorn
+from uvicorn import Config, Server
+
+from .hostutil import parse_endpoints
+from .logging import (
+    install_access_log,
+    patch_lifespan_logging,
+    patch_log_config,
+    patch_server_error_middleware,
+    use_color,
+)
+from .startupbox import print_box
+
+tracerite.load()  # Early load on CLI load (import server); uvicorn workers reload via log config
+
+# Install force color to aid tracerite and any external software to use full color when available
+# Define NO_COLOR or FORCE_COLOR beforehand to avoid this
+if "FORCE_COLOR" not in os.environ and use_color():
+    os.environ["FORCE_COLOR"] = "3"
+
+logger = logging.getLogger(__name__)
+
+_WILDCARD_HOSTS = frozenset({"0.0.0.0", "::"})  # noqa: S104
+
+
+def _connect_url(endpoints: list[dict]) -> str:
+    """Return a URL the user can connect to for the first TCP endpoint.
+
+    Wildcard binds (0.0.0.0, ::) are shown as localhost, as that is the
+    address a user can actually open. Returns "" for unix-socket-only setups.
+    """
+    for endpoint in endpoints:
+        host = endpoint.get("host")
+        if host is None:
+            continue
+        if host in _WILDCARD_HOSTS:
+            host = "localhost"
+        elif ":" in host:  # IPv6 literal
+            host = f"[{host}]"
+        return f"http://{host}:{endpoint['port']}"
+    return ""
+
+
+def _print_startup_box(template: str, app: str, endpoints: list[dict]) -> None:
+    """Format the startup box template and print it.
+
+    Available fields: ``{module}`` (top-level package of the app path),
+    ``{name}`` (module with spaces instead of underscores), ``{Name}``
+    (also capitalized), ``{version}`` (from installed package metadata,
+    "dev" when not installed) and ``{url}``.
+    """
+    module = app.split(":", 1)[0].split(".", 1)[0]
+    name = module.replace("_", " ")
+    try:
+        version = importlib.metadata.version(module)
+    except importlib.metadata.PackageNotFoundError:
+        version = ""
+    values = {
+        "module": module,
+        "name": name,
+        "Name": name.title(),
+        "version": version,
+        "url": _connect_url(endpoints),
+    }
+    print_box(template.format_map(values))
+
+
+def run(  # noqa: PLR0913
+    app: str,
+    *,
+    listen: str | list[str] | None = None,
+    default_port: int = 8000,
+    reload: bool | Path = False,
+    workers: int | None = None,
+    access_log: bool = True,
+    startup_box: str | None = "{Name} {version}\n{url}",
+    log_config: Any = uvicorn.config.LOGGING_CONFIG,  # noqa: ANN401
+    **uvicorn_config: Any,  # noqa: ANN401
+) -> None:
+    """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. If a Path is given, reload watches that
+            directory. True enables reload without setting a reload directory.
+            False disables reload and clears any reload_dirs.
+        workers: Number of worker processes (requires uvicorn.run, single endpoint only).
+        access_log: Enable our colored HTTP/WebSocket access logging middleware
+            (uvicorn's own access logging is always bypassed).
+        startup_box: Template for the startup box printed to stderr before
+            serving (see _print_startup_box for fields), None to not print it.
+        log_config: Logging config passed to uvicorn. Dict configs are patched
+            best-effort (see fastapi_vue.logging.patch_log_config): tracerite
+            loading and WebSocket chatter filtering are always installed, and
+            when access_log is enabled the access formatting is rewired too.
+        **uvicorn_config: Additional uvicorn config options (overrides all other settings).
+
+    """
+    endpoints = parse_endpoints(listen, default_port)
+    if not endpoints:
+        msg = "No endpoints to serve; check listen configuration"
+        raise ValueError(msg)
+
+    if startup_box:
+        _print_startup_box(startup_box, app, endpoints)
+
+    if isinstance(reload, Path):
+        uvicorn_config["reload_dirs"] = [str(reload)]
+    elif not reload:
+        uvicorn_config.pop("reload_dirs", None)
+
+    if access_log:
+        install_access_log()
+    patch_lifespan_logging()
+    patch_server_error_middleware()
+    uvicorn_config["access_log"] = False  # We always bypass uvicorn's own access logging
+    uvicorn_config["log_config"] = patch_log_config(log_config, access_log=access_log)
+
+    conf: dict[str, object] = {"app": app, "reload": bool(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, asyncio.CancelledError):
+        if reload or workers:
+            serve_multiprocess(endpoints, **conf)
+        else:
+            asyncio.run(serve(endpoints, **conf))
+
+
+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:
+        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: 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]
+        logger.warning(
+            "Current mode supports only one endpoint. Listening: %s, skipped: %s",
+            eps[0],
+            " ".join(eps[1:]),
+        )
+    uvicorn.run(**kwargs, **endpoints[0])

fastapi-vue/fastapi_vue/startupbox.py

@@ -0,0 +1,20 @@
+"""Startup banner: a rounded unicode box printed to stderr (not via logging)."""
+
+import sys
+
+from .termwidth import display_width, pad_display
+
+
+def print_box(text: str) -> None:
+    """Print text in a rounded unicode box sized to its longest line.
+
+    The text is split on newlines as-is (not trimmed). Padding accounts
+    for ANSI colors and unicode display width (see termwidth.display_width).
+    """
+    lines = text.split("\n")
+    width = max(map(display_width, lines))
+    border = "─" * (width + 2)
+    out = [f"╭{border}╮"]
+    out.extend(f"│ {pad_display(line, width)} │" for line in lines)
+    out.append(f"╰{border}╯")
+    sys.stderr.write("\n".join(out) + "\n")

fastapi-vue/fastapi_vue/staticfiles.py

@@ -0,0 +1,293 @@
+"""FastAPI static file serving with zstd compression and SPA support."""
+
+from __future__ import annotations
+
+import fnmatch
+import logging
+import mimetypes
+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
+
+logger = logging.getLogger("uvicorn.error")  # Use FastAPI logging style
+
+__all__ = ["Frontend"]
+
+
+class Assets:
+    """Default cached value to /assets/."""
+
+    @staticmethod
+    def parse(cached: str | list[str] | Assets) -> list[str]:
+        match cached:
+            case Assets():
+                return ["/assets/"]
+            case str():
+                return [cached]
+            case list():
+                return cached
+            case _:
+                msg = f"Invalid cached value: {cached!r}"
+                raise ValueError(msg)
+
+
+class Frontend:
+    """Static file server with automatic zstd compression and caching.
+
+    Features:
+    - Automatic zstd compression for compressible files
+    - ETag-based caching of immutable assets
+    - SPA (Single Page Application) support
+    - /favicon.ico with correct MIME type (image/png etc)
+    - 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 are immutable (default: "/assets/")
+        favicon: Wildcard path to favicon. E.g. /assets/logo*.png matches Vite output
+        zstdlevel: Zstd compression level (default: 18)
+
+    """
+
+    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 | 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 if cached is not None else Assets())
+        self.zstdlevel = zstdlevel
+        self.favicon = favicon
+        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) -> 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():
+            logger.error(
+                "Missing %s - no frontend (try uv build)",
+                self.base,
+            )
+        else:
+            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 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["/"] = (
+                msg.encode(),
+                None,
+                {
+                    "etag": "error",
+                    "content-type": "text/plain",
+                    "cache-control": "no-store",
+                },
+            )
+        return www
+
+    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)
+        If debug is None, uses app.debug (app passed to frontend.route)
+        """
+        if debug is None:
+            debug = getattr(self._app, "debug", False)
+        if debug:
+            # 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(
+                "%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: str = "/") -> None:
+        """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", response_model=None)(
+                self.handle
+            )
+
+    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
+
+        # 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)
+        debug = getattr(self._app, "debug", False)
+        paths = self._devmode_paths if debug else self.www.keys()
+        handler = _devmode_respond if debug 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 debug else partial(handler, name=p),
+                methods=["GET", "HEAD"],
+                name=f"frontend{p.replace('/', '_')}",
+            )
+            for p in paths
+        ]
+
+    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"},
+            )
+        return Response(content=data, headers=headers)
+
+    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)
+        files = self._devmode_paths if debug 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 debug else self._respond)(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] Use Vite devserver instead."},
+    )

fastapi-vue/fastapi_vue/termwidth.py

@@ -0,0 +1,37 @@
+"""Terminal display width calculation for unicode and ANSI-colored text."""
+
+import re
+import unicodedata
+
+ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;:]*[A-Za-z]")
+
+
+def _is_wide(char: str) -> bool:
+    """Return True for characters rendered as two terminal columns."""
+    if unicodedata.east_asian_width(char) in {"F", "W"}:
+        return True
+    cp = ord(char)
+    return unicodedata.category(char) == "So" and (
+        0x2600 <= cp <= 0x27BF or 0x1F300 <= cp <= 0x1F9FF or 0x1FA00 <= cp <= 0x1FAFF
+    )
+
+
+def display_width(text: str) -> int:
+    """Calculate the display width of a string in terminal columns.
+
+    ANSI escape codes are ignored. Wide characters (East Asian F/W and
+    emoji) count as two columns, combining marks and format characters
+    (e.g. emoji variation selectors) as zero.
+    """
+    plain = ANSI_ESCAPE_RE.sub("", text)
+    width = 0
+    for char in plain:
+        if unicodedata.category(char) in {"Mn", "Mc", "Me", "Cf"}:
+            continue
+        width += 2 if _is_wide(char) else 1
+    return width
+
+
+def pad_display(text: str, width: int) -> str:
+    """Pad text with trailing spaces to the given display width (no truncation)."""
+    return text + " " * max(width - display_width(text), 0)

fastapi-vue/pyproject.toml

@@ -0,0 +1,25 @@
+[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",
+    "tracerite>=2.6.5",
+]
+
+[project.urls]
+Homepage = "https://vasanko.com/coders/fastapi-vue"
+Repository = "https://git.zi.fi/LeoVasanko/fastapi-vue-setup"
+Issues = "https://github.com/LeoVasanko/fastapi-vue-setup"
+
+[build-system]
+requires = ["hatchling", "hatch-vcs"]
+build-backend = "hatchling.build"
+
+[tool.hatch.version]
+source = "vcs"
+raw-options.root = ".."

pyproject.toml

@@ -9,12 +9,14 @@ description = "Tool to create or patch FastAPI+Vue projects with integrated buil
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-    "tomli-w>=1.0.0",
+    "ruff>=0.14.13",
+    "tomlkit>=0.12.0",
 ]
 
 [project.urls]
-Homepage = "https://git.zi.fi/LeoVasanko/fastapi-vue-setup"
+Homepage = "https://vasanko.com/coders/fastapi-vue"
-Repository = "https://github.com/LeoVasanko/fastapi-vue-setup"
+Repository = "https://git.zi.fi/LeoVasanko/fastapi-vue-setup"
+Issues = "https://github.com/LeoVasanko/fastapi-vue-setup"
 
 [project.scripts]
 fastapi-vue-setup = "fastapi_vue_setup:main"
@@ -26,4 +28,24 @@ source = "vcs"
 include = ["fastapi_vue_setup.py", "template/**/*", "_version.py"]
 
 [dependency-groups]
-dev = ["ruff"]
+dev = ["ruff", "fastapi-vue"]
+
+[tool.uv.sources]
+fastapi-vue = { path = "fastapi-vue", editable = true }
+
+[tool.uv.workspace]
+members = [
+    "f",
+]
+
+[tool.ruff]
+line-length = 100
+
+[tool.ruff.lint]
+select = ["ALL"]
+ignore = ["CPY", "D203", "D213", "COM812", "PLR2004"]
+
+[tool.ruff.lint.per-file-ignores]
+"template/**" = ["F821"]  # Undefined names are template placeholders
+"template/scripts/devserver.py" = ["N806"]  # MODULE_NAME is a template variable
+"fastapi_vue_setup.py" = ["PLR", "C901", "T201", "RUF001"]

scripts/release.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env -S uv run
+"""Build and release both fastapi-vue and fastapi-vue-setup packages."""
+
+import shutil
+import subprocess
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+DIST = ROOT / "dist"
+FASTAPI_VUE = ROOT / "fastapi-vue"
+
+
+def main() -> None:
+    """Build both packages to dist directory."""
+    # Clear the dist directory
+    if DIST.exists():
+        shutil.rmtree(DIST)
+    DIST.mkdir()
+
+    # Build fastapi-vue (subdirectory) to root dist
+    subprocess.run(["uv", "build", "--out-dir", str(DIST)], cwd=FASTAPI_VUE, check=True)  # noqa: S603, S607
+
+    # Build fastapi-vue-setup (root)
+    subprocess.run(["uv", "build", "--out-dir", str(DIST)], cwd=ROOT, check=True)  # noqa: S603, S607
+
+
+if __name__ == "__main__":
+    main()

template/backend/__init__.py

@@ -0,0 +1 @@
+"""Backend package with FastAPI application and Vue frontend integration."""

template/backend/__main__.py

@@ -1,29 +1,32 @@
-import sys
+# auto-upgrade@fastapi-vue-setup - remove this if you modify this file
+"""Command-line entry point for running the backend server."""
 
-import uvicorn
+import argparse
+import os
+from pathlib import Path
+
+from fastapi_vue import server
+
+DEFAULT_PORT = TEMPLATE_DEFAULT_PORT
+DEVMODE = os.getenv("ENVPREFIX_DEV") == "1"
 
 
-def main():
+def main() -> None:
-    """Run the FastAPI application using uvicorn."""
+    """Run the backend server with optional arguments."""
-
+    parser = argparse.ArgumentParser(description="Run the MODULE_NAME server.")
-    if len(sys.argv) > 1:
+    parser.add_argument(
-        endpoint = sys.argv[1]
+        "-l",
-        if ":" in endpoint:
+        "--listen",
-            host, port = endpoint.rsplit(":", 1)
+        action="append",
-            host = host or "localhost"
+        help=(f"Endpoint (default: localhost:{DEFAULT_PORT})."),
-            port = int(port)
+    )
-        else:
+    args = parser.parse_args()
-            host = "localhost"
+    server.run(
-            port = int(endpoint)
+        "APP_MODULE:APP_VAR",
-    else:
+        listen=args.listen,
-        host = "localhost"
+        default_port=DEFAULT_PORT,
-        port = 5080
+        server_header=False,
-
+        reload=Path(__file__).parent if DEVMODE else False,
-    uvicorn.run(
-        "MODULE_NAME.app:app",
-        host=host,
-        port=port,
-        log_level="info",
     )
 
 

template/backend/app.py

@@ -1,23 +1,25 @@
+"""FastAPI application module with Vue frontend integration."""
+
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from pathlib import Path
 
 from fastapi import FastAPI
 from fastapi_vue import Frontend
+from MAIN_MODULE import DEVMODE
 
 # Vue Frontend static files
-frontend = Frontend(
+frontend = Frontend(Path(__file__).with_name("frontend-build"))
-    Path(__file__).with_name("frontend-build"), spa=True, cached=["/assets/"]
-)
 
 
 @asynccontextmanager
-async def lifespan(app: FastAPI):
+async def lifespan(_app: FastAPI) -> AsyncGenerator:
     """Manage app startup and shutdown resources."""
     await frontend.load()
     yield
 
 
-app = FastAPI(title="PROJECT_TITLE", lifespan=lifespan)
+app = FastAPI(title="PROJECT_TITLE", debug=DEVMODE, lifespan=lifespan)
 
 
 # Add API routes here...
@@ -25,9 +27,10 @@ app = FastAPI(title="PROJECT_TITLE", lifespan=lifespan)
 
 # Health check endpoint for the Vue demo app to verify the backend is running
 @app.get("/api/health")
-async def health_check():
+async def health_check() -> dict:
+    """Return backend status for health monitoring."""
     return {"status": "ok"}
 
 
-# Final catch-all route for frontend files (keep at end of file)
+# Serve the Vue frontend (needs to be last if SPA catch-all is used)
 frontend.route(app, "/")

template/frontend/vite-plugin-fastapi.js

@@ -1,17 +1,19 @@
 /**
  * FastAPI-Vue Vite Plugin
+ * auto-upgrade@fastapi-vue-setup -- remove this if you edit the plugin
  *
  * 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
  *
- * Environment variables (with defaults):
+ * Options:
- *   FASTAPI_VUE_BACKEND_URL=http://localhost:5180 - Backend API URL for proxying
+ *   paths - Array of paths to proxy (default: ["/api"])
  */
 
-const backendUrl = process.env.FASTAPI_VUE_BACKEND_URL || "http://localhost:5180"
-
 export default function fastapiVue({ paths = ["/api"] } = {}) {
+  const backendUrl = process.env.ENVPREFIX_BACKEND_URL || "http://localhost:TEMPLATE_DEV_PORT"
+
   // Build proxy configuration for each path
   const proxy = {}
   for (const path of paths) {
@@ -23,8 +25,9 @@ export default function fastapiVue({ paths = ["/api"] } = {}) {
   }
 
   return {
-    name: "fastapi-vite",
+    name: "vite-plugin-fastapi-MODULE_NAME",
     config: () => ({
+      clearScreen: false,
       server: { proxy },
       build: {
         outDir: "../MODULE_NAME/frontend-build",

template/scripts/devserver.py

@@ -1,301 +1,94 @@
 #!/usr/bin/env -S uv run
-"""Run Vite development server for frontend and FastAPI backend with auto-reload.
+# auto-upgrade@fastapi-vue-setup - remove this if you modify this file
-
+"""Run Vite development server for Vue app and FastAPI backend with auto-reload."""
-Usage:
-    uv run scripts/devserver.py [host:port] [--backend host:port]
-
-The optional host:port argument sets where the Vite frontend listens.
-Supported forms: host[:port], :port (all interfaces), or just port.
-The --backend option sets where the FastAPI backend listens (default: localhost:5180).
-
-Environment:
-    JS_RUNTIME          Path or name of JS runtime to use (deno, npm/node or bun).
-    FASTAPI_VUE_FRONTEND_URL    Set by this script for the backend to know where Vite is.
-"""
 
 import argparse
 import asyncio
-import contextlib
-import ipaddress
 import os
+import subprocess
+import sys
 from pathlib import Path
-from sys import stderr
-from urllib.parse import urlparse
 
-import httpx
+import tracerite
 
-exec((Path(__file__).parent / "fastapi-vue/util.py").read_text("UTF-8"))  # noqa: S102
+# Import util.py 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 (
+    ProcessGroup,
+    check_ports_free,
+    logger,
+    ready,
+    setup_cli,
+    setup_vite,
+)
 
-DEFAULT_HOST = "localhost"
+DEFAULT_VITE_PORT = TEMPLATE_VITE_PORT
-DEFAULT_VITE_PORT = 5173
+DEFAULT_DEV_PORT = TEMPLATE_DEV_PORT
-DEFAULT_BACKEND_PORT = 5180
+HEALTH = TEMPLATE_HEALTH
-FRONTEND_PATH = Path(__file__).parent.parent / "frontend"
-
-EPILOG = """
-  scripts/devserver.py                       # Default ports on localhost
-  scripts/devserver.py 3000                  # Vite on localhost:3000
-  scripts/devserver.py :3000 --backend 8000  # *:3000, localhost:8000
-"""
-
-BUN_BUG = """\
-┃ ⚠️  Bun cannot correctly proxy API requests to the backend.
-┃ Bug report: https://github.com/oven-sh/bun/issues/9882
-┃
-┃ Consider using deno or npm instead for development.
-"""
-
-
-def parse_endpoint(
-    value: str | None, default_port: int = DEFAULT_VITE_PORT
-) -> tuple[str | None, int, bool]:
-    """Parse an endpoint for Vite or backend.
-
-    Returns (host, port, all_ifaces).
-    """
-    if not value:
-        return DEFAULT_HOST, default_port, False
-
-    # Port only (numeric) -> localhost:port
-    if value.isdigit():
-        return DEFAULT_HOST, int(value), False
-
-    # Leading colon :port -> bind all interfaces
-    if value.startswith(":") and value != ":":
-        port_part = value[1:]
-        if not port_part.isdigit():
-            raise SystemExit(f"Invalid port in '{value}'")
-        return None, int(port_part), True
-
-    # Unbracketed IPv6 (cannot safely contain a port)
-    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 value, default_port, False
-
-    # Use urllib.parse for everything else
-    parsed = urlparse(f"//{value}")
-    host = parsed.hostname or DEFAULT_HOST
-    port = parsed.port or default_port
-
-    return host, port, False
-
-
-def resolve_frontend_tools(
-    vite_host: str | None, vite_port: int, all_ifaces: bool
-) -> tuple[list[str], list[str], str]:
-    """Resolve frontend install and dev commands.
-
-    Returns (install_cmd, dev_cmd, tool_name).
-    Raises SystemExit if tools are not available.
-    """
-    if not (FRONTEND_PATH / "package.json").exists():
-        stderr.write(f"┃ ⚠️  Frontend source not found at {FRONTEND_PATH}\n")
-        raise SystemExit(1)
-
-    result = find_js_runtime()  # noqa # type: ignore
-    if result is None:
-        if not os.environ.get("JS_RUNTIME"):
-            stderr.write("┃ ⚠️  deno, npm or bun needed to run the frontend server.\n")
-        raise SystemExit(1)
-
-    tool, name = result
-
-    install_args = {
-        "deno": ("install", "--quiet", "--allow-scripts=npm:vue-demi"),
-        "npm": ("install", "--silent"),
-        "bun": ("install", "--silent"),
-    }
-    dev_args = {
-        "deno": ("run", "dev", "--"),
-        "npm": ("--silent", "run", "dev", "--"),
-        "bun": ("run", "dev", "--"),
-    }
-
-    install_cmd = [tool, *install_args[name]]
-    dev_cmd = [tool, *dev_args[name], "--port", str(vite_port)]
-
-    if all_ifaces:
-        dev_cmd.append("--host")
-    elif vite_host:
-        dev_cmd.extend(["--host", vite_host])
-
-    if name == "bun":
-        stderr.write(BUN_BUG)
-
-    return install_cmd, dev_cmd, name
-
-
-async def wait_for_backend(backend_host: str, backend_port: int):
-    """Wait for the backend to be ready by polling the health endpoint."""
-    max_attempts = 50
-    async with httpx.AsyncClient() as client:
-        for attempt in range(max_attempts):
-            try:
-                await client.get(f"http://{backend_host}:{backend_port}", timeout=1.0)
-                stderr.write("✓ Backend ready!\n")
-                return True
-            except httpx.RequestError:
-                if attempt == max_attempts - 1:
-                    stderr.write("┃ ⚠️  Backend didn't start in time\n")
-                    return False
-                await asyncio.sleep(0.1)
-    return False
-
-
-async def _terminate_process(proc: asyncio.subprocess.Process, name: str) -> None:
-    """Gracefully terminate a subprocess."""
-    if proc.returncode is not None:
-        return
-    try:
-        proc.terminate()
-    except ProcessLookupError:
-        return
-    try:
-        await asyncio.wait_for(proc.wait(), timeout=2)
-    except TimeoutError:
-        try:
-            proc.kill()
-        except ProcessLookupError:
-            return
-        await proc.wait()
 
 
 async def run_devserver(
-    vite_host: str | None,
+    listen: str,
-    vite_port: int,
+    backend: str,
-    all_ifaces: bool,
+    extra_args: list[str] | None = None,
-    backend_host: str,
-    backend_port: int,
 ) -> None:
-    """Run the development server with install, backend, and frontend."""
+    """Start Vite and FastAPI dev servers with hot reload."""
-    install_cmd, dev_cmd, tool_name = resolve_frontend_tools(
+    reporoot = Path(__file__).parent.parent
-        vite_host, vite_port, all_ifaces
+    front = reporoot / "frontend"
-    )
+    if not (front / "package.json").exists():
-
+        logger.warning("Frontend source not found at %s", front)
-    # Tell the backend where the Vite dev server is
-    os.environ["FASTAPI_VUE_FRONTEND_URL"] = (
-        f"http://{vite_host or 'localhost'}:{vite_port}"
-    )
-    # Tell Vite where the backend is (for proxying /api requests)
-    os.environ["FASTAPI_VUE_BACKEND_URL"] = f"http://{backend_host}:{backend_port}"
-    cwd = str(Path(__file__).parent.parent)
-    frontend_cwd = str(FRONTEND_PATH)
-
-    backend_proc: asyncio.subprocess.Process | None = None
-    install_proc: asyncio.subprocess.Process | None = None
-    frontend_proc: asyncio.subprocess.Process | None = None
-
-    try:
-        # Start install (concurrent with backend)
-        stderr.write(f">>> {tool_name} {' '.join(install_cmd[1:])}\n")
-        install_proc = await asyncio.create_subprocess_exec(
-            *install_cmd, cwd=frontend_cwd
-        )
-
-        await asyncio.sleep(0.1)
-
-        # Start backend (concurrent with install)
-        backend_cmd = [
-            "uvicorn",
-            "MODULE_NAME.app:app",
-            "--host",
-            backend_host,
-            "--port",
-            str(backend_port),
-            "--reload",
-        ]
-        stderr.write(f">>> {' '.join(backend_cmd)}\n")
-        backend_proc = await asyncio.create_subprocess_exec(*backend_cmd, cwd=cwd)
-
-        # Wait for install to complete and backend to be ready
-        install_task = asyncio.create_task(install_proc.wait(), name="install")
-        backend_ready_task = asyncio.create_task(
-            wait_for_backend(backend_host, backend_port), name="backend_ready"
-        )
-
-        done, pending = await asyncio.wait(
-            {install_task, backend_ready_task},
-            return_when=asyncio.FIRST_COMPLETED,
-        )
-
-        for task in done:
-            if task.get_name() == "install":
-                if task.result() != 0:
-                    stderr.write("┃ ⚠️  Install failed\n")
-                    raise SystemExit(1)
-            elif task.get_name() == "backend_ready" and not task.result():
         raise SystemExit(1)
 
-        if pending:
+    viteurl, npm_install, vite = setup_vite(listen, DEFAULT_VITE_PORT)
-            done2, _ = await asyncio.wait(pending)
+    backurl, MODULE_NAME = setup_cli("PROJECT_CLI", backend, DEFAULT_DEV_PORT)
-            for task in done2:
-                if task.get_name() == "install":
-                    if task.result() != 0:
-                        stderr.write("┃ ⚠️  Install failed\n")
-                        raise SystemExit(1)
-                elif task.get_name() == "backend_ready" and not task.result():
-                    raise SystemExit(1)
 
-        install_proc = None
+    # Tell everyone via environment (vite proxy and backend devmode use these)
+    os.environ["ENVPREFIX_VITE_URL"] = viteurl
+    os.environ["ENVPREFIX_BACKEND_URL"] = backurl
+    os.environ["ENVPREFIX_DEV"] = "1"
 
-        # Start Vite dev server
+    async with ProcessGroup() as pg:
-        stderr.write(f">>> {tool_name} {' '.join(dev_cmd[1:])}\n")
+        pg.create_task(check_ports_free(viteurl, backurl))
-        frontend_proc = await asyncio.create_subprocess_exec(*dev_cmd, cwd=frontend_cwd)
+        npm_i = await pg.spawn(*npm_install, cwd=front)
-
+        await pg.spawn(*MODULE_NAME, *(extra_args or []), vital=True)
-        # Wait for either process to exit
+        await pg.wait(npm_i, ready(backurl, path=HEALTH))
-        done, pending = await asyncio.wait(
+        await pg.spawn(*vite, cwd=front, vital=True)
-            {
-                asyncio.create_task(backend_proc.wait(), name="backend"),
-                asyncio.create_task(frontend_proc.wait(), name="frontend"),
-            },
-            return_when=asyncio.FIRST_COMPLETED,
-        )
-        for t in done:
-            t.result()
-        for t in pending:
-            t.cancel()
-
-    except asyncio.CancelledError:
-        stderr.write("\n✓ Shutting down...\n")
-    finally:
-        if frontend_proc is not None:
-            await _terminate_process(frontend_proc, "frontend")
-        if install_proc is not None:
-            await _terminate_process(install_proc, "install")
-        if backend_proc is not None:
-            await _terminate_process(backend_proc, "backend")
 
 
-def main():
+def main() -> None:
+    """Parse CLI arguments and run the devserver."""
+    tracerite.load()
     parser = argparse.ArgumentParser(
         description="Run Vite and FastAPI development servers",
         formatter_class=argparse.RawDescriptionHelpFormatter,
-        epilog=EPILOG,
+        epilog=HELP_EPILOG,
     )
     parser.add_argument(
-        "frontend",
+        "-l",
-        nargs="?",
+        "--listen",
-        metavar="host:port",
+        metavar="addr",
-        help="Vite frontend endpoint (default: localhost:5173)",
+        help=f"Vite (default: localhost:{DEFAULT_VITE_PORT})",
     )
     parser.add_argument(
         "--backend",
-        metavar="host:port",
+        metavar="addr",
-        help="FastAPI backend endpoint (default: localhost:5180)",
+        help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})",
     )
-    args = parser.parse_args()
+    args, extra_args = parser.parse_known_args()
+    try:
+        asyncio.run(run_devserver(args.listen, args.backend, extra_args))
+    except* KeyboardInterrupt:
+        pass  # user stopped the devserver: normal exit
+    except* (subprocess.SubprocessError, RuntimeError):
+        raise SystemExit(1) from None  # logged in devutil already; exit 1
 
-    vite_host, vite_port, all_ifaces = parse_endpoint(args.frontend, DEFAULT_VITE_PORT)
-    backend_host, backend_port, _ = parse_endpoint(args.backend, DEFAULT_BACKEND_PORT)
-    # Backend host defaults to localhost (never None)
-    backend_host = backend_host or DEFAULT_HOST
 
-    with contextlib.suppress(KeyboardInterrupt):
+HELP_EPILOG = """
-        asyncio.run(
+  Other options are forwarded to PROJECT_CLI [args]
-            run_devserver(vite_host, vite_port, all_ifaces, backend_host, backend_port)
+
-        )
+  JS_RUNTIME environment variable can be used to select the JS runtime:
+  npm, deno, bun, or full path to the runtime executable (node maps to npm).
+"""
 
 
 if __name__ == "__main__":

template/scripts/fastapi-vue/build-frontend.py

@@ -1,34 +0,0 @@
-"""Hatch build hook for building Vue frontend during package build."""
-
-import subprocess
-from pathlib import Path
-from sys import stderr
-
-from hatchling.builders.hooks.plugin.interface import BuildHookInterface  # type: ignore
-
-exec(Path(__file__).with_name("util.py").read_text("UTF-8"))  # noqa: S102
-
-
-def run(cmd, **kwargs):
-    """Run a command and display it."""
-    display_cmd = [Path(cmd[0]).name, *cmd[1:]]
-    stderr.write(f"### {' '.join(display_cmd)}\n")
-    subprocess.run(cmd, check=True, **kwargs)
-
-
-class CustomBuildHook(BuildHookInterface):
-    """Build hook that compiles Vue frontend before packaging."""
-
-    def initialize(self, version, build_data):
-        super().initialize(version, build_data)
-        stderr.write(">>> Building the frontend\n")
-
-        install_cmd, build_cmd = find_build_tool()  # noqa # type: ignore
-
-        try:
-            run(install_cmd, cwd="frontend")
-            stderr.write("\n")
-            run(build_cmd, cwd="frontend")
-        except Exception as e:
-            stderr.write(f"Error occurred while building frontend: {e}\n")
-            raise

template/scripts/fastapi-vue/buildhook.py

@@ -0,0 +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
+
+sys.path.insert(0, str(Path(__file__).parent))
+from buildutil import build
+
+
+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")

template/scripts/fastapi-vue/buildutil.py

@@ -0,0 +1,231 @@
+# ruff: noqa: INP001
+"""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
+
+MIN_NODE_VERSION = 20
+
+
+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(  # 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 >= MIN_NODE_VERSION:
+                return
+            msg = f"Node.js {version_str} found, but v{MIN_NODE_VERSION}+ required"
+            raise RuntimeError(msg)
+    except (subprocess.CalledProcessError, FileNotFoundError, ValueError):
+        pass
+    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 RuntimeError if no suitable runtime is found.
+    """
+    options = ["npm", "deno", "bun"]
+
+    # Check for JS_RUNTIME environment variable
+    if result := _find_runtime_from_env(options):
+        return result
+
+    # Auto-detect
+    return _auto_detect_runtime(options)
+
+
+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.
+    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", "-A", "npm:vite"),
+        "npm": ("--silent", "run", "dev", "--"),
+        "bun": ("run", "dev", "--"),
+    }
+
+    tool, name = find_js_runtime()
+
+    if name == "bun":
+        logger.warning(
+            "Bun has a WS proxy bug (github.com/oven-sh/bun/issues/9882). Consider npm.",
+        )
+
+    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) from None
+
+    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)  # noqa: S603
+
+    try:
+        run(install_cmd)
+        logger.info("")
+        run(build_cmd)
+    except subprocess.CalledProcessError:
+        raise SystemExit(1) from None

template/scripts/fastapi-vue/devutil.py

@@ -0,0 +1,222 @@
+# ruff: noqa: INP001
+"""Utilities meant for devserver script, used only in source repository with dev deps."""
+
+import asyncio
+import sys
+from asyncio.subprocess import Process
+from collections.abc import Awaitable
+from contextlib import suppress
+from pathlib import Path
+from subprocess import CalledProcessError
+from typing import Any
+from urllib.parse import urlsplit
+
+from buildutil import find_dev_tool, find_install_tool, logger
+from fastapi_vue.hostutil import parse_endpoint
+
+
+class ProcessGroup(asyncio.TaskGroup):
+    """TaskGroup with structured ownership of async subprocesses."""
+
+    def __init__(self, *, terminate_timeout: float = 10) -> None:
+        """Set the grace period before terminate() escalates to kill()."""
+        super().__init__()
+        self._terminate_timeout = terminate_timeout
+        self._cmds: dict[Process, tuple[str, ...]] = {}
+
+    async def spawn(self, *cmd: str, cwd: str | None = None, vital: bool = False) -> Process:
+        """Spawn and own a subprocess. If a vital process exits, the group cancels."""
+
+        async def run() -> None:
+            name = Path(cmd[0]).stem
+            logger.info(">>> %s", " ".join([name, *cmd[1:]]))
+            try:
+                proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd)
+                self._cmds[proc] = cmd
+                started.set_result(proc)
+            except Exception as e:  # noqa: BLE001
+                started.set_exception(e)
+                return
+
+            try:
+                returncode = await proc.wait()
+            finally:
+                with suppress(ProcessLookupError):
+                    proc.terminate()
+                try:
+                    await asyncio.wait_for(proc.wait(), self._terminate_timeout)
+                except TimeoutError:
+                    with suppress(ProcessLookupError):
+                        proc.kill()
+                    await proc.wait()
+
+            if vital:
+                logger.warning("Vital process %s exited", name)
+                raise CalledProcessError(returncode, cmd)
+
+        started = asyncio.get_running_loop().create_future()
+        self.create_task(run())
+        return await asyncio.shield(started)
+
+    async def wait(self, *waitables: Process | Awaitable) -> tuple[Any, ...]:
+        """Wait concurrently and return results in argument order."""
+
+        async def task(w: Process | Awaitable) -> Any:  # noqa: ANN401
+            if not isinstance(w, Process):
+                return await w
+            if retcode := await w.wait():
+                cmd = self._cmds[w]
+                logger.warning("Process %s exited with status %d", Path(cmd[0]).stem, retcode)
+                raise CalledProcessError(retcode, cmd)
+            return retcode
+
+        async with asyncio.TaskGroup() as group:
+            tasks = [group.create_task(task(w)) for w in waitables]
+
+        return tuple(task.result() for task in tasks)
+
+
+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(errors="replace").split("\r\n"):
+        if line.lower().startswith("server:"):
+            return line[7:].strip()
+    return ""
+
+
+async def check_ports_free(*urls: str) -> None:
+    """Verify URLs are not responding (ports are free).
+
+    Meant to run as a task inside a TaskGroup. Logs the conflict and raises
+    RuntimeError (handled like a failed process) if any URL responds.
+    """
+    servers = await asyncio.gather(*(http_get_server(url, timeout=0.1) for url in urls))
+    for url, server in zip(urls, servers, strict=True):
+        if server is not None:
+            logger.error("Conflicting %s already running at %s", server or "server", url)
+            raise RuntimeError(url)
+
+
+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 TimeoutError if server doesn't start in time.
+    """
+    if not path:
+        return
+
+    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:
+            raise TimeoutError(f"Backend at {url} didn't start in time")  # noqa: EM102, TRY003
+        await asyncio.sleep(0.1)
+
+
+def setup_vite(
+    endpoint: str,
+    default_port: int = 5173,
+) -> 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_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}")
+    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 = 8000,
+) -> tuple[str, list[str]]:
+    """Parse backend endpoint and build uvicorn command.
+
+    Returns (url, uvicorn_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"]
+    reload_dir = module.split(".", maxsplit=1)[0]  # Don't reload on frontend changes
+
+    cmd = [
+        sys.executable,
+        "-m",
+        "uvicorn",
+        module,
+        f"--host={host}",
+        f"--port={port}",
+        "--reload",
+        f"--reload-dir={reload_dir}",
+        "--forwarded-allow-ips=*",
+    ]
+    return f"http://{host}:{port}", cmd
+
+
+def setup_cli(
+    cli: str,
+    endpoint: str,
+    default_port: int = 8000,
+) -> tuple[str, list[str]]:
+    """Parse backend endpoint and build CLI command.
+
+    Returns (url, cli_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"]
+
+    # 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

template/scripts/fastapi-vue/util.py

@@ -1,88 +0,0 @@
-"""Shared utilities for build and dev scripts."""
-
-import os
-import shutil
-from pathlib import Path
-from sys import stderr
-
-
-def find_js_runtime() -> tuple[str, str] | None:
-    """Find a JavaScript runtime from JS_RUNTIME env or auto-detect.
-
-    Returns (tool_path, tool_name) where tool_name is "deno", "npm", or "bun".
-    Returns None if no runtime is found.
-    """
-    options = ["deno", "npm", "bun"]
-
-    # 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:
-                    stderr.write(f"┃ ⚠️  JS_RUNTIME={js_runtime_env} not found\n")
-                    return None
-                return tool, option
-        stderr.write(f"┃ ⚠️  JS_RUNTIME={js_runtime_env} not recognized\n")
-        return None
-
-    # Auto-detect
-    for option in options:
-        if tool := shutil.which(option):
-            return tool, option
-    return None
-
-
-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"),
-    }
-    # Use build-only for deno to avoid npm-run-all2 issues with run-p
-    # (run-p tries to spawn npm which doesn't exist in deno)
-    build = {
-        "deno": ("task", "build-only"),
-        "npm": ("run", "build"),
-        "bun": ("--bun", "run", "build"),
-    }
-
-    result = find_js_runtime()
-    if result is None:
-        raise RuntimeError(
-            "Deno, npm or Bun is required for building but none was found"
-        )
-
-    tool, name = result
-    return [tool, *install[name]], [tool, *build[name]]
-
-
-def find_dev_tool():
-    """Find JavaScript runtime and construct dev command.
-
-    Returns (dev_cmd, tool_name) or (None, None) if not found.
-    """
-    dev_args = {
-        "deno": ("run", "dev", "--"),
-        "npm": ("--silent", "run", "dev", "--"),
-        "bun": ("run", "dev", "--"),
-    }
-
-    result = find_js_runtime()
-    if result is None:
-        return None, None
-
-    tool, name = result
-    return [tool, *dev_args[name]], name