Compare commits

..
29 Commits
Author SHA1 Message Date
LeoVasanko 4f09622241 Suppress watchfiles "INFO: 1 change detected" which is useless given we already get a WARNING about which file and Reloading... 2026-09-02 13:56:25 +00:00
LeoVasanko 372d91bf49 Pin fastapi-vue minimum patch version to that of fastapi-vue-setup when ran, to force upgrades. 2026-08-31 21:14:04 +00:00
LeoVasanko b0b216b784 Run backend as module (python -m) in devserver instead of PATH-installed CLI. 2026-08-31 20:26:30 +00:00
LeoVasanko 475a2cdc3c Cleaner and non-deprecated typing for template app module. Avoids ruff errors with strict configurations. 2026-08-31 20:14:39 +00:00
LeoVasanko 8774224545 Apply lifespan error logging patch in reload/worker subprocesses too. 2026-08-31 20:04:25 +00:00
LeoVasanko cb742000a8 Render debug-mode HTTP errors with tracerite, replacing Starlette's formatter. 2026-08-31 20:00:39 +00:00
LeoVasanko be8dc0c513 Don't clear screen on vite startup as that hides our own startup banner. 2026-08-31 19:53:22 +00:00
LeoVasanko e4c21109d9 Prettier logging of errors in lifespan startup/shutdown. 2026-08-31 19:23:59 +00:00
LeoVasanko 1287ba4077 Load tracerite for devserver script too. 2026-08-31 19:23:07 +00:00
LeoVasanko 9c51b89226 Make missing frontend files non-fatal: log an error and serve the placeholder page instead of raising during lifespan startup 2026-08-31 18:48:27 +00:00
LeoVasanko d5e94a2186 Startup box: server.run() prints app name, version and connect URL in a rounded unicode box on stderr. Shared termwidth module for ANSI-aware unicode display width and padding. 2026-08-31 18:22:06 +00:00
LeoVasanko c65d8eaa12 Target project no longer depends on httpx but depends on ~matching fastapi-vue version. Ruff cleanup. 2026-08-31 17:07:49 +00:00
LeoVasanko cb0b1f067d A hack to silence INFO level log chatter from uvicorn, while still doing this via log config. 2026-08-31 16:22:16 +00:00
LeoVasanko 5dbe9a0dcd Ignore executable flag of the temporary file being used to build devserver script. 2026-08-31 15:52:56 +00:00
LeoVasanko 0bba199376 Implement a pretty logging config by default. Emoji levels. Fixes root logger info being lost and warning+ printing with lastResort without any formatting. Replaces uvicorn's default formatter and log config that the author is not willing to fix (any time soon). 2026-08-31 15:52:23 +00:00
LeoVasanko 4232304a00 Added TraceRite CLI error logging. Simplified logging setup. 2026-08-31 05:34:33 +00:00
LeoVasanko 77a34753d9 Add colored HTTP/WebSocket access logging to fastapi_vue.server.
run() gains access_log (default on) and log_config parameters.  When
enabled, uvicorn's own access log is disabled (access_log=False) and
replaced by an ASGI middleware (adapted from the uvicorn access-log
branch) that logs colored request lines, WebSocket open/close pairs,
and pre-accept rejections in open format.

