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

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/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

@@ -1,26 +1,91 @@
 """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 run(
+
+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 = False,
+    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.
@@ -29,8 +94,18 @@ def run(
         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 (requires uvicorn.run, single endpoint only).
+        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).
 
     """
@@ -39,7 +114,22 @@ def run(
         msg = "No endpoints to serve; check listen configuration"
         raise ValueError(msg)
 
-    conf: dict[str, object] = {"app": app, "reload": reload, "workers": workers}
+    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

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

@@ -114,36 +114,39 @@ class Frontend:
         """Load static files from disk with compression."""
         www: dict[str, tuple[bytes, bytes | None, dict]] = {}
         if not self.base.exists():
-            msg = f"Frontend folder {self.base} not found (try uv build)"
+            logger.error(
-            raise ValueError(msg)
+                "Missing %s - no frontend (try uv build)",
-        paths = [PurePath()]
+                self.base,
-        while paths:
+            )
-            current = self.base / paths.pop(0)
+        else:
-            for p in current.iterdir():
+            paths = [PurePath()]
-                rel = p.relative_to(self.base)
+            while paths:
-                if p.is_dir():
+                current = self.base / paths.pop(0)
-                    paths.append(rel)
+                for p in current.iterdir():
-                    continue
+                    rel = p.relative_to(self.base)
-                # Read file
+                    if p.is_dir():
-                name = "/" + rel.as_posix()
+                        paths.append(rel)
-                mime = mimetypes.guess_type(name)[0] or "application/octet-stream"
+                        continue
-                name = name.removesuffix(self.index)
+                    # Read file
-                data = p.read_bytes()
+                    name = "/" + rel.as_posix()
-                etag = urlsafe_b64encode(blake3(data).digest(9)).decode()
+                    mime = mimetypes.guess_type(name)[0] or "application/octet-stream"
-                if mime.startswith("text/"):
+                    name = name.removesuffix(self.index)
-                    mime += "; charset=UTF-8"
+                    data = p.read_bytes()
-                mtime = p.stat().st_mtime
+                    etag = urlsafe_b64encode(blake3(data).digest(9)).decode()
-                cached = any(name.startswith(prefix) for prefix in self.cached_paths)
+                    if mime.startswith("text/"):
-                headers = {
+                        mime += "; charset=UTF-8"
-                    "etag": f'"{etag}"',
+                    mtime = p.stat().st_mtime
-                    "last-modified": format_date_time(mtime),
+                    cached = any(name.startswith(prefix) for prefix in self.cached_paths)
-                    "cache-control": ("max-age=31536000, immutable" if cached else "no-cache"),
+                    headers = {
-                    "content-type": mime,
+                        "etag": f'"{etag}"',
-                }
+                        "last-modified": format_date_time(mtime),
-                zstd = ZstdCompressor(self.zstdlevel).compress(data)
+                        "cache-control": ("max-age=31536000, immutable" if cached else "no-cache"),
-                if len(zstd) >= len(data):
+                        "content-type": mime,
-                    zstd = None
+                    }
-                www[name] = data, zstd, headers
+                    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", ""):
@@ -220,7 +223,9 @@ class Frontend:
         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)
+            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)."""

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

@@ -8,11 +8,13 @@ dependencies = [
     "fastapi>=0.115.0",
     "zstandard>=0.23.0",
     "blake3>=1.0.8",
+    "tracerite>=2.6.5",
 ]
 
 [project.urls]
-Homepage = "https://git.zi.fi/LeoVasanko/fastapi-vue"
+Homepage = "https://vasanko.com/coders/fastapi-vue"
-Repository = "https://github.com/LeoVasanko/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"]

fastapi_vue_setup.py

@@ -9,6 +9,7 @@ Options:
     --module-name NAME      Python module name (auto-detected from pyproject.toml)
     --ports DEFAULT,VITE,DEV  Port configuration (default: 3100,3100,3200)
     --dry                   Show what would be done without making changes
+    -- ARGS                 Extra arguments forwarded to create-vue (e.g. -- --default)
 """
 
 import argparse