log_config dicts (uvicorn's default or user-supplied) are patched
best-effort via a registered patch pipeline in fastapi_vue.logging:
swapping in our AccessFormatter, routing our output to a private
fastapi_vue.access logger (uvicorn gates its protocol-level access
logging on uvicorn.access.hasHandlers(), so that logger must stay
handlerless), and filtering stock WebSocket handshake chatter from
uvicorn.error.  Non-dict or structurally unexpected configs pass
through untouched.

The middleware is installed by patching Config.load/load_app, hooked
into AccessFormatter instantiation so reload/worker subprocesses
(which re-apply log_config without re-calling run()) are covered.
2026-08-31 04:16:43 +00:00
LeoVasanko 52ca3f4f49 Update project URLs. 2026-08-31 02:18:08 +00:00
LeoVasanko ebc13c0ee4 Add ruff ignores for the check phase to avoid useless diagnostics. 2026-08-26 01:07:24 +00:00
LeoVasanko 7a64d8f73b Disable server: uvicorn header. 2026-08-25 23:42:21 +00:00
LeoVasanko a5b7d88a89 Fix typo in devserver.py comment
'Tell the everyone' -> 'Tell everyone via environment'.
2026-08-25 23:41:46 +00:00
LeoVasanko 03aa31cedb Implement smarter reload handing on fastapi_vue.server(), simplify __main__, reload the main folder regardless of CWD. 2026-06-29 18:49:18 +00:00
LeoVasanko ec6a084969 Fix hardcoded project dir in template __main__.py (regression in commit 1b24a7). 2026-06-29 18:18:04 +00:00
LeoVasanko 15a6710c17 [lint] Format template after change. 2026-03-08 17:35:02 +00:00
LeoVasanko dcaf415032 Fix a typing issue with the fastapi-vue staticfiles module causing problems with SPA mode handler. 2026-03-08 17:30:42 +00:00
LeoVasanko 79316eebd4 Rename build-frontend.py to buildhook.py. 2026-03-08 15:40:49 +00:00
LeoVasanko 682d70cb23 Cleanup for ALL ruff checks, and re-ruff to target project settings when installing templates, avoiding formatting errors after patching. 2026-03-08 15:25:52 +00:00
LeoVasanko 1aaf040db7 Make devserver backend health check endpoint configurable, preserved in upgrades and optional. 2026-03-07 23:55:10 +00:00
LeoVasanko aac27da67c Suppress CancelledError along with KeyboardInterrupt as a normal exit from server.run(). 2026-02-18 17:50:34 +00:00
21 changed files with 1539 additions and 381 deletions
+1 -1
View File
@@ -84,7 +84,7 @@ my-app/
└── scripts/
├── devserver.py # Run Vite and FastAPI together in dev mode
└── fastapi-vue/ # Dev utilities (only on the source tree)
├── build-frontend.py
├── buildhook.py
├── buildutil.py
└── devutil.py
```
+2
View File
@@ -1,3 +1,5 @@
"""FastAPI Vue integration - serve Vue frontend from FastAPI."""
from .staticfiles import Frontend
__all__ = ["Frontend"]
+418
View File
@@ -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)
+68 -33
View File
@@ -1,8 +1,60 @@
"""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.
@@ -23,6 +75,7 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
- [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}]
@@ -31,51 +84,33 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
if value.isdigit():
return [{"host": "localhost", "port": int(value)}]
# Leading colon :port -> bind all interfaces (0.0.0.0 + ::)
if value.startswith(":") and value != ":":
port_part = value[1:]
if not port_part.isdigit():
raise SystemExit(f"Invalid port in '{value}'")
port = int(port_part)
return [{"host": "0.0.0.0", "port": port}, {"host": "::", "port": port}] # noqa: S104
# Try specialized parsers in order
result = _parse_all_interfaces(value)
if result is not None:
return result
# UNIX domain socket (unix:/path or just /path)
if value.startswith("/"):
return [{"uds": value}]
if value.startswith("unix:"):
uds_path = value[5:] or None
if uds_path is None:
raise SystemExit("unix: path must not be empty")
return [{"uds": uds_path}]
result = _parse_unix_socket(value)
if result is not None:
return result
# Unbracketed IPv6 (cannot safely contain a port) -> detect by multiple colons
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 [{"host": value, "port": default_port}]
result = _parse_unbracketed_ipv6(value, default_port)
if result is not None:
return result
# Use urllib.parse for everything else (host[:port], [ipv6][:port])
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}]
# 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
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}"]
+387
View File
@@ -0,0 +1,387 @@
"""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 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)
_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 = sys.stdout.isatty()
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
+100 -14
View File
@@ -1,54 +1,143 @@
"""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,
)
from .startupbox import print_box
tracerite.load() # Early load on CLI load (import server); uvicorn workers reload via log config
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,
**uvicorn_config,
):
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 (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).
"""
endpoints = parse_endpoints(listen, default_port)
if not endpoints:
raise ValueError("No endpoints to serve; check listen configuration")
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
conf["forwarded_allow_ips"] = proxy
conf.update(uvicorn_config)
with suppress(KeyboardInterrupt):
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) -> None:
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:
@@ -59,13 +148,10 @@ async def serve(endpoints: list[dict], **kwargs) -> None:
await asyncio.gather(*(Server(Config(**kwargs, **ep)).serve() for ep in endpoints))
def serve_multiprocess(endpoints: list[dict], **kwargs) -> None:
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
]
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],
+20
View File
@@ -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")
+69 -57
View File
@@ -25,7 +25,7 @@ __all__ = ["Frontend"]
class Assets:
"""Default cached value to /assets/"""
"""Default cached value to /assets/."""
@staticmethod
def parse(cached: str | list[str] | Assets) -> list[str]:
@@ -37,7 +37,8 @@ class Assets:
case list():
return cached
case _:
raise ValueError(f"Invalid cached value: {cached!r}")
msg = f"Invalid cached value: {cached!r}"
raise ValueError(msg)
class Frontend:
@@ -55,27 +56,29 @@ class Frontend:
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: May use wildcards of full path. E.g. /assets/logo*.png matches logo.hash.png created by Vite
favicon: Wildcard path to favicon. E.g. /assets/logo*.png matches Vite output
zstdlevel: Zstd compression level (default: 18)
"""
def __init__(
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 = Assets(),
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)
self.cached_paths = Assets.parse(cached if cached is not None else Assets())
self.zstdlevel = zstdlevel
self.favicon = favicon
self._app: FastAPI | None = None
@@ -107,47 +110,48 @@ class Frontend:
paths.add("/favicon.ico")
return paths
def _load(self):
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():
raise ValueError(f"Frontend folder {self.base} not found (try uv build)")
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:
if 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
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["/"] = (
@@ -161,7 +165,7 @@ class Frontend:
)
return www
async def load(self, *, debug: bool | None = None, log: bool = True):
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)
@@ -187,13 +191,19 @@ class Frontend:
ratio = comp / raw * 100 if raw else 100.0
if log and self.www:
logger.info(
f"{self.base.name}: {len(self.www)} files in {1000 * duration:.1f} ms | "
f"zstd {len(compfiles)} files {1e-6 * raw:.2f}->{1e-6 * comp:.2f} MB ({ratio:.0f} %)"
"%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="/"):
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.
@@ -204,6 +214,7 @@ class Frontend:
Args:
app: FastAPI application instance
mount_path: Path where the frontend should be mounted (default: "/")
"""
self._app = app
self._mount_path = mount_path.rstrip("/")
@@ -212,9 +223,11 @@ 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")(self.handle)
app.api_route(path, methods=["GET", "HEAD"], name="frontend", response_model=None)(
self.handle
)
def _register_routes(self):
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
@@ -240,18 +253,19 @@ class Frontend:
for p in paths
]
def _respond(self, request: Request, name: str):
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"}
content=zstd,
headers={**headers, "content-encoding": "zstd"},
)
return Response(content=data, headers=headers)
def handle(self, request: Request, path: str):
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)
@@ -271,11 +285,9 @@ class Frontend:
return (_devmode_respond if debug else self._respond)(request, name)
def _devmode_respond(request: 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] Not serving frontend files here. Should you connect to Vite instead?"
},
content={"detail": "[devmode] Use Vite devserver instead."},
)
+37
View File
@@ -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)
+4 -2
View File
@@ -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"
Repository = "https://github.com/LeoVasanko/fastapi-vue"
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"]
+198 -152
View File
@@ -1,4 +1,4 @@
"""FastAPI-Vue Integration Tool
"""FastAPI-Vue Integration Tool.
Create new FastAPI+Vue projects or patch existing ones with integrated build/dev systems.
@@ -13,8 +13,11 @@ Options:
import argparse
import ast
import contextlib
import hashlib
import importlib.metadata
import os
import platform
import re
import shutil
import subprocess
@@ -53,46 +56,60 @@ def ruff_format_content(
temp_file = target_path.with_suffix(".new.py")
try:
temp_file.write_text(content, "UTF-8", newline="\n")
# Sort imports first
subprocess.run(
[
"uv",
"run",
"--with",
"ruff",
if mode == "isort":
# Sort imports only (ignore exit code)
result = subprocess.run( # noqa: S603
[ # noqa: S607
"ruff",
"check",
"--select",
"I",
"--fix",
"--output-format=concise",
str(temp_file),
],
cwd=target_path.parent,
capture_output=True,
check=False,
)
if result.returncode != 0:
print(result.stdout.decode())
return temp_file.read_text("UTF-8")
# Full mode: fix all auto-fixable lint violations (ignore exit code)
result = subprocess.run( # noqa: S603
[ # noqa: S607
"ruff",
"check",
"--select",
"I",
"--ignore=EXE001,INP001,N999,CPY001",
"--fix",
"--output-format=concise",
str(temp_file),
],
cwd=target_path.parent,
capture_output=True,
check=False,
)
if mode == "isort":
return temp_file.read_text("UTF-8")
# Then format
result = subprocess.run(
["uv", "run", "--with", "ruff", "ruff", "format", str(temp_file)],
if result.returncode != 0:
print(result.stdout.decode())
# Then format (ignore exit code)
result = subprocess.run( # noqa: S603
["ruff", "format", str(temp_file)], # noqa: S607
cwd=target_path.parent,
capture_output=True,
check=False,
)
if result.returncode == 0:
return temp_file.read_text("UTF-8")
except Exception:
if result.returncode != 0:
print(result.stdout.decode())
return temp_file.read_text("UTF-8")
except OSError:
pass
finally:
try:
with contextlib.suppress(Exception):
temp_file.unlink(missing_ok=True)
except Exception:
pass
return content
def uv_add_packages(
packages: list[str], *, cwd: Path, group: str | None = None
) -> None:
def uv_add_packages(packages: list[str], *, cwd: Path, group: str | None = None) -> None:
"""Add packages using uv."""
cmd = ["uv", "add", "-q", "-U"]
if group:
@@ -100,7 +117,7 @@ def uv_add_packages(
else:
cmd.append("--no-sync")
cmd.extend(packages)
result = subprocess.run(cmd, cwd=cwd, check=False)
result = subprocess.run(cmd, cwd=cwd, check=False) # noqa: S603
if result.returncode != 0:
label = f" ({group})" if group else ""
print(f"⚠️ Failed to add{label} dependencies")
@@ -110,6 +127,9 @@ def uv_add_packages(
# If vite == dev, dev is incremented by 100
DEFAULT_PORTS = (3100, 3100, 3200)
# Default health check path for devserver backend readiness check
DEFAULT_HEALTH = "/api/health?from=devserver.py"
# Marker comment indicating file can be auto-upgraded
# Users should remove this line to prevent automatic updates
UPGRADE_MARKER = "auto-upgrade@fastapi-vue-setup"
@@ -123,9 +143,7 @@ PYPROJECT_ADDITIONS = {
"artifacts": ["MODULE_NAME/frontend-build"],
"targets": {
"sdist": {
"hooks": {
"custom": {"path": "scripts/fastapi-vue/build-frontend.py"}
},
"hooks": {"custom": {"path": "scripts/fastapi-vue/buildhook.py"}},
}
},
"only-packages": True,
@@ -134,6 +152,10 @@ PYPROJECT_ADDITIONS = {
},
}
# Old build hook path that should be migrated to the new name
OLD_BUILD_HOOK_PATH = "scripts/fastapi-vue/build-frontend.py"
NEW_BUILD_HOOK_PATH = "scripts/fastapi-vue/buildhook.py"
# Frontend instantiation block for patching existing apps
FRONTEND_BLOCK = """
@@ -144,7 +166,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
@@ -233,7 +255,8 @@ def parse_ports(ports_str: str | None) -> tuple[int, int, int]:
vite = int(parts[1])
dev = int(parts[2])
else:
raise ValueError(f"Invalid ports format: {ports_str}")
msg = f"Invalid ports format: {ports_str}"
raise ValueError(msg)
# Auto-adjust dev if it conflicts with vite
if dev == vite:
@@ -260,9 +283,7 @@ def find_import_insertion_line(source: str) -> int:
return 2 if source.startswith("#!") else 1
def extract_existing_ports(
project_dir: Path, main: Path
) -> tuple[int, int, int] | None:
def extract_existing_ports(project_dir: Path, main: Path) -> tuple[int, int, int] | None:
"""Extract existing port configuration from project files.
Returns (default, vite, dev) or None if not found.
@@ -300,6 +321,30 @@ def extract_existing_ports(
return None
# Sentinel for "not found" in extract_existing_health
_HEALTH_NOT_FOUND = object()
def extract_existing_health(project_dir: Path) -> str | object:
"""Extract existing health path configuration from devserver.py.
Returns:
- The path string (may be empty to disable)
- _HEALTH_NOT_FOUND sentinel if not found or file doesn't exist
"""
devserver_file = project_dir / "scripts" / "devserver.py"
if not devserver_file.exists():
return _HEALTH_NOT_FOUND
content = devserver_file.read_text("UTF-8")
# Match HEALTH = "/path" or HEALTH = ""
match = re.search(r'^HEALTH\s*=\s*"([^"]*)"', content, re.MULTILINE)
if match:
return match.group(1)
return _HEALTH_NOT_FOUND
def load_template(path: str) -> str:
"""Load a template file from the template directory."""
return (TEMPLATE_DIR / path).read_text("UTF-8")
@@ -320,9 +365,7 @@ def find_module_name(project_dir: Path) -> str | None:
return None
def find_fastapi_app(
module_dir: Path, project_dir: Path | None = None
) -> tuple[Path, str] | None:
def find_fastapi_app(module_dir: Path, project_dir: Path | None = None) -> tuple[Path, str] | None:
"""Find the FastAPI app in a module directory.
Returns (file_path, app_variable_name) or None if not found.
@@ -360,9 +403,7 @@ def find_fastapi_app(
return None
def _find_app_via_entrypoint(
module_dir: Path, project_dir: Path
) -> tuple[Path, str] | None:
def _find_app_via_entrypoint(module_dir: Path, project_dir: Path) -> tuple[Path, str] | None:
"""Find FastAPI app by following the CLI entrypoint in pyproject.toml.
If pyproject.toml has a script like `myapp = "myapp.subpkg.__main__:main"`,
@@ -374,7 +415,7 @@ def _find_app_via_entrypoint(
try:
data = tomlkit.parse(pyproject.read_text("UTF-8"))
except Exception:
except (OSError, ValueError):
return None
scripts = data.get("project", {}).get("scripts", {})
@@ -384,7 +425,7 @@ def _find_app_via_entrypoint(
module_name = module_dir.name
# Find script entries that reference this module
for script_name, entry in scripts.items():
for entry in scripts.values():
if not isinstance(entry, str):
continue
# Parse entry like "module.subpkg.__main__:main"
@@ -442,7 +483,7 @@ def _add_devmode_to_main(content: str) -> str:
insert_idx = 0
for i, line in enumerate(lines):
stripped = line.strip()
if stripped.startswith("import ") or stripped.startswith("from "):
if stripped.startswith(("import ", "from ")):
insert_idx = i + 1
elif stripped and not stripped.startswith("#"):
break
@@ -475,7 +516,7 @@ def _find_existing_cli_module_path(project_dir: Path, module_name: str) -> str |
try:
data = tomlkit.parse(pyproject.read_text("UTF-8"))
except Exception:
except (OSError, ValueError):
return None
scripts = data.get("project", {}).get("scripts", {})
@@ -483,12 +524,11 @@ def _find_existing_cli_module_path(project_dir: Path, module_name: str) -> str |
return None
# Look for any script that references our module
for script_name, entry in scripts.items():
if isinstance(entry, str) and entry.startswith(f"{module_name}."):
for entry in scripts.values():
if isinstance(entry, str) and entry.startswith(f"{module_name}.") and ":" in entry:
# Extract module path from "module.subpkg.__main__:main"
if ":" in entry:
module_path, _ = entry.rsplit(":", 1)
return module_path
module_path, _ = entry.rsplit(":", 1)
return module_path
return None
@@ -502,7 +542,7 @@ def _follow_init_reexport(init_file: Path, subpkg_dir: Path) -> tuple[Path, str]
"""
try:
content = init_file.read_text("UTF-8")
except Exception:
except OSError:
return None
# Look for: from .module import app (or similar variable names)
@@ -536,7 +576,7 @@ def _find_app_in_file(path: Path) -> str | None:
"""Find FastAPI app variable name in a file."""
try:
content = path.read_text("UTF-8")
except Exception:
except OSError:
return None
# Look for FastAPI() instantiation patterns
@@ -548,17 +588,15 @@ def _find_app_in_file(path: Path) -> str | None:
return None
def render_template(template: str, **kwargs) -> str:
"""Simple template rendering replacing KEY with value."""
def render_template(template: str, **kwargs: str) -> str:
"""Render a template, replacing KEY with value."""
result = template
for key, value in kwargs.items():
result = result.replace(key, value)
return result
def patch_app_file(
path: Path, main_module_path: str, app_var: str, dry: bool = False
) -> bool:
def patch_app_file(path: Path, main_module_path: str, app_var: str, *, dry: bool = False) -> bool:
"""Patch an existing app.py with frontend integration.
Inserts imports at top (ruff will sort them), route at bottom,
@@ -602,9 +640,7 @@ def patch_app_file(
content = content.rstrip("\n") + "\n" + import_text
else:
# Insert at the found position
content = (
"".join(lines[:insert_idx]) + import_text + "".join(lines[insert_idx:])
)
content = "".join(lines[:insert_idx]) + import_text + "".join(lines[insert_idx:])
# Insert FRONTEND_BLOCK after last import (only if Frontend wasn't already there)
if not has_frontend:
@@ -612,7 +648,7 @@ def patch_app_file(
last_import_idx = 0
for i, line in enumerate(lines):
stripped = line.strip()
if stripped.startswith("import ") or stripped.startswith("from "):
if stripped.startswith(("import ", "from ")):
last_import_idx = i
elif stripped and not stripped.startswith("#") and last_import_idx > 0:
break
@@ -623,9 +659,7 @@ def patch_app_file(
if route_line not in content:
lines = content.split("\n")
lines.append("")
lines.append(
"# Serve the Vue frontend (needs to be last if SPA catch-all is used)"
)
lines.append("# Serve the Vue frontend (needs to be last if SPA catch-all is used)")
lines.append(route_line)
content = "\n".join(lines)
@@ -636,10 +670,7 @@ def patch_app_file(
args = match.group(2)
if "debug" not in args:
# Add debug=DEVMODE as last argument
if args.strip():
new_args = f"{args}, debug=DEVMODE"
else:
new_args = "debug=DEVMODE"
new_args = f"{args}, debug=DEVMODE" if args.strip() else "debug=DEVMODE"
content = (
content[: match.start()]
+ match.group(1)
@@ -675,11 +706,7 @@ def patch_app_file(
if insert_idx >= len(lines):
content = content.rstrip("\n") + "\n" + import_text
else:
content = (
"".join(lines[:insert_idx])
+ import_text
+ "".join(lines[insert_idx:])
)
content = "".join(lines[:insert_idx]) + import_text + "".join(lines[insert_idx:])
# Insert lifespan block before the FastAPI() call
fastapi_line_pattern = r"^(\w+\s*=\s*FastAPI\s*\()"
@@ -697,10 +724,7 @@ def patch_app_file(
fastapi_match = re.search(fastapi_pattern, content, re.DOTALL)
if fastapi_match and "lifespan" not in fastapi_match.group(2):
args = fastapi_match.group(2)
if args.strip():
new_args = f"{args}, lifespan=lifespan"
else:
new_args = "lifespan=lifespan"
new_args = f"{args}, lifespan=lifespan" if args.strip() else "lifespan=lifespan"
content = (
content[: fastapi_match.start()]
+ fastapi_match.group(1)
@@ -744,7 +768,7 @@ def patch_app_file(
def patch_vite_config(
path: Path,
module_name: str,
*,
dry: bool = False,
) -> bool:
"""Patch an existing vite.config.js/ts by adding fastapi-vue plugin.
@@ -775,15 +799,13 @@ def patch_vite_config(
# Insert after the last import line before non-import content
if not import_inserted:
stripped = line.strip()
if stripped.startswith("import ") or stripped.startswith("from "):
# Check if next line is not an import
if i + 1 < len(lines):
next_stripped = lines[i + 1].strip()
if not next_stripped.startswith(
"import "
) and not next_stripped.startswith("from "):
new_lines.append(import_line)
import_inserted = True
if stripped.startswith(("import ", "from ")) and i + 1 < len(lines):
next_stripped = lines[i + 1].strip()
if not next_stripped.startswith("import ") and not next_stripped.startswith(
"from "
):
new_lines.append(import_line)
import_inserted = True
if not import_inserted:
# No imports found, add at top
@@ -816,7 +838,7 @@ def patch_vite_config(
return True
def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool:
def patch_frontend_health_check(frontend_dir: Path, *, dry: bool = False) -> bool:
"""Patch Vue app to include FastAPI backend health check.
Tries HelloWorld.vue first (full demo), then falls back to App.vue (minimal).
@@ -858,9 +880,7 @@ def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool:
is_typescript = 'lang="ts"' in content
# Build the script content based on JS/TS
script_addition = (
TS_HEALTH_CHECK_SCRIPT if is_typescript else JS_HEALTH_CHECK_SCRIPT
)
script_addition = TS_HEALTH_CHECK_SCRIPT if is_typescript else JS_HEALTH_CHECK_SCRIPT
# Insert script addition before </script>
script_end_match = re.search(r"</script>", content)
@@ -878,9 +898,7 @@ def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool:
# Insert before closing </h3>
h3_close = content.find(" </h3>")
if h3_close == -1:
print(
f"⚠️ Skipping {target_file} (no </h3> tag found for status insertion)"
)
print(f"⚠️ Skipping {target_file} (no </h3> tag found for status insertion)")
return False
before, after = content[:h3_close], content[h3_close:]
content = f"{before}{indent(STATUS_SPAN_TEMPLATE, ' ')}{after}"
@@ -918,12 +936,10 @@ def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool:
# SHA-256 of old vite-plugin-fastapi.js (before auto-upgrade marker was added)
# with module name replaced by MODULE_NAME in the outDir path
_OLD_VITE_PLUGIN_SHA256 = (
"93713e879c15a25c750a70ce1de684adeaf11b0c723c38da56e5e7ba207f6632"
)
_OLD_VITE_PLUGIN_SHA256 = "93713e879c15a25c750a70ce1de684adeaf11b0c723c38da56e5e7ba207f6632"
def _upgrade_old_vite_plugin(path: Path, module_name: str, dry: bool = False) -> None:
def _upgrade_old_vite_plugin(path: Path, module_name: str, *, dry: bool = False) -> None:
"""Remove old vite-plugin-fastapi.js that lacks auto-upgrade marker.
Old versions didn't have the upgrade marker, so write_file skips them as
@@ -935,7 +951,6 @@ def _upgrade_old_vite_plugin(path: Path, module_name: str, dry: bool = False) ->
content = path.read_text("UTF-8")
if UPGRADE_MARKER in content:
return # Already new format, write_file handles it
import hashlib
normalized = content.replace(
f"../{module_name}/frontend-build", "../MODULE_NAME/frontend-build"
@@ -957,6 +972,7 @@ _new_files_written: list[tuple[Path, Path]] = []
def write_file(
path: Path,
content: str,
*,
overwrite: bool = True,
dry: bool = False,
executable: bool = False,
@@ -994,7 +1010,7 @@ def write_file(
if fallback_path is not None:
# Write to fallback path instead
return _write_fallback_file(
path, fallback_path, content, dry, executable
path, fallback_path, content, dry=dry, executable=executable
)
print(f"️ Skipping {path} (customized by user)")
return False
@@ -1017,6 +1033,7 @@ def _write_fallback_file(
original_path: Path,
fallback_path: Path,
content: str,
*,
dry: bool,
executable: bool,
) -> bool:
@@ -1074,8 +1091,6 @@ def merge_pyproject(
if "requires-python" in data["project"]:
req = data["project"]["requires-python"]
# Parse minimum version from strings like ">=3.10" or ">=3.9,<4"
import re
match = re.search(r">=\s*(\d+)\.(\d+)", req)
if match:
major, minor = int(match.group(1)), int(match.group(2))
@@ -1121,9 +1136,12 @@ def merge_pyproject(
if "custom" not in hatch_build["targets"]["sdist"]["hooks"]:
hatch_build["targets"]["sdist"]["hooks"]["custom"] = tomlkit.table()
if "path" not in hatch_build["targets"]["sdist"]["hooks"]["custom"]:
hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] = hatch_additions[
"targets"
]["sdist"]["hooks"]["custom"]["path"]
hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] = hatch_additions["targets"][
"sdist"
]["hooks"]["custom"]["path"]
elif hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] == OLD_BUILD_HOOK_PATH:
# Migrate old build hook path to new name
hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] = NEW_BUILD_HOOK_PATH
return data
@@ -1167,7 +1185,7 @@ def find_js_runtime() -> tuple[str, str] | None:
return None
def ensure_python_project(project_dir: Path, dry: bool = False) -> bool:
def ensure_python_project(project_dir: Path, *, dry: bool = False) -> bool:
"""Ensure pyproject.toml exists, run uv init if needed."""
pyproject = project_dir / "pyproject.toml"
if pyproject.exists():
@@ -1179,7 +1197,7 @@ def ensure_python_project(project_dir: Path, dry: bool = False) -> bool:
print("📦 No pyproject.toml found, initializing Python project...")
print(">>> uv init")
result = subprocess.run(["uv", "init", str(project_dir)], check=False)
result = subprocess.run(["uv", "init", str(project_dir)], check=False) # noqa: S603, S607
if result.returncode != 0:
print("❌ uv init failed")
return False
@@ -1193,7 +1211,7 @@ 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, *, 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"
@@ -1225,11 +1243,7 @@ def ensure_frontend(project_dir: Path, dry: bool = False) -> bool:
print(f">>> {' '.join(create_cmd)}")
print("(Follow the prompts to configure your Vue app)")
print()
result = subprocess.run(
create_cmd,
cwd=project_dir,
check=False,
)
result = subprocess.run(create_cmd, cwd=project_dir, check=False) # noqa: S603
if result.returncode != 0:
print("❌ create-vue failed")
return False
@@ -1251,10 +1265,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
project_path = Path(args.project_dir)
# Handle both "." and "/path/to/project"
if project_path.is_absolute():
project_dir = project_path
else:
project_dir = Path.cwd() / project_path
project_dir = project_path if project_path.is_absolute() else Path.cwd() / project_path
project_dir = project_dir.resolve()
dry = args.dry
@@ -1274,11 +1285,11 @@ 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):
if not ensure_frontend(project_dir, dry=dry):
return 1
# Step 2: Ensure Python project exists
if not ensure_python_project(project_dir, dry):
if not ensure_python_project(project_dir, dry=dry):
return 1
# Detect module name
@@ -1319,17 +1330,34 @@ def cmd_setup(args: argparse.Namespace) -> int:
default_port, vite_port, dev_port = DEFAULT_PORTS
ports_note = "(--ports to override)"
print(
f"📡 Ports: default={default_port}, vite={vite_port}, dev={dev_port} {ports_note}"
)
print(f"📡 Ports: default={default_port}, vite={vite_port}, dev={dev_port} {ports_note}")
# Determine health path configuration
# Priority: --health argument > existing project value > default
if args.health is not None:
health = args.health
health_note = "(--health)" if health else "(disabled via --health)"
else:
existing_health = extract_existing_health(project_dir)
if existing_health is not _HEALTH_NOT_FOUND:
# Explicitly configured (path string, possibly empty to disable)
health = existing_health
health_note = "(kept for upgrade)"
else:
# Not found - use default
health = DEFAULT_HEALTH
health_note = "(--health to override)"
if health:
print(f"🏥 Health check: {health} {health_note}")
else:
print(f"🏥 Health check: disabled {health_note}")
# Title for templates
project_title = module_name.replace("_", " ").title()
# Find existing FastAPI app
app_info = (
find_fastapi_app(module_dir, project_dir) if module_dir.exists() else None
)
app_info = find_fastapi_app(module_dir, project_dir) if module_dir.exists() else None
# Template variables
tpl_vars = {
@@ -1338,6 +1366,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
"TEMPLATE_DEFAULT_PORT": str(default_port),
"TEMPLATE_VITE_PORT": str(vite_port),
"TEMPLATE_DEV_PORT": str(dev_port),
"TEMPLATE_HEALTH": f'"{health}"',
"ENVPREFIX": module_name.upper(),
"PROJECT_CLI": module_name,
"MAIN_MODULE": main_module_path,
@@ -1347,7 +1376,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:
@@ -1376,9 +1405,27 @@ def cmd_setup(args: argparse.Namespace) -> int:
obsolete_util.unlink()
print(f"🗑️ Removed obsolete {obsolete_util}")
# Remove obsolete build-frontend.py if present (renamed to buildhook.py)
obsolete_build_hook = fastapi_vue_scripts / "build-frontend.py"
if obsolete_build_hook.exists():
if dry:
print(f"🗑️ Would remove obsolete {obsolete_build_hook}")
else:
obsolete_build_hook.unlink()
print(f"🗑️ Removed obsolete {obsolete_build_hook}")
# Remove obsolete __init__.py if present (folder is no longer a module)
obsolete_init = fastapi_vue_scripts / "__init__.py"
if obsolete_init.exists():
if dry:
print(f"🗑️ Would remove obsolete {obsolete_init}")
else:
obsolete_init.unlink()
print(f"🗑️ Removed obsolete {obsolete_init}")
# Copy all files from the template's fastapi-vue folder
template_fastapi_vue_dir = TEMPLATE_DIR / "scripts" / "fastapi-vue"
for template_file in template_fastapi_vue_dir.iterdir():
for template_file in sorted(template_fastapi_vue_dir.iterdir()):
if template_file.is_file():
dest_path = fastapi_vue_scripts / template_file.name
template = template_file.read_text("UTF-8")
@@ -1483,7 +1530,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
# Install the vite plugin file (always update)
plugin_file = frontend_dir / "vite-plugin-fastapi.js"
# Upgrade old plugin versions that lack the auto-upgrade marker
_upgrade_old_vite_plugin(plugin_file, module_name, dry)
_upgrade_old_vite_plugin(plugin_file, module_name, dry=dry)
template = load_template("frontend/vite-plugin-fastapi.js")
content = render_template(template, **tpl_vars)
write_file(plugin_file, content, overwrite=True, dry=dry)
@@ -1493,15 +1540,15 @@ def cmd_setup(args: argparse.Namespace) -> int:
vite_config_js = frontend_dir / "vite.config.js"
if vite_config_ts.exists():
patch_vite_config(vite_config_ts, module_name, dry)
patch_vite_config(vite_config_ts, dry=dry)
elif vite_config_js.exists():
patch_vite_config(vite_config_js, module_name, dry)
patch_vite_config(vite_config_js, dry=dry)
else:
print("⚠️ No vite.config.ts or vite.config.js found in frontend/")
print(" Run create-vue first to generate a Vite config to patch.")
# Patch Vue app with backend health check
patch_frontend_health_check(frontend_dir, dry)
patch_frontend_health_check(frontend_dir, dry=dry)
# === Update pyproject.toml ===
pyproject_path = project_dir / "pyproject.toml"
@@ -1540,9 +1587,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
else:
nl = b"\r\n" if b"\r\n" in gitignore_content else b"\n"
suffix = b"" if gitignore_content.endswith(nl) else nl
gitignore_path.write_bytes(
gitignore_content + suffix + gitignore_entry.encode() + nl
)
gitignore_path.write_bytes(gitignore_content + suffix + gitignore_entry.encode() + nl)
print(f"✅ Added {gitignore_entry} to .gitignore")
elif dry:
print(f"✅ Would create .gitignore with {gitignore_entry}")
@@ -1551,23 +1596,27 @@ 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(["httpx"], cwd=project_dir, group="dev")
uv_add_packages(["fastapi[standard]", fastapi_vue_req], cwd=project_dir)
print()
print_boxed("Setup complete!")
# Show cd command only if project is not in current directory
cd_cmd = "" if project_dir == Path.cwd() else f"cd {project_dir}; "
cd_cmd = "" if project_dir == Path.cwd() else f"cd {project_dir} && "
script_name = module_name.replace("_", "-")
message = SETUP_COMPLETE_MESSAGE.replace("CD_CMD", cd_cmd).replace(
"SCRIPT_NAME", script_name
)
message = SETUP_COMPLETE_MESSAGE.replace("CD_CMD", cd_cmd).replace("SCRIPT_NAME", script_name)
if platform.system() == "Windows":
message = message.replace(" && ", "; ")
print(message)
# Show merge note if any .new.py files were written
@@ -1592,25 +1641,19 @@ def cmd_setup(args: argparse.Namespace) -> int:
def is_uninitialized_folder(path: Path) -> bool:
"""Check if a folder appears to be completely uninitialized."""
return (
not (path / "pyproject.toml").exists() and not (path / "package.json").exists()
)
return not (path / "pyproject.toml").exists() and not (path / "package.json").exists()
def is_already_patched(path: Path) -> bool:
"""Check if a folder has already been patched by fastapi-vue-setup."""
# Check for our scripts directory
if (path / "scripts" / "fastapi-vue").exists():
return True
# Check for vite plugin in frontend
if (path / "frontend" / "vite-plugin-fastapi.js").exists():
return True
return False
# Check for our scripts directory for vite plugin in frontend
scriptdir = path / "scripts" / "fastapi-vue"
viteplugin = path / "frontend" / "vite-plugin-fastapi.js"
return scriptdir.exists() or viteplugin.exists()
def main() -> int:
"""CLI entry point."""
parser = argparse.ArgumentParser(
description=f"fastapi-vue-setup {version} - FastAPI + Vue project setup tool",
formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -1636,8 +1679,11 @@ Examples:
help="Port configuration as comma-separated values (default: 3100,3100,3200)",
)
parser.add_argument(
"--dry", "--dry-run", action="store_true", help="Show what would be done"
"--health",
metavar="PATH",
help='Health check endpoint (disable waiting for backend startup by setting "")',
)
parser.add_argument("--dry", "--dry-run", action="store_true", help="Show what would be done")
args = parser.parse_args()
+20 -2
View File
@@ -14,8 +14,9 @@ dependencies = [
]
[project.urls]
Homepage = "https://git.zi.fi/LeoVasanko/fastapi-vue-setup"
Repository = "https://github.com/LeoVasanko/fastapi-vue-setup"
Homepage = "https://vasanko.com/coders/fastapi-vue"
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"
@@ -31,3 +32,20 @@ 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"]
+4 -11
View File
@@ -10,25 +10,18 @@ DIST = ROOT / "dist"
FASTAPI_VUE = ROOT / "fastapi-vue"
def main():
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,
)
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,
)
subprocess.run(["uv", "build", "--out-dir", str(DIST)], cwd=ROOT, check=True) # noqa: S603, S607
if __name__ == "__main__":
+1
View File
@@ -0,0 +1 @@
"""Backend package with FastAPI application and Vue frontend integration."""
+7 -3
View File
@@ -1,6 +1,9 @@
# auto-upgrade@fastapi-vue-setup - remove this if you modify this file
"""Command-line entry point for running the backend server."""
import argparse
import os
from pathlib import Path
from fastapi_vue import server
@@ -8,7 +11,8 @@ DEFAULT_PORT = TEMPLATE_DEFAULT_PORT
DEVMODE = os.getenv("ENVPREFIX_DEV") == "1"
def main():
def main() -> None:
"""Run the backend server with optional arguments."""
parser = argparse.ArgumentParser(description="Run the MODULE_NAME server.")
parser.add_argument(
"-l",
@@ -17,12 +21,12 @@ def main():
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,
)
+6 -2
View File
@@ -1,3 +1,6 @@
"""FastAPI application module with Vue frontend integration."""
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from pathlib import Path
@@ -10,7 +13,7 @@ frontend = Frontend(Path(__file__).with_name("frontend-build"))
@asynccontextmanager
async def lifespan(app: FastAPI):
async def lifespan(_app: FastAPI) -> AsyncGenerator:
"""Manage app startup and shutdown resources."""
await frontend.load()
yield
@@ -24,7 +27,8 @@ 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():
async def health_check() -> dict:
"""Return backend status for health monitoring."""
return {"status": "ok"}
+2
View File
@@ -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",
+13 -5
View File
@@ -9,9 +9,11 @@ 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 ( # type: ignore
from devutil import (
ProcessGroup,
check_ports_free,
logger,
@@ -22,11 +24,15 @@ from devutil import ( # type: ignore
DEFAULT_VITE_PORT = TEMPLATE_VITE_PORT
DEFAULT_DEV_PORT = TEMPLATE_DEV_PORT
HEALTH = TEMPLATE_HEALTH
async def run_devserver(
listen: str, backend: str, extra_args: list[str] | None = None
listen: str,
backend: str,
extra_args: list[str] | None = None,
) -> None:
"""Start Vite and FastAPI dev servers with hot reload."""
reporoot = Path(__file__).parent.parent
front = reporoot / "frontend"
if not (front / "package.json").exists():
@@ -36,7 +42,7 @@ 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"
@@ -45,11 +51,13 @@ async def run_devserver(
npm_i = await pg.spawn(*npm_install, cwd=front)
await check_ports_free(viteurl, backurl)
await pg.spawn(*MODULE_NAME, *(extra_args or []))
await pg.wait(npm_i, ready(backurl, path="/api/health?from=devserver.py"))
await pg.wait(npm_i, ready(backurl, path=HEALTH))
await pg.spawn(*vite, cwd=front)
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,
@@ -1,15 +1,19 @@
# ruff: noqa: INP001
"""Hatch build hook for building Vue frontend during package build."""
import sys
from pathlib import Path
from hatchling.builders.hooks.plugin.interface import BuildHookInterface # type: ignore
from hatchling.builders.hooks.plugin.interface import BuildHookInterface
sys.path.insert(0, str(Path(__file__).parent))
from buildutil import build
class CustomBuildHook(BuildHookInterface):
def initialize(self, version, build_data):
class CustomBuildHook(BuildHookInterface): # type: ignore[misc]
"""Hatch build hook that builds Vue frontend during package build."""
def initialize(self, version: str, build_data: dict) -> None: # type: ignore[override]
"""Build frontend before package is built."""
super().initialize(version, build_data)
build("frontend")
+98 -58
View File
@@ -1,3 +1,4 @@
# ruff: noqa: INP001
"""Utilities used at build time and in devserver script. No dependencies."""
import logging
@@ -7,6 +8,8 @@ import shutil
import subprocess
from pathlib import Path
MIN_NODE_VERSION = 20
class _PrefixFormatter(logging.Formatter):
"""Formatter that adds prefix based on log level."""
@@ -30,82 +33,119 @@ def _check_node_version(node_path: str) -> None:
Raises RuntimeError if version is too old or cannot be determined.
"""
try:
result = subprocess.run(
[node_path, "--version"], capture_output=True, text=True, check=True
result = subprocess.run( # noqa: S603
[node_path, "--version"],
capture_output=True,
text=True,
check=True,
)
version_str = result.stdout.strip()
# Parse version like "v20.10.0" or "v18.17.1"
match = re.match(r"v(\d+)", version_str)
if match:
major_version = int(match.group(1))
if major_version >= 20:
if major_version >= MIN_NODE_VERSION:
return
raise RuntimeError(
f"Node.js {version_str} found, but v20+ required (install with nvm)"
)
msg = f"Node.js {version_str} found, but v{MIN_NODE_VERSION}+ required"
raise RuntimeError(msg)
except (subprocess.CalledProcessError, FileNotFoundError, ValueError):
pass
raise RuntimeError("Could not determine Node.js version")
msg = "Could not determine Node.js version"
raise RuntimeError(msg)
def _validate_npm_runtime(tool: str) -> bool:
"""Validate npm runtime by checking Node.js version. Returns True if valid."""
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path is None:
return False
try:
_check_node_version(node_path)
except RuntimeError:
return False
return True
def _find_runtime_from_env(options: list[str]) -> tuple[str, str] | None:
"""Find runtime specified by JS_RUNTIME environment variable."""
js_runtime_env = os.environ.get("JS_RUNTIME")
if not js_runtime_env:
return None
js_runtime = js_runtime_env
js_path = Path(js_runtime)
runtime_name = js_path.name
# Map node to npm
if runtime_name == "node":
runtime_name = "npm"
js_runtime = str(js_path.parent / "npm") if js_path.parent.name else "npm"
for option in options:
if option != runtime_name and not runtime_name.startswith(option):
continue
tool = shutil.which(js_runtime)
if tool is None:
msg = f"JS_RUNTIME={js_runtime_env}: {option} not found"
raise RuntimeError(msg)
if option == "npm":
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path is None:
msg = f"JS_RUNTIME={js_runtime_env}: node not found"
raise RuntimeError(msg)
_check_node_version(node_path)
return tool, option
msg = f"JS_RUNTIME={js_runtime_env} not recognized"
raise RuntimeError(msg)
def _auto_detect_runtime(options: list[str]) -> tuple[str, str]:
"""Auto-detect JavaScript runtime from available options."""
node_version_error: RuntimeError | None = None
for option in options:
tool = shutil.which(option)
if not tool:
continue
if option == "npm" and not _validate_npm_runtime(tool):
try:
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path:
_check_node_version(node_path)
except RuntimeError as e:
node_version_error = e
continue
return tool, option
if node_version_error:
raise node_version_error
msg = "Node.js (v20+), Deno or Bun is required but none was found"
raise RuntimeError(msg)
def find_js_runtime() -> tuple[str, str]:
"""Find a JavaScript runtime from JS_RUNTIME env or auto-detect.
Returns (tool_path, tool_name) where tool_name is "deno", "npm", or "bun".
Raises JSRuntimeError if no suitable runtime is found.
Raises RuntimeError if no suitable runtime is found.
"""
options = ["npm", "deno", "bun"]
node_version_error: RuntimeError | None = None
# Check for JS_RUNTIME environment variable
if js_runtime_env := os.environ.get("JS_RUNTIME"):
js_runtime = js_runtime_env
js_path = Path(js_runtime)
runtime_name = js_path.name
# Map node to npm
if runtime_name == "node":
runtime_name = "npm"
js_runtime = str(js_path.parent / "npm") if js_path.parent.name else "npm"
for option in options:
if option == runtime_name or runtime_name.startswith(option):
tool = shutil.which(js_runtime)
if tool is None:
raise RuntimeError(
f"JS_RUNTIME={js_runtime_env}: {option} not found"
)
# Check Node.js version if using npm
if option == "npm":
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path is None:
raise RuntimeError(
f"JS_RUNTIME={js_runtime_env}: node not found"
)
_check_node_version(node_path) # Raises on failure
return tool, option
raise RuntimeError(f"JS_RUNTIME={js_runtime_env} not recognized")
if result := _find_runtime_from_env(options):
return result
# Auto-detect
for option in options:
if tool := shutil.which(option):
# Check Node.js version if using npm
if option == "npm":
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path is None:
continue
try:
_check_node_version(node_path)
except RuntimeError as e:
node_version_error = e
continue # Try next runtime
return tool, option
# No runtime found - provide helpful error
if node_version_error:
raise node_version_error
raise RuntimeError("Node.js (v20+), Deno or Bun is required but none was found")
return _auto_detect_runtime(options)
def find_build_tool():
def find_build_tool() -> tuple[list[str], list[str]]:
"""Find JavaScript runtime and construct install/build commands.
Returns (install_cmd, build_cmd) tuples of command lists.
@@ -143,7 +183,7 @@ def find_dev_tool() -> list[str]:
if name == "bun":
logger.warning(
"Bun has a bug in WS proxying (https://github.com/oven-sh/bun/issues/9882). Consider using npm instead."
"Bun has a WS proxy bug (github.com/oven-sh/bun/issues/9882). Consider npm.",
)
return [tool, *dev_args[name]]
@@ -176,16 +216,16 @@ def build(folder: str = "frontend") -> None:
install_cmd, build_cmd = find_build_tool()
except RuntimeError as e:
logger.warning(e)
raise SystemExit(1)
raise SystemExit(1) from None
def run(cmd):
def run(cmd: list[str]) -> None:
display_cmd = [Path(cmd[0]).stem, *cmd[1:]]
logger.info("### %s", " ".join(display_cmd))
subprocess.run(cmd, check=True, cwd=folder)
subprocess.run(cmd, check=True, cwd=folder) # noqa: S603
try:
run(install_cmd)
logger.info("")
run(build_cmd)
except subprocess.CalledProcessError:
raise SystemExit(1)
raise SystemExit(1) from None
+77 -38
View File
@@ -1,27 +1,33 @@
# ruff: noqa: INP001
"""Utilities meant for devserver script, used only in source repository with dev deps."""
import asyncio
import subprocess
import sys
from collections.abc import Coroutine
from contextlib import suppress
from pathlib import Path
from typing import Any
from typing import TYPE_CHECKING, Any, Self
from urllib.parse import urlsplit
import httpx
from buildutil import find_dev_tool, find_install_tool, logger
from fastapi_vue.hostutil import parse_endpoint
if TYPE_CHECKING:
from collections.abc import Coroutine
class ProcessGroup:
"""Manage async subprocesses with automatic cleanup, like TaskGroup for processes."""
def __init__(self):
def __init__(self) -> None:
"""Initialize empty process tracking."""
self._procs: list[asyncio.subprocess.Process] = []
self._cmds: dict[int, str] = {} # pid -> command name
async def spawn(
self, *cmd: str, cwd: str | None = None
self,
*cmd: str,
cwd: str | None = None,
) -> asyncio.subprocess.Process:
"""Spawn a subprocess and track it."""
cmd_name = Path(cmd[0]).stem
@@ -32,7 +38,8 @@ class ProcessGroup:
return proc
async def wait(
self, *waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]"
self,
*waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]",
) -> None:
"""Wait for processes/coroutines to complete, raise SystemExit on failure."""
@@ -43,8 +50,7 @@ class ProcessGroup:
raise subprocess.CalledProcessError(returncode, cmd_name)
tasks = [
wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w
for w in waitables
wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w for w in waitables
]
try:
await asyncio.gather(*tasks)
@@ -52,14 +58,15 @@ class ProcessGroup:
logger.warning("%s failed with exit status %d", e.cmd, e.returncode)
raise SystemExit(1) from None
async def __aenter__(self):
async def __aenter__(self) -> Self:
"""Enter the async context manager."""
return self
async def __aexit__(self, exc_type, *_):
async def __aexit__(self, exc_type: type[BaseException] | None, *_: object) -> None:
"""Wait for one process to exit, terminate others, then wait for all."""
await self._cleanup(immediate=exc_type is not None)
async def _cleanup(self, immediate: bool = False):
async def _cleanup(self, *, immediate: bool = False) -> None:
running = [p for p in self._procs if p.returncode is None]
if not running:
return
@@ -87,7 +94,7 @@ class ProcessGroup:
asyncio.wait_for(
asyncio.gather(*[p.wait() for p in still_running]),
timeout=10,
)
),
)
except TimeoutError:
for p in self._procs:
@@ -97,43 +104,69 @@ class ProcessGroup:
await p.wait()
async def http_get_server(url: str, timeout: float) -> str | None: # noqa: ASYNC109
"""GET url with plain asyncio streams, return the response Server header.
Returns an empty string when the server responds without a Server header,
and None when the server is unreachable or doesn't answer in time.
"""
parts = urlsplit(url)
host = parts.hostname or "localhost"
port = parts.port or (443 if parts.scheme == "https" else 80)
path = parts.path or "/"
if parts.query:
path += f"?{parts.query}"
try:
async with asyncio.timeout(timeout):
reader, writer = await asyncio.open_connection(host, port)
try:
writer.write(f"GET {path} HTTP/1.0\r\nHost: {host}\r\n\r\n".encode())
await writer.drain()
data = await reader.readuntil(b"\r\n\r\n")
finally:
writer.close()
except (OSError, EOFError, ValueError, TimeoutError):
return None
for line in data.decode("latin-1").split("\r\n"):
if line.lower().startswith("server:"):
return line.split(":", 1)[1].strip()
return ""
async def check_ports_free(*urls: str) -> None:
"""Verify URLs are not responding (ports are free). Raise SystemExit if any respond."""
async def check(client: httpx.AsyncClient, url: str) -> None:
with suppress(httpx.RequestError):
res = await client.get(url, timeout=0.1)
server = res.headers.get("server", "server")
logger.warning("Conflicting %s already running at %s", server, url)
async def check(url: str) -> None:
server = await http_get_server(url, timeout=0.1)
if server is not None:
logger.warning("Conflicting %s already running at %s", server or "server", url)
raise SystemExit(1)
async with httpx.AsyncClient() as client:
await asyncio.gather(*[check(client, url) for url in urls])
await asyncio.gather(*[check(url) for url in urls])
async def ready(url: str, path: str = "") -> None:
async def ready(url: str, path: str = "", max_attempts: int = 50) -> None:
"""Wait for the server to be ready by polling an endpoint.
Use empty path to disable the check and make this return immediately.
Raises SystemExit(1) if server doesn't start in time.
"""
max_attempts = 50
full_url = f"{url}{path}"
if not path:
return
async with httpx.AsyncClient() as client:
for attempt in range(max_attempts):
try:
await client.get(full_url, timeout=1.0)
logger.info("✓ Backend ready!")
return
except httpx.RequestError:
if attempt == max_attempts - 1:
logger.warning("Backend didn't start in time")
raise SystemExit(1)
await asyncio.sleep(0.1)
for attempt in range(max_attempts):
if await http_get_server(f"{url}{path}", timeout=1.0) is not None:
logger.info("✓ Backend ready!")
return
if attempt == max_attempts - 1:
logger.warning("Backend didn't start in time")
raise SystemExit(1)
await asyncio.sleep(0.1)
def setup_vite(
endpoint: str, default_port: int = 5173
endpoint: str,
default_port: int = 5173,
) -> tuple[str, list[str], list[str]]:
"""Parse frontend endpoint and build commands.
@@ -159,7 +192,9 @@ def setup_vite(
def setup_fastapi(
endpoint: str, module: str, default_port: int = 8000
endpoint: str,
module: str,
default_port: int = 8000,
) -> tuple[str, list[str]]:
"""Parse backend endpoint and build uvicorn command.
@@ -174,7 +209,7 @@ def setup_fastapi(
host = endpoints[0]["host"]
port = endpoints[0]["port"]
reload_dir = module.split(".")[0] # Don't reload on frontend changes
reload_dir = module.split(".", maxsplit=1)[0] # Don't reload on frontend changes
cmd = [
sys.executable,
@@ -191,7 +226,9 @@ def setup_fastapi(
def setup_cli(
cli: str, endpoint: str, default_port: int = 8000
cli: str,
endpoint: str,
default_port: int = 8000,
) -> tuple[str, list[str]]:
"""Parse backend endpoint and build CLI command.
@@ -207,5 +244,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