@@ -80,6 +81,7 @@ def ruff_format_content(
             [  # noqa: S607
                 "ruff",
                 "check",
+                "--ignore=EXE001,INP001,N999,CPY001",
                 "--fix",
                 "--output-format=concise",
                 str(temp_file),
@@ -165,7 +167,7 @@ frontend = Frontend(Path(__file__).with_name("frontend-build"))
 # Lifespan block for patching apps that don't have one
 LIFESPAN_BLOCK = """
 @asynccontextmanager
-async def lifespan(app: FastAPI):
+async def lifespan(_app: FastAPI):
     \"\"\"Manage app startup and shutdown resources.\"\"\"
     await frontend.load()
     yield
@@ -1210,7 +1212,9 @@ def ensure_python_project(project_dir: Path, *, dry: bool = False) -> bool:
     return True
 
 
-def ensure_frontend(project_dir: Path, *, dry: bool = False) -> bool:
+def ensure_frontend(
+    project_dir: Path, *, vue_args: list[str] | None = None, dry: bool = False
+) -> bool:
     """Ensure frontend directory exists with a Vue project, run create-vue if needed."""
     frontend_dir = project_dir / "frontend"
     package_json = frontend_dir / "package.json"
@@ -1233,6 +1237,10 @@ def ensure_frontend(project_dir: Path, *, dry: bool = False) -> bool:
         "bun": [js_tool, "create", "vue@latest", "frontend"],
     }
     create_cmd = create_vue_commands[js_name]
+    if vue_args:
+        # npm needs a `--` separator so it doesn't eat the arguments;
+        # create-vue runs non-interactively when given feature flags (e.g. --default)
+        create_cmd = [*create_cmd, *(["--"] if js_name == "npm" else []), *vue_args]
 
     if dry:
         print(f"🎨 Would run: {' '.join(create_cmd)}")
@@ -1240,7 +1248,8 @@ def ensure_frontend(project_dir: Path, *, dry: bool = False) -> bool:
 
     print("🎨 No frontend/ found, creating Vue project...")
     print(f">>> {' '.join(create_cmd)}")
-    print("(Follow the prompts to configure your Vue app)")
+    if not vue_args:
+        print("(Follow the prompts to configure your Vue app)")
     print()
     result = subprocess.run(create_cmd, cwd=project_dir, check=False)  # noqa: S603
     if result.returncode != 0:
@@ -1284,7 +1293,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
     print(f"🔧 Setting up project: {project_dir}")
 
     # Step 1: Ensure frontend exists (do this first so cancellation doesn't leave partial setup)
-    if not ensure_frontend(project_dir, dry=dry):
+    if not ensure_frontend(project_dir, vue_args=args.vue_args, dry=dry):
         return 1
 
     # Step 2: Ensure Python project exists
@@ -1375,7 +1384,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
         app_file, app_var = app_info
         print(f"📍 Found FastAPI app: {app_var} in {app_file.name}")
         tpl_vars["APP_VAR"] = app_var
-        # Dotted module path relative to project dir (e.g. "paskia.fastapi.mainapp")
+        # Dotted module path relative to project dir (e.g. "{module_name}.api.main")
         app_module = ".".join(app_file.relative_to(project_dir).with_suffix("").parts)
         tpl_vars["APP_MODULE"] = app_module
     else:
@@ -1595,12 +1604,16 @@ def cmd_setup(args: argparse.Namespace) -> int:
         print("✅ Created .gitignore")
 
     # === Add dependencies using uv ===
+    # Pin fastapi-vue to the same major.minor.patch as this setup tool (both are
+    # released from the same tags). This makes freshly set up projects request the
+    # matching patch release directly, while `~=` still allows compatible updates.
+    mmp = re.match(r"(\d+)\.(\d+)\.(\d+)", version)
+    fastapi_vue_req = f"fastapi-vue~={mmp[1]}.{mmp[2]}.{mmp[3]}" if mmp else "fastapi-vue"
     if dry:
-        print("📦 Would add: fastapi[standard], fastapi-vue, httpx (dev only)")
+        print(f"📦 Would add: fastapi[standard], {fastapi_vue_req}")
     else:
         print("📦 Dependencies")
-        uv_add_packages(["fastapi[standard]", "fastapi-vue"], cwd=project_dir)
+        uv_add_packages(["fastapi[standard]", fastapi_vue_req], cwd=project_dir)
-        uv_add_packages(["httpx"], cwd=project_dir, group="dev")
 
     print()
     print_boxed("Setup complete!")
@@ -1658,6 +1671,8 @@ Examples:
   fastapi-vue-setup .                  Set up integration in current directory
   fastapi-vue-setup . --dry            Preview what would be done
   fastapi-vue-setup . --ports=8000,5173,8080  Change default ports (backend, vite dev, backend dev)
+  fastapi-vue-setup my-app -- --default  Non-interactive create-vue (extra args after --
+                                         are forwarded to create-vue, e.g. --default, --ts)
 """,
     )
     parser.add_argument(
@@ -1680,7 +1695,15 @@ Examples:
     )
     parser.add_argument("--dry", "--dry-run", action="store_true", help="Show what would be done")
 
-    args = parser.parse_args()
+    # Everything after a standalone `--` is forwarded verbatim to create-vue
+    argv = sys.argv[1:]
+    if "--" in argv:
+        split = argv.index("--")
+        ours, vue_args = argv[:split], argv[split + 1 :]
+    else:
+        ours, vue_args = argv, []
+    args = parser.parse_args(ours)
+    args.vue_args = vue_args
 
     if args.project_dir is None:
         parser.print_help()

pyproject.toml

@@ -14,8 +14,9 @@ dependencies = [
 ]
 
 [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"
@@ -42,7 +43,7 @@ line-length = 100
 
 [tool.ruff.lint]
 select = ["ALL"]
-ignore = ["D203", "D213", "COM812"]  # Conflicting with D211, D212 and formatting
+ignore = ["CPY", "D203", "D213", "COM812", "PLR2004"]
 
 [tool.ruff.lint.per-file-ignores]
 "template/**" = ["F821"]  # Undefined names are template placeholders

template/backend/__main__.py

@@ -3,6 +3,7 @@
 
 import argparse
 import os
+from pathlib import Path
 
 from fastapi_vue import server
 
@@ -20,12 +21,12 @@ def main() -> None:
         help=(f"Endpoint (default: localhost:{DEFAULT_PORT})."),
     )
     args = parser.parse_args()
-    dev = {"reload": True, "reload_dirs": ["paskia"]} if DEVMODE else {}
     server.run(
         "APP_MODULE:APP_VAR",
         listen=args.listen,
         default_port=DEFAULT_PORT,
-        **dev,
+        server_header=False,
+        reload=Path(__file__).parent if DEVMODE else False,
     )
 
 

template/backend/app.py

@@ -1,6 +1,6 @@
 """FastAPI application module with Vue frontend integration."""
 
-from collections.abc import AsyncIterator
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from pathlib import Path
 
@@ -13,7 +13,7 @@ frontend = Frontend(Path(__file__).with_name("frontend-build"))
 
 
 @asynccontextmanager
-async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
+async def lifespan(_app: FastAPI) -> AsyncGenerator:
     """Manage app startup and shutdown resources."""
     await frontend.load()
     yield
@@ -27,7 +27,7 @@ app = FastAPI(title="PROJECT_TITLE", debug=DEVMODE, lifespan=lifespan)
 
 # Health check endpoint for the Vue demo app to verify the backend is running
 @app.get("/api/health")
-async def health_check() -> dict[str, str]:
+async def health_check() -> dict:
     """Return backend status for health monitoring."""
     return {"status": "ok"}
 

template/frontend/vite-plugin-fastapi.js

@@ -5,6 +5,7 @@
  * Configures Vite for FastAPI backend integration:
  * - Proxies /api/* requests to the FastAPI backend
  * - Builds to the Python module's frontend-build directory
+ * - Disables Vite's screen clearing on startup
  *
  * Options:
  *   paths - Array of paths to proxy (default: ["/api"])
@@ -26,6 +27,7 @@ export default function fastapiVue({ paths = ["/api"] } = {}) {
   return {
     name: "vite-plugin-fastapi-MODULE_NAME",
     config: () => ({
+      clearScreen: false,
       server: { proxy },
       build: {
         outDir: "../MODULE_NAME/frontend-build",

template/scripts/devserver.py

@@ -5,10 +5,12 @@
 import argparse
 import asyncio
 import os
+import subprocess
 import sys
-from contextlib import suppress
 from pathlib import Path
 
+import tracerite
+
 # 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 (
@@ -40,21 +42,22 @@ async def run_devserver(
     viteurl, npm_install, vite = setup_vite(listen, DEFAULT_VITE_PORT)
     backurl, MODULE_NAME = setup_cli("PROJECT_CLI", backend, DEFAULT_DEV_PORT)
 
-    # Tell the everyone by environment (vite proxy and backend devmode use these)
+    # 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"
 
     async with ProcessGroup() as pg:
+        pg.create_task(check_ports_free(viteurl, backurl))
         npm_i = await pg.spawn(*npm_install, cwd=front)
-        await check_ports_free(viteurl, backurl)
+        await pg.spawn(*MODULE_NAME, *(extra_args or []), vital=True)
-        await pg.spawn(*MODULE_NAME, *(extra_args or []))
         await pg.wait(npm_i, ready(backurl, path=HEALTH))
-        await pg.spawn(*vite, cwd=front)
+        await pg.spawn(*vite, cwd=front, vital=True)
 
 
 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,
@@ -72,8 +75,12 @@ def main() -> None:
         help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})",
     )
     args, extra_args = parser.parse_known_args()
-    with suppress(KeyboardInterrupt):
+    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
 
 
 HELP_EPILOG = """

template/scripts/fastapi-vue/devutil.py

@@ -2,143 +2,138 @@
 """Utilities meant for devserver script, used only in source repository with dev deps."""
 
 import asyncio
-import subprocess
 import sys
+from asyncio.subprocess import Process
+from collections.abc import Awaitable
 from contextlib import suppress
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Self
+from subprocess import CalledProcessError
+from typing import Any
+from urllib.parse import urlsplit
 
-import httpx
 from buildutil import find_dev_tool, find_install_tool, logger
 from fastapi_vue.hostutil import parse_endpoint
 
-if TYPE_CHECKING:
-    from collections.abc import Coroutine
 
+class ProcessGroup(asyncio.TaskGroup):
+    """TaskGroup with structured ownership of async subprocesses."""
 
-class ProcessGroup:
+    def __init__(self, *, terminate_timeout: float = 10) -> None:
-    """Manage async subprocesses with automatic cleanup, like TaskGroup for processes."""
+        """Set the grace period before terminate() escalates to kill()."""
+        super().__init__()
+        self._terminate_timeout = terminate_timeout
+        self._cmds: dict[Process, tuple[str, ...]] = {}
 
-    def __init__(self) -> None:
+    async def spawn(self, *cmd: str, cwd: str | None = None, vital: bool = False) -> Process:
-        """Initialize empty process tracking."""
+        """Spawn and own a subprocess. If a vital process exits, the group cancels."""
-        self._procs: list[asyncio.subprocess.Process] = []
-        self._cmds: dict[int, str] = {}  # pid -> command name
 
-    async def spawn(
+        async def run() -> None:
-        self,
+            name = Path(cmd[0]).stem
-        *cmd: str,
+            logger.info(">>> %s", " ".join([name, *cmd[1:]]))
-        cwd: str | None = None,
+            try:
-    ) -> asyncio.subprocess.Process:
+                proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd)
-        """Spawn a subprocess and track it."""
+                self._cmds[proc] = cmd
-        cmd_name = Path(cmd[0]).stem
+                started.set_result(proc)
-        logger.info(">>> %s", " ".join([cmd_name, *cmd[1:]]))
+            except Exception as e:  # noqa: BLE001
-        proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd)
+                started.set_exception(e)
-        self._procs.append(proc)
+                return
-        self._cmds[proc.pid] = cmd_name
-        return proc
 
-    async def wait(
+            try:
-        self,
+                returncode = await proc.wait()
-        *waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]",
+            finally:
-    ) -> None:
-        """Wait for processes/coroutines to complete, raise SystemExit on failure."""
-
-        async def wait_proc(proc: asyncio.subprocess.Process) -> None:
-            returncode = await proc.wait()
-            if returncode != 0:
-                cmd_name = self._cmds.get(proc.pid, "unknown")
-                raise subprocess.CalledProcessError(returncode, cmd_name)
-
-        tasks = [
-            wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w for w in waitables
-        ]
-        try:
-            await asyncio.gather(*tasks)
-        except subprocess.CalledProcessError as e:
-            logger.warning("%s failed with exit status %d", e.cmd, e.returncode)
-            raise SystemExit(1) from None
-
-    async def __aenter__(self) -> Self:
-        """Enter the async context manager."""
-        return self
-
-    async def __aexit__(self, exc_type: type[BaseException] | None, *_: object) -> None:
-        """Wait for one process to exit, terminate others, then wait for all."""
-        await self._cleanup(immediate=exc_type is not None)
-
-    async def _cleanup(self, *, immediate: bool = False) -> None:
-        running = [p for p in self._procs if p.returncode is None]
-        if not running:
-            return
-
-        if not immediate:
-            # Wait for any one process to exit
-            with suppress(asyncio.CancelledError):
-                await asyncio.wait(
-                    [asyncio.create_task(p.wait()) for p in running],
-                    return_when=asyncio.FIRST_COMPLETED,
-                )
-
-        # Terminate remaining processes
-        for p in self._procs:
-            if p.returncode is None:
                 with suppress(ProcessLookupError):
-                    p.terminate()
+                    proc.terminate()
-
-        # Wait for all to finish (with overall timeout), shielded from cancellation
-        still_running = [p for p in self._procs if p.returncode is None]
-        if still_running:
-            with suppress(asyncio.CancelledError):
                 try:
-                    await asyncio.shield(
+                    await asyncio.wait_for(proc.wait(), self._terminate_timeout)
-                        asyncio.wait_for(
-                            asyncio.gather(*[p.wait() for p in still_running]),
-                            timeout=10,
-                        ),
-                    )
                 except TimeoutError:
-                    for p in self._procs:
+                    with suppress(ProcessLookupError):
-                        if p.returncode is None:
+                        proc.kill()
-                            with suppress(ProcessLookupError):
+                    await proc.wait()
-                                p.kill()
+
-                            await p.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). Raise SystemExit if any respond."""
+    """Verify URLs are not responding (ports are free).
 
-    async def check(client: httpx.AsyncClient, url: str) -> None:
+    Meant to run as a task inside a TaskGroup. Logs the conflict and raises
-        with suppress(httpx.RequestError):
+    RuntimeError (handled like a failed process) if any URL responds.
-            res = await client.get(url, timeout=0.1)
+    """
-            server = res.headers.get("server", "server")
+    servers = await asyncio.gather(*(http_get_server(url, timeout=0.1) for url in urls))
-            logger.warning("Conflicting %s already running at %s", server, url)
+    for url, server in zip(urls, servers, strict=True):
-            raise SystemExit(1)
+        if server is not None:
-
+            logger.error("Conflicting %s already running at %s", server or "server", url)
-    async with httpx.AsyncClient() as client:
+            raise RuntimeError(url)
-        await asyncio.gather(*[check(client, url) for url in urls])
 
 
 async def ready(url: str, path: str = "", max_attempts: int = 50) -> None:
     """Wait for the server to be ready by polling an endpoint.
 
     Use empty path to disable the check and make this return immediately.
-    Raises SystemExit(1) if server doesn't start in time.
+    Raises TimeoutError if server doesn't start in time.
     """
     if not path:
         return
 
-    async with httpx.AsyncClient() as client:
+    for attempt in range(max_attempts):
-        for attempt in range(max_attempts):
+        if await http_get_server(f"{url}{path}", timeout=1.0) is not None:
-            try:
+            logger.info("✓ Backend ready!")
-                await client.get(f"{url}{path}", timeout=1.0)
+            return
-            except httpx.RequestError:
+        if attempt == max_attempts - 1:
-                if attempt == max_attempts - 1:
+            raise TimeoutError(f"Backend at {url} didn't start in time")  # noqa: EM102, TRY003
-                    logger.warning("Backend didn't start in time")
+        await asyncio.sleep(0.1)
-                    raise SystemExit(1) from None
-                await asyncio.sleep(0.1)
-            else:
-                logger.info("✓ Backend ready!")
-                return
 
 
 def setup_vite(
@@ -221,5 +216,7 @@ def setup_cli(
     host = endpoints[0]["host"]
     port = endpoints[0]["port"]
 
-    cmd = [cli, f"--listen={host}:{port}"]
+    # Run the package as a module with the current interpreter, instead of
+    # relying on a PATH-installed CLI entry point.
+    cmd = [sys.executable, "-m", cli, f"--listen={host}:{port}"]
     return f"http://{host}:{port}", cmd