Compare commits

...
9 Commits
Author SHA1 Message Date
LeoVasanko 372d249a85 Fix setup for new uv init layouts; use backend/ package dir for new projects 2026-10-27 22:19:22 +00:00
LeoVasanko a19c9028c5 Detect color support on stderr, matching where logs are written 2026-10-21 19:49:20 +00:00
LeoVasanko 465206067b Expose WebSocket connection id as ws.state.logid on accept
Store the two-digit WS id (colored when stderr supports color, plain
otherwise) in the ASGI scope state so user code can include it in its
own log lines via websocket.state.logid.

Move terminal helpers (use_color, strip_ansi, display width) into a new
dependency-free fastapi_vue.tty module, absorbing termwidth.py, to keep
imports one-directional and avoid the logging/accesslog cycle.
2026-10-21 19:40:24 +00:00
LeoVasanko c869783b0b Build/dev emoji logging improved. Restored old plain lines, using intentionally different emoji than the main package. 2026-09-18 17:36:17 +00:00
LeoVasanko c7577e9ff7 Guard access log lines against unreset injected colors
App-supplied extra fields may carry raw ANSI color codes without a
reset, bleeding into the next line's IP column and following output.
Ensure access log lines begin and end with a reset, adding one only
where missing to avoid duplicate resets on well-formed lines.
2026-09-18 11:07:36 +00:00
LeoVasanko fa287eba20 Make buildutil logging use same emojis as the main fastapi-vue logging. 2026-09-18 02:52:52 +00:00
LeoVasanko 0f5b526df6 Add public setup_logging() for standalone pretty logging
Optionally usable early in CLI mains, devservers and scripts that log
before (or without) server.run().  Shares patch_log_config with the
server path (same log_config overrides), but installs no server-side
patches: Formatter gains an install flag gating the lifespan/error
middleware monkeypatches, and the NullHandler backdoor is skipped.
dev=None follows env.dev for the root level, True/False override it.
Pre-existing loggers survive the dictConfig (disable_existing_loggers
is set False).
2026-09-18 02:49:16 +00:00
LeoVasanko 1d4a0a5422 Add env() config binding with teleport() for CLI-to-worker config passing 2026-09-17 02:54:08 +00:00
LeoVasanko 69a1ba3239 Clean up README 2026-09-17 00:25:53 +00:00
15 changed files with 498 additions and 114 deletions

No files matched your search

+43 -9
View File
@@ -77,21 +77,55 @@ A startup box with the app name, version and connect URL is printed before servi
Other arguments are generally passed to `uvicorn.run`, although some like `log_config` receive our modifications. Other arguments are generally passed to `uvicorn.run`, although some like `log_config` receive our modifications.
> As a deployment option, environment `FORWARDED_ALLOW_IPS` controls `X-Forwarded` trusted IPs (default: `127.0.0.1,::1` works for typical setups).
### Logging and exceptions ### Logging and exceptions
Pretty logging is configured automatically across the host process and all workers, at INFO in development and WARNING in production, with emoji level prefixes, colored access logs, and tracebacks rendered by [tracerite](https://pypi.org/project/tracerite/). With `FastAPI(debug=True)`, **Internal Server Error** responses use tracerite formatting as well. Pretty logging is configured automatically across the host process and all workers, at INFO in development and WARNING in production, with emoji level prefixes, colored access logs, and tracebacks rendered by [tracerite](https://pypi.org/project/tracerite/). With `FastAPI(debug=True)`, **Internal Server Error** responses use tracerite formatting as well.
Application code can simply use `logging.info()` through `logging.exception()`, or ordinary `logging.getLogger("myapp")` loggers, without setting up logging itself. Set any logger's level when part of the application should be quieter or more verbose, for example `log_config={"loggers": {"myapp": {"level": "DEBUG"}}}`, accepting additions and overrides using [Python's logging configuration schema](https://docs.python.org/3/library/logging.config.html#logging-config-dictschema). Application code can simply use `logging.info()` through `logging.exception()`, or ordinary `logging.getLogger("myapp")` loggers, without setting up logging itself. Set any logger's level when part of the application should be quieter or more verbose, for example `log_config={"loggers": {"myapp": {"level": "DEBUG"}}}`, accepting additions and overrides using [Python's logging configuration schema](https://docs.python.org/3/library/logging.config.html#logging-config-dictschema).
### Environment (fastapi_vue.env) ## Environment
We use environment variables to pass values between program components, from devserver script setting dev mode and telling backend and frontend URLs, to your CLI, which in turn runs the FastAPI app that may also need access to this information. The variables are prefixed by the current application name to avoid conflicts. The CLI entry point should set one like `os.environ["FASTAPI_VUE"] = "MY_APP"`, before using `server.run` Environment variables are used to pass values across process boundaries, where ordinary Python variables cannot be shared. We provide runtime passing mainly intended for dev environment passing into the main application CLI, as well as config passing intended for the CLI to pass things to FastAPI side.
The following properties read the environment and return `None` when variables haven't been set: Set the application prefix before `server.run()`, at top of your CLI main:
- `fastapi_vue.env.prefix` — the prefix itself ```python
- `fastapi_vue.env.dev` — running in development mode, from e.g. `MY_APP_DEV=1` os.environ["FASTAPI_VUE"] = "MY_APP"
- `fastapi_vue.env.vite_url`, `fastapi_vue.env.backend_url` — URLs set by the devserver ```
### Runtime environment
```python
from fastapi_vue import env
env.prefix # application prefix
env.dev # development mode (bool)
env.vite_url # frontend URL (dev)
env.backend_url # backend URL (dev)
```
The three prefixed variables are set by devserver script and can be read anywhere in your application. Unset values return `None`.
### Teleportation
Mainly intended for passing application config from CLI main to all FastAPI workers and through reloader. Put shared configuration in its own module so that any part of your application can import the same `config` variable:
```python
from dataclasses import dataclass
from fastapi_vue import env
@dataclass
class Config:
project: str = "."
read_only: bool = False
config = env(Config)
```
The config values should be set (in CLI main) before teleportation, which occurs in `server.run()` for all registered env objects. Then everyone who imports the object receives those values. Modifications after that point however do not transit to other workers.
Initially the passed in dataclass or msgspec.Struct is constructed with default values to its fields. Any number of env definitions may be added for different things, each getting a prefixed env variable by type name like `MY_APP_CONFIG` above. Beside `server.run`, pass objects to your own processes with `fastapi_vue.teleport()` if needed e.g. from FastAPI app to its workers. The data format in these variables is a JSON object.
### Proxy configuration
You may set `FORWARDED_ALLOW_IPS` to specify which connecting IP addresses are trusted to provide `X-Forwarded-*` headers. This is a server setup option rather than an application setting: the devserver does not set it, and it does not use the application-name prefix. It may therefore be set globally for the whole server. The default `127.0.0.1,::1` works for typical setups where Caddy, Nginx or another frontend server runs on the same machine.
+3 -2
View File
@@ -1,6 +1,7 @@
"""FastAPI Vue integration - serve Vue frontend from FastAPI.""" """FastAPI Vue integration - serve Vue frontend from FastAPI."""
from .environ import env from .environ import env, teleport
from .logging import setup_logging
from .staticfiles import Frontend from .staticfiles import Frontend
__all__ = ["Frontend", "env"] __all__ = ["Frontend", "env", "setup_logging", "teleport"]
+2 -1
View File
@@ -9,7 +9,7 @@ import time
from ipaddress import IPv6Address from ipaddress import IPv6Address
from typing import TYPE_CHECKING, cast from typing import TYPE_CHECKING, cast
from .termwidth import pad_display from .tty import pad_display, use_color
if TYPE_CHECKING: if TYPE_CHECKING:
from uvicorn._types import ( from uvicorn._types import (
@@ -388,6 +388,7 @@ class AccessLogMiddleware:
if message["type"] == "websocket.accept" and not accepted: if message["type"] == "websocket.accept" and not accepted:
accepted = True accepted = True
ws_id = _next_ws_id() ws_id = _next_ws_id()
www_scope["state"]["logid"] = f"{_WS_OPEN}{ws_id}{_RESET}" if use_color() else ws_id
fields = _ws_open_extra(www_scope, ws_id, origin, _extra()) fields = _ws_open_extra(www_scope, ws_id, origin, _extra())
logger.info(_assemble_access_log(fields), extra=fields) logger.info(_assemble_access_log(fields), extra=fields)
elif message["type"] == "websocket.http.response.start" and not closed: elif message["type"] == "websocket.http.response.start" and not closed:
+83
View File
@@ -4,12 +4,48 @@ The project entry point (generated __main__.py) sets FASTAPI_VUE to the
project-specific prefix (e.g. "MY_APP"). Project settings are then passed project-specific prefix (e.g. "MY_APP"). Project settings are then passed
as "<PREFIX>_*" environment variables; this module is the single place as "<PREFIX>_*" environment variables; this module is the single place
that resolves those names. that resolves those names.
Call env with a dataclass or msgspec.Struct type to bind an object of
that type, e.g. env(Config). It is decoded from "<PREFIX>_<CLASS NAME>"
when set (in spawned server processes) and default-constructed otherwise
(in the CLI entry point, which mutates it before server.run() calls
teleport()).
""" """
import dataclasses
import json
import os import os
import re
from typing import Any, TypeVar
PREFIX_VARIABLE = "FASTAPI_VUE" PREFIX_VARIABLE = "FASTAPI_VUE"
# Names already used by fastapi-vue itself; bindings may not take them
RESERVED_NAMES = frozenset({"DEV", "VITE_URL", "BACKEND_URL"})
T = TypeVar("T")
def _mangle(name: str) -> str:
"""Class name to env name: underscores normalized, uppercased."""
return re.sub(r"_+", "_", name).strip("_").upper()
def _encode(obj: Any) -> str: # noqa: ANN401
if hasattr(type(obj), "__struct_fields__"):
import msgspec # noqa: PLC0415
return msgspec.json.encode(obj).decode()
return json.dumps(dataclasses.asdict(obj))
def _decode(raw: str, type_: type[T]) -> T:
if hasattr(type_, "__struct_fields__"):
import msgspec # noqa: PLC0415
return msgspec.json.decode(raw, type=type_)
return type_(**json.loads(raw))
class _Env: class _Env:
"""Lazy accessors for the project's "<PREFIX>_*" environment variables. """Lazy accessors for the project's "<PREFIX>_*" environment variables.
@@ -18,6 +54,37 @@ class _Env:
or the variable itself is not set. or the variable itself is not set.
""" """
def __init__(self) -> None:
self._bindings: dict[str, tuple[type, Any]] = {}
def __call__(self, type_: type[T], *, name: str | None = None) -> T:
"""Bind and return an object of the given type.
The type must be a dataclass or msgspec.Struct with defaults for
all fields. Decoded from the "<PREFIX>_<NAME>" variable when set,
default-constructed otherwise. The variable name is derived from
the class name unless overridden with name=. Repeated calls with
the same type return the same object; conflicting names raise
KeyError.
"""
var = name if name is not None else _mangle(type_.__name__)
if var in RESERVED_NAMES:
msg = f"{var} is reserved for fastapi-vue itself"
raise KeyError(msg)
if bound := self._bindings.get(var):
bound_type, obj = bound
if bound_type is not type_:
msg = f"{var} is already bound to {bound_type}"
raise KeyError(msg)
return obj
if not (hasattr(type_, "__struct_fields__") or dataclasses.is_dataclass(type_)):
msg = f"{type_} must be a dataclass or msgspec.Struct"
raise TypeError(msg)
raw = self._get(var)
obj = _decode(raw, type_) if raw else type_()
self._bindings[var] = (type_, obj)
return obj
@property @property
def prefix(self) -> str | None: def prefix(self) -> str | None:
"""Return the project prefix from the FASTAPI_VUE environment variable.""" """Return the project prefix from the FASTAPI_VUE environment variable."""
@@ -44,3 +111,19 @@ class _Env:
env = _Env() env = _Env()
def teleport() -> None:
"""Serialize objects bound via env() into environment variables.
Called by server.run() before spawning workers, so mutations made in
the CLI entry point propagate to them. Call directly only when spawning
server processes by other means.
"""
if not env._bindings: # noqa: SLF001
return
if not (prefix := env.prefix):
msg = f"{PREFIX_VARIABLE} is not set; cannot teleport bound objects"
raise RuntimeError(msg)
for var, (_, obj) in env._bindings.items(): # noqa: SLF001
os.environ[f"{prefix}_{var}"] = _encode(obj)
+59 -40
View File
@@ -10,9 +10,6 @@ from __future__ import annotations
import io import io
import logging import logging
import os
import re
import sys
from contextlib import suppress from contextlib import suppress
from copy import deepcopy from copy import deepcopy
from typing import TYPE_CHECKING, Literal from typing import TYPE_CHECKING, Literal
@@ -32,34 +29,15 @@ if TYPE_CHECKING:
from .accesslog import AccessLogMiddleware from .accesslog import AccessLogMiddleware
from .environ import env from .environ import env
from .tty import strip_ansi, use_color
ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m") RESET = "\033[0m"
ACCESS_LOG_FMT = "%(client)s %(status)s %(method)s %(host)s%(path)s %(extra)s%(timing)s" ACCESS_LOG_FMT = "%(client)s %(status)s %(method)s %(host)s%(path)s %(extra)s%(timing)s"
ACCESS_LOGGER = "fastapi_vue.access" ACCESS_LOGGER = "fastapi_vue.access"
def strip_ansi(text: str) -> str:
"""Remove ANSI escape codes from text."""
return ANSI_ESCAPE_RE.sub("", text)
def use_color(stream: io.TextIOBase = sys.stderr) -> bool:
"""Test if the stream supports color codes."""
if os.environ.get("NO_COLOR"): # Non empty means no (no-color.org)
return False
if os.environ.get("FORCE_COLOR", "") not in {"", "0"}: # force-color.org, node
return True
if hasattr(stream, "isatty") and stream.isatty():
return True
with suppress(KeyError, ValueError, OSError): # Journald does color (-ocat)
dev, ino = map(int, os.environ["JOURNAL_STREAM"].split(":", 1))
st = os.fstat(stream.fileno())
return st.st_dev == dev and st.st_ino == ino
return False
_LEVEL_EMOJI = { _LEVEL_EMOJI = {
logging.DEBUG: "🐛", logging.DEBUG: "🐛",
logging.INFO: "🔷", logging.INFO: "🔷",
@@ -98,20 +76,22 @@ class Formatter(logging.Formatter):
use_colors: bool | None = None, # noqa: FBT001 # mirrors logging.Formatter use_colors: bool | None = None, # noqa: FBT001 # mirrors logging.Formatter
*, *,
access: bool = False, access: bool = False,
install: bool = True,
) -> None: ) -> None:
"""Load tracerite, optionally install the access log, detect color support.""" """Load tracerite, optionally install server patches and access log."""
tracerite.load() tracerite.load()
tracerite.load_suppressions( tracerite.load_suppressions(
extra={"starlette.routing": "until", "fastapi.routing": "until"} extra={"starlette.routing": "until", "fastapi.routing": "until"}
) )
patch_lifespan_logging() if install:
patch_server_error_middleware() patch_lifespan_logging()
patch_server_error_middleware()
if access: if access:
install_access_log() install_access_log()
if use_colors in (True, False): if use_colors in (True, False):
self.use_colors = use_colors self.use_colors = use_colors
else: else:
self.use_colors = use_color(sys.stdout) self.use_colors = use_color()
super().__init__(fmt=fmt, datefmt=datefmt, style=style) super().__init__(fmt=fmt, datefmt=datefmt, style=style)
def formatMessage(self, record: logging.LogRecord) -> str: # noqa: N802 def formatMessage(self, record: logging.LogRecord) -> str: # noqa: N802
@@ -120,7 +100,14 @@ class Formatter(logging.Formatter):
return _level_prefix(record) + record.getMessage() return _level_prefix(record) + record.getMessage()
formatted = super().formatMessage(record) formatted = super().formatMessage(record)
if not self.use_colors: if not self.use_colors:
formatted = strip_ansi(formatted) return strip_ansi(formatted)
# Guard against app-supplied fields (``extra``) carrying raw color
# codes without a reset: ensure the line begins and ends with a
# reset, but only add one where it's missing to avoid duplicates.
if not formatted.startswith(RESET):
formatted = RESET + formatted
if not formatted.endswith(RESET):
formatted += RESET
return formatted return formatted
@@ -309,7 +296,35 @@ def _merge_log_config(base: dict, overlay: dict) -> dict:
return base return base
def patch_log_config(log_config, *, access_log: bool = True): # noqa: ANN001, ANN201 def setup_logging(*, log_config: dict | None = None, dev: bool | None = None) -> None:
"""Set up pretty logging standalone, outside of ``server.run()``.
Optional helper for CLI mains, devservers and scripts that log before
(or without) starting the server. Loads tracerite directly and applies
the same patching as the server path (see ``patch_log_config``, a
private helper) with ``logging.config.dictConfig``: partial dicts merge
over uvicorn's default config, so only customizations are needed. The
server-side patches (error middleware, access log) are not installed —
there is no server here. The root logger level is INFO with *dev*
true, WARNING otherwise; *dev* of None follows ``env.dev``. An
explicit level in *log_config* always wins::
import fastapi_vue
fastapi_vue.setup_logging(log_config={"loggers": {"myapp": {"level": "DEBUG"}}})
"""
if log_config is not None and not isinstance(log_config, dict):
msg = f"setup_logging requires a dict log_config, got {type(log_config).__name__}"
raise TypeError(msg)
import logging.config
tracerite.load()
config = patch_log_config(log_config or {}, access_log=False, install=False, dev=dev)
# Standalone setup must not disable loggers created before this call.
config.setdefault("disable_existing_loggers", False)
logging.config.dictConfig(config)
def patch_log_config(log_config, *, access_log: bool = True, install: bool = True, dev: bool | None = None): # noqa: ANN001, ANN201
"""Patch a uvicorn log_config dict for our logging, best-effort. """Patch a uvicorn log_config dict for our logging, best-effort.
A dict without a ``version`` key is treated as a partial config: it is A dict without a ``version`` key is treated as a partial config: it is
@@ -330,9 +345,11 @@ def patch_log_config(log_config, *, access_log: bool = True): # noqa: ANN001, A
``watchfiles.main`` logger is lifted to WARNING so its INFO "N changes ``watchfiles.main`` logger is lifted to WARNING so its INFO "N changes
detected" line is dropped while the WARNING "Reloading..." line (logged detected" line is dropped while the WARNING "Reloading..." line (logged
to ``uvicorn.error``) still shows; a user-supplied level wins. to ``uvicorn.error``) still shows; a user-supplied level wins.
With ``access_log``, additionally rewires the ``access`` formatter to With ``install=False`` (standalone use via ``setup_logging``) the
our Formatter and attaches its handler to our ``fastapi_vue.access`` NullHandler backdoor and the server-side patches in Formatter are
logger. We must not skipped. 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 attach handlers to ``uvicorn.access``: uvicorn gates its own
protocol-level access logging on ``uvicorn.access.hasHandlers()``. protocol-level access logging on ``uvicorn.access.hasHandlers()``.
""" """
@@ -342,12 +359,13 @@ def patch_log_config(log_config, *, access_log: bool = True): # noqa: ANN001, A
if "version" not in config: if "version" not in config:
config = _merge_log_config(deepcopy(LOGGING_CONFIG), config) config = _merge_log_config(deepcopy(LOGGING_CONFIG), config)
with suppress(Exception): if install:
config["formatters"]["fastapi_vue"] = {"()": "fastapi_vue.logging.Formatter"} with suppress(Exception):
config["handlers"]["fastapi_vue"] = { config["formatters"]["fastapi_vue"] = {"()": "fastapi_vue.logging.Formatter"}
"class": "logging.NullHandler", config["handlers"]["fastapi_vue"] = {
"formatter": "fastapi_vue", "class": "logging.NullHandler",
} "formatter": "fastapi_vue",
}
with suppress(Exception): with suppress(Exception):
filters = config.setdefault("filters", {}) filters = config.setdefault("filters", {})
@@ -367,6 +385,7 @@ def patch_log_config(log_config, *, access_log: bool = True): # noqa: ANN001, A
"()": "fastapi_vue.logging.Formatter", "()": "fastapi_vue.logging.Formatter",
"fmt": "%(message)s", "fmt": "%(message)s",
"use_colors": None, "use_colors": None,
"install": install,
} }
# uvicorn's default config leaves the root logger handlerless, eating # uvicorn's default config leaves the root logger handlerless, eating
@@ -375,7 +394,7 @@ def patch_log_config(log_config, *, access_log: bool = True): # noqa: ANN001, A
# with Python's default; dev keeps INFO. Subloggers can override. # with Python's default; dev keeps INFO. Subloggers can override.
with suppress(Exception): with suppress(Exception):
root = config.setdefault("root", {}) root = config.setdefault("root", {})
root.setdefault("level", "INFO" if env.dev else "WARNING") root.setdefault("level", "INFO" if (env.dev if dev is None else dev) else "WARNING")
if "default" in config.get("handlers", {}): if "default" in config.get("handlers", {}):
root_handlers = root.setdefault("handlers", []) root_handlers = root.setdefault("handlers", [])
if "default" not in root_handlers: if "default" not in root_handlers:
+4 -2
View File
@@ -15,16 +15,16 @@ from uvicorn import Config, Server
from uvicorn.main import STARTUP_FAILURE from uvicorn.main import STARTUP_FAILURE
from uvicorn.supervisors import ChangeReload, Multiprocess from uvicorn.supervisors import ChangeReload, Multiprocess
from .environ import env from .environ import env, teleport
from .hostutil import parse_endpoints from .hostutil import parse_endpoints
from .logging import ( from .logging import (
install_access_log, install_access_log,
patch_lifespan_logging, patch_lifespan_logging,
patch_log_config, patch_log_config,
patch_server_error_middleware, patch_server_error_middleware,
use_color,
) )
from .startupbox import print_box from .startupbox import print_box
from .tty import use_color
tracerite.load() # Early load on CLI load (import server); uvicorn workers reload via log config tracerite.load() # Early load on CLI load (import server); uvicorn workers reload via log config
@@ -148,6 +148,8 @@ def run( # noqa: PLR0913
msg = "No endpoints to serve; check listen configuration" msg = "No endpoints to serve; check listen configuration"
raise ValueError(msg) raise ValueError(msg)
teleport() # Serialize bound objects before spawning workers
if startup_box: if startup_box:
print_startup_box(startup_box, app, endpoints) print_startup_box(startup_box, app, endpoints)
+2 -2
View File
@@ -2,14 +2,14 @@
import sys import sys
from .termwidth import display_width, pad_display from .tty import display_width, pad_display
def print_box(text: str) -> None: def print_box(text: str) -> None:
"""Print text in a rounded unicode box sized to its longest line. """Print text in a rounded unicode box sized to its longest line.
The text is split on newlines as-is (not trimmed). Padding accounts The text is split on newlines as-is (not trimmed). Padding accounts
for ANSI colors and unicode display width (see termwidth.display_width). for ANSI colors and unicode display width (see tty.display_width).
""" """
lines = text.split("\n") lines = text.split("\n")
width = max(map(display_width, lines)) width = max(map(display_width, lines))
@@ -1,11 +1,37 @@
"""Terminal display width calculation for unicode and ANSI-colored text.""" """Terminal color detection, ANSI helpers, and display width calculation."""
from __future__ import annotations
import io
import os
import re import re
import sys
import unicodedata import unicodedata
from contextlib import suppress
ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;:]*[A-Za-z]") ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;:]*[A-Za-z]")
def strip_ansi(text: str) -> str:
"""Remove ANSI escape codes from text."""
return ANSI_ESCAPE_RE.sub("", text)
def use_color(stream: io.TextIOBase = sys.stderr) -> bool:
"""Test if the stream supports color codes."""
if os.environ.get("NO_COLOR"): # Non empty means no (no-color.org)
return False
if os.environ.get("FORCE_COLOR", "") not in {"", "0"}: # force-color.org, node
return True
if hasattr(stream, "isatty") and stream.isatty():
return True
with suppress(KeyError, ValueError, OSError): # Journald does color (-ocat)
dev, ino = map(int, os.environ["JOURNAL_STREAM"].split(":", 1))
st = os.fstat(stream.fileno())
return st.st_dev == dev and st.st_ino == ino
return False
def _is_wide(char: str) -> bool: def _is_wide(char: str) -> bool:
"""Return True for characters rendered as two terminal columns.""" """Return True for characters rendered as two terminal columns."""
if unicodedata.east_asian_width(char) in {"F", "W"}: if unicodedata.east_asian_width(char) in {"F", "W"}:
+5
View File
@@ -23,3 +23,8 @@ build-backend = "hatchling.build"
[tool.hatch.version] [tool.hatch.version]
source = "vcs" source = "vcs"
raw-options.root = ".." raw-options.root = ".."
[dependency-groups]
dev = [
"msgspec>=0.21.1",
]
+126
View File
@@ -0,0 +1,126 @@
"""Tests for env() object binding and teleport()."""
import dataclasses
import json
import os
from collections.abc import Iterator
import msgspec
import pytest
from fastapi_vue import env, teleport
from fastapi_vue.environ import PREFIX_VARIABLE, _Env
@pytest.fixture(autouse=True)
def _clean_bindings(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
env._bindings.clear() # noqa: SLF001
monkeypatch.setenv(PREFIX_VARIABLE, "TEST")
yield
env._bindings.clear() # noqa: SLF001
class StructConfig(msgspec.Struct):
"""msgspec struct config."""
host: str = "localhost"
port: int = 8000
@dataclasses.dataclass
class DataclassConfig:
"""Dataclass config."""
host: str = "localhost"
port: int = 8000
def test_default_construction_and_identity() -> None:
"""Unset env: default-constructed; repeated binds return the same object."""
config = env(StructConfig)
assert config.host == "localhost"
assert env(StructConfig) is config
def test_name_mangling() -> None:
"""Class names uppercase as-is; repeated/edge underscores normalize."""
env(StructConfig)
teleport()
assert json.loads(os.environ["TEST_STRUCTCONFIG"])["port"] == 8000
@dataclasses.dataclass
class _my__config_: # noqa: N801
x: int = 0
env(_my__config_)
teleport()
assert "TEST_MY_CONFIG" in os.environ
def test_name_override() -> None:
"""name= overrides the mangled class name verbatim."""
env(DataclassConfig, name="SETTINGS")
teleport()
assert "TEST_SETTINGS" in os.environ
def test_name_conflict() -> None:
"""A different type with a colliding env name raises KeyError."""
env(StructConfig)
with pytest.raises(KeyError, match="already bound"):
env(DataclassConfig, name="STRUCTCONFIG")
def test_reserved_names() -> None:
"""Names used by fastapi-vue itself cannot be bound, by mangle or name=."""
with pytest.raises(KeyError, match="reserved"):
env(DataclassConfig, name="DEV")
@dataclasses.dataclass
class Dev:
x: int = 0
with pytest.raises(KeyError, match="reserved"):
env(Dev)
def test_unsupported_type() -> None:
"""Only dataclasses and msgspec.Structs can be bound."""
with pytest.raises(TypeError, match=r"dataclass or msgspec\.Struct"):
env(dict)
def test_struct_teleport_and_decode() -> None:
"""A mutated struct teleports; a fresh registry decodes it (worker view)."""
config = env(StructConfig)
config.port = 9000
teleport()
decoded = _Env()(StructConfig)
assert decoded == config
assert decoded is not config
def test_dataclass_teleport_and_decode() -> None:
"""Dataclasses round-trip through the environment via stdlib json."""
config = env(DataclassConfig)
config.host = "example.com"
teleport()
assert _Env()(DataclassConfig) == config
def test_decode_on_bind_when_set(monkeypatch: pytest.MonkeyPatch) -> None:
"""An already-set variable is decoded at binding time."""
monkeypatch.setenv("TEST_STRUCTCONFIG", '{"host": "example.com", "port": 1}')
assert env(StructConfig).host == "example.com"
def test_teleport_without_prefix(monkeypatch: pytest.MonkeyPatch) -> None:
"""teleport() with bindings but no FASTAPI_VUE prefix fails loudly."""
monkeypatch.delenv(PREFIX_VARIABLE)
env(StructConfig)
with pytest.raises(RuntimeError, match=PREFIX_VARIABLE):
teleport()
def test_teleport_noop_without_bindings() -> None:
"""No bindings: teleport() needs no prefix and does nothing."""
teleport()
+126 -51
View File
@@ -138,18 +138,23 @@ DEFAULT_HEALTH = "/api/health?from=devserver.py"
UPGRADE_MARKER = "auto-upgrade@fastapi-vue-setup" UPGRADE_MARKER = "auto-upgrade@fastapi-vue-setup"
# pyproject.toml additions for patched projects # pyproject.toml additions for patched projects
# packages/only-include use the package path relative to the project root
# (MODULE_NAME placeholder), so both flat and nested layouts work; the wheel
# target strips the leading directories, mapping e.g. backend/my_app -> my_app
PYPROJECT_ADDITIONS = { PYPROJECT_ADDITIONS = {
"tool": { "tool": {
"hatch": { "hatch": {
"build": { "build": {
"packages": ["MODULE_NAME"],
"artifacts": ["MODULE_NAME/frontend-build"], "artifacts": ["MODULE_NAME/frontend-build"],
"targets": { "targets": {
"wheel": {
"packages": ["MODULE_NAME"],
},
"sdist": { "sdist": {
"only-include": ["MODULE_NAME", "scripts/fastapi-vue"],
"hooks": {"custom": {"path": "scripts/fastapi-vue/buildhook.py"}}, "hooks": {"custom": {"path": "scripts/fastapi-vue/buildhook.py"}},
} },
}, },
"only-packages": True,
} }
} }
}, },
@@ -368,6 +373,19 @@ def find_module_name(project_dir: Path) -> str | None:
return None return None
def find_module_dir(project_dir: Path, module_name: str) -> Path:
"""Locate the Python package directory.
Supports flat (<module>), src/<module>, and backend/<module> layouts.
For new projects (no package yet), defaults to the backend/ layout.
"""
for sub in (module_name, f"src/{module_name}", f"backend/{module_name}"):
candidate = project_dir / sub
if candidate.is_dir():
return candidate
return project_dir / "backend" / module_name
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. """Find the FastAPI app in a module directory.
@@ -1180,13 +1198,26 @@ def _write_fallback_file(
def merge_pyproject( def merge_pyproject(
data: tomlkit.TOMLDocument, additions: dict, module_name: str data: tomlkit.TOMLDocument, additions: dict, module_rel: str
) -> tomlkit.TOMLDocument: ) -> tomlkit.TOMLDocument:
"""Merge additions into pyproject.toml data, preserving comments and existing values.""" """Merge additions into pyproject.toml data, preserving comments and existing values.
module_rel is the package path relative to the project root (e.g. "backend/my_app").
"""
# Ensure hatchling build system is configured # Ensure hatchling build system is configured
if "build-system" not in data: if "build-system" not in data:
data["build-system"] = tomlkit.table() data["build-system"] = tomlkit.table()
# Replace uv_build with hatchling: the frontend build hook and artifacts
# handling require hatch, which uv_build does not support
if data["build-system"].get("build-backend") == "uv_build":
data["build-system"]["requires"] = ["hatchling"]
data["build-system"]["build-backend"] = "hatchling.build"
uv_tool = data.get("tool", {}).get("uv", {}) if "tool" in data else {}
if "build-backend" in uv_tool:
del data["tool"]["uv"]["build-backend"]
print("ℹ️ Replaced uv_build with hatchling (required for the frontend build hook)")
# Add hatchling to requires if not present, preserving existing requires # Add hatchling to requires if not present, preserving existing requires
if "requires" not in data["build-system"]: if "requires" not in data["build-system"]:
data["build-system"]["requires"] = ["hatchling"] data["build-system"]["requires"] = ["hatchling"]
@@ -1226,38 +1257,45 @@ def merge_pyproject(
hatch_build = data["tool"]["hatch"]["build"] hatch_build = data["tool"]["hatch"]["build"]
hatch_additions = additions["tool"]["hatch"]["build"] hatch_additions = additions["tool"]["hatch"]["build"]
# Set packages if not already set
if "packages" not in hatch_build:
hatch_build["packages"] = [
p.replace("MODULE_NAME", module_name) for p in hatch_additions["packages"]
]
# Set artifacts if not already set # Set artifacts if not already set
if "artifacts" not in hatch_build: if "artifacts" not in hatch_build:
hatch_build["artifacts"] = [ hatch_build["artifacts"] = [
a.replace("MODULE_NAME", module_name) for a in hatch_additions["artifacts"] a.replace("MODULE_NAME", module_rel) for a in hatch_additions["artifacts"]
] ]
# Set only-packages if not already set # Add wheel/sdist targets
if "only-packages" not in hatch_build:
hatch_build["only-packages"] = hatch_additions["only-packages"]
# Add sdist target with custom hook
if "targets" not in hatch_build: if "targets" not in hatch_build:
hatch_build["targets"] = tomlkit.table() hatch_build["targets"] = tomlkit.table()
if "sdist" not in hatch_build["targets"]: targets = hatch_build["targets"]
hatch_build["targets"]["sdist"] = tomlkit.table() target_additions = hatch_additions["targets"]
if "hooks" not in hatch_build["targets"]["sdist"]:
hatch_build["targets"]["sdist"]["hooks"] = tomlkit.table() # Wheel: set packages if neither the wheel target nor the top-level
if "custom" not in hatch_build["targets"]["sdist"]["hooks"]: # hatch build config (older setups) defines them
hatch_build["targets"]["sdist"]["hooks"]["custom"] = tomlkit.table() if "wheel" not in targets:
if "path" not in hatch_build["targets"]["sdist"]["hooks"]["custom"]: targets["wheel"] = tomlkit.table()
hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] = hatch_additions["targets"][ if "packages" not in targets["wheel"] and "packages" not in hatch_build:
"sdist" targets["wheel"]["packages"] = [
]["hooks"]["custom"]["path"] p.replace("MODULE_NAME", module_rel) for p in target_additions["wheel"]["packages"]
elif hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] == OLD_BUILD_HOOK_PATH: ]
# Sdist: only-include (keeps the package directory prefix so wheels built
# from the sdist find the same layout) plus the frontend build hook
if "sdist" not in targets:
targets["sdist"] = tomlkit.table()
sdist = targets["sdist"]
if "only-include" not in sdist:
sdist["only-include"] = [
p.replace("MODULE_NAME", module_rel) for p in target_additions["sdist"]["only-include"]
]
if "hooks" not in sdist:
sdist["hooks"] = tomlkit.table()
if "custom" not in sdist["hooks"]:
sdist["hooks"]["custom"] = tomlkit.table()
if "path" not in sdist["hooks"]["custom"]:
sdist["hooks"]["custom"]["path"] = target_additions["sdist"]["hooks"]["custom"]["path"]
elif sdist["hooks"]["custom"]["path"] == OLD_BUILD_HOOK_PATH:
# Migrate old build hook path to new name # Migrate old build hook path to new name
hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] = NEW_BUILD_HOOK_PATH sdist["hooks"]["custom"]["path"] = NEW_BUILD_HOOK_PATH
return data return data
@@ -1308,17 +1346,20 @@ def ensure_python_project(project_dir: Path, *, dry: bool = False) -> bool:
return True return True
if dry: if dry:
print(f"📦 Would run: uv init {project_dir}") print(f"📦 Would run: uv init --bare --build-backend hatchling {project_dir}")
return True return True
print("📦 No pyproject.toml found, initializing Python project...") print("📦 No pyproject.toml found, initializing Python project...")
print(">>> uv init") print(">>> uv init --bare --build-backend hatchling")
result = subprocess.run(["uv", "init", str(project_dir)], check=False) # noqa: S603, S607 result = subprocess.run( # noqa: S603
["uv", "init", "--bare", "--vcs", "git", "--build-backend", "hatchling", str(project_dir)], # noqa: S607
check=False,
)
if result.returncode != 0: if result.returncode != 0:
print("❌ uv init failed") print("❌ uv init failed")
return False return False
# Remove files created by uv init that we don't need # Remove boilerplate files created by older uv init versions
for filename in ["hello.py", "main.py", ".python-version"]: for filename in ["hello.py", "main.py", ".python-version"]:
filepath = project_dir / filename filepath = project_dir / filename
if filepath.exists(): if filepath.exists():
@@ -1412,6 +1453,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
return 1 return 1
# Step 2: Ensure Python project exists # Step 2: Ensure Python project exists
fresh_project = not (project_dir / "pyproject.toml").exists()
if not ensure_python_project(project_dir, dry=dry): if not ensure_python_project(project_dir, dry=dry):
return 1 return 1
@@ -1424,7 +1466,10 @@ def cmd_setup(args: argparse.Namespace) -> int:
else: else:
print(f"📦 Module: {module_name}") print(f"📦 Module: {module_name}")
module_dir = project_dir / module_name module_dir = find_module_dir(project_dir, module_name)
module_rel = module_dir.relative_to(project_dir).as_posix()
if module_rel != module_name:
print(f"📂 Package layout: {module_rel}/")
scripts_dir = project_dir / "scripts" scripts_dir = project_dir / "scripts"
fastapi_vue_scripts = scripts_dir / "fastapi-vue" fastapi_vue_scripts = scripts_dir / "fastapi-vue"
@@ -1436,8 +1481,9 @@ def cmd_setup(args: argparse.Namespace) -> int:
if existing_cli_module: if existing_cli_module:
print(f"ℹ️ Using existing CLI: {existing_cli_module}") print(f"ℹ️ Using existing CLI: {existing_cli_module}")
# Resolve the CLI main module to a file path # Resolve the CLI main module to a file path (relative to the import root,
main = project_dir / Path(*main_module_path.split(".")).with_suffix(".py") # i.e. the parent of the package directory)
main = module_dir.parent / Path(*main_module_path.split(".")).with_suffix(".py")
# Determine port configuration # Determine port configuration
# Priority: --ports argument > existing project values > defaults # Priority: --ports argument > existing project values > defaults
@@ -1485,6 +1531,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
# Template variables # Template variables
tpl_vars = { tpl_vars = {
"MODULE_NAME": module_name, "MODULE_NAME": module_name,
"MODULE_DIR": module_rel,
"PROJECT_TITLE": project_title, "PROJECT_TITLE": project_title,
"TEMPLATE_DEFAULT_PORT": str(default_port), "TEMPLATE_DEFAULT_PORT": str(default_port),
"TEMPLATE_VITE_PORT": str(vite_port), "TEMPLATE_VITE_PORT": str(vite_port),
@@ -1499,8 +1546,8 @@ def cmd_setup(args: argparse.Namespace) -> int:
app_file, app_var = app_info app_file, app_var = app_info
print(f"📍 Found FastAPI app: {app_var} in {app_file.name}") print(f"📍 Found FastAPI app: {app_var} in {app_file.name}")
tpl_vars["APP_VAR"] = app_var tpl_vars["APP_VAR"] = app_var
# Dotted module path relative to project dir (e.g. "{module_name}.api.main") # Dotted module path (e.g. "{module_name}.api.main")
app_module = ".".join(app_file.relative_to(project_dir).with_suffix("").parts) app_module = ".".join(app_file.relative_to(module_dir.parent).with_suffix("").parts)
tpl_vars["APP_MODULE"] = app_module tpl_vars["APP_MODULE"] = app_module
else: else:
print("📍 No existing FastAPI app found, will create new one") print("📍 No existing FastAPI app found, will create new one")
@@ -1583,12 +1630,15 @@ def cmd_setup(args: argparse.Namespace) -> int:
) )
else: else:
# No app: create full app.py # No app: create full app.py
# Create __init__.py if missing # Create __init__.py if missing (overwrite uv init hello-world boilerplate)
init_file = module_dir / "__init__.py" init_file = module_dir / "__init__.py"
if not init_file.exists(): boilerplate = init_file.exists() and re.search(
r'^def main\(\) -> None:\s*\n\s*print\("Hello from', init_file.read_text("UTF-8")
)
if not init_file.exists() or boilerplate:
template = load_template("backend/__init__.py") template = load_template("backend/__init__.py")
content = render_template(template, **tpl_vars) content = render_template(template, **tpl_vars)
write_file(init_file, content, overwrite=False, dry=dry) write_file(init_file, content, overwrite=False, dry=dry, force=boilerplate)
# Create app.py # Create app.py
app_file_path = module_dir / "app.py" app_file_path = module_dir / "app.py"
@@ -1669,9 +1719,10 @@ def cmd_setup(args: argparse.Namespace) -> int:
plugin_file = frontend_dir / "vite-plugin-fastapi.js" plugin_file = frontend_dir / "vite-plugin-fastapi.js"
# Upgrade old plugin versions that lack the auto-upgrade marker # Upgrade old plugin versions that lack the auto-upgrade marker
_upgrade_old_vite_plugin(plugin_file, module_name, dry=dry) _upgrade_old_vite_plugin(plugin_file, module_name, dry=dry)
template = load_template("frontend/vite-plugin-fastapi.js") for plugin_template in ("vite-plugin-fastapi.js", "vite-plugin-fastapi.d.ts"):
content = render_template(template, **tpl_vars) template = load_template(f"frontend/{plugin_template}")
write_file(plugin_file, content, overwrite=True, dry=dry) content = render_template(template, **tpl_vars)
write_file(frontend_dir / plugin_template, content, overwrite=True, dry=dry)
# Find existing vite config (prefer .ts, fall back to .js) # Find existing vite config (prefer .ts, fall back to .js)
vite_config_ts = frontend_dir / "vite.config.ts" vite_config_ts = frontend_dir / "vite.config.ts"
@@ -1694,13 +1745,29 @@ def cmd_setup(args: argparse.Namespace) -> int:
old_content = pyproject_path.read_text("UTF-8") old_content = pyproject_path.read_text("UTF-8")
data = tomlkit.parse(old_content) data = tomlkit.parse(old_content)
updated = merge_pyproject(data, PYPROJECT_ADDITIONS, module_name) updated = merge_pyproject(data, PYPROJECT_ADDITIONS, module_rel)
# Only add script entry if project doesn't already have scripts # Ensure a script entry pointing at our __main__:main
if "scripts" not in updated["project"] or not updated["project"]["scripts"]: scripts = updated["project"].get("scripts")
main_entry = f"{module_name}.__main__:main"
if not scripts:
updated["project"]["scripts"] = tomlkit.table() updated["project"]["scripts"] = tomlkit.table()
script_name = module_name.replace("_", "-") updated["project"]["scripts"][module_name.replace("_", "-")] = main_entry
updated["project"]["scripts"][script_name] = f"{module_name}.__main__:main" else:
has_our_entry = any(
isinstance(entry, str) and entry.startswith(f"{module_name}.")
for entry in scripts.values()
)
if not has_our_entry:
# Rewrite uv init boilerplate ("my-app = \"my_app:main\"") if present
boilerplate = next(
(name for name, entry in scripts.items() if entry == f"{module_name}:main"),
None,
)
if boilerplate:
scripts[boilerplate] = main_entry
else:
scripts[module_name.replace("_", "-")] = main_entry
# Check if content actually changed # Check if content actually changed
new_content = tomlkit.dumps(updated) new_content = tomlkit.dumps(updated)
@@ -1715,8 +1782,16 @@ def cmd_setup(args: argparse.Namespace) -> int:
# === Update .gitignore === # === Update .gitignore ===
gitignore_path = project_dir / ".gitignore" gitignore_path = project_dir / ".gitignore"
gitignore_entry = f"/{module_name}/frontend-build" gitignore_entry = f"/{module_rel}/frontend-build"
if gitignore_path.exists(): if fresh_project:
# New project: opinionated .gitignore (ignore dotfiles and lockfiles)
content = f"{gitignore_entry}\n.*\n!.gitignore\n*.lock\n"
if dry:
print(f"✅ Would create opinionated .gitignore in {project_dir}")
else:
gitignore_path.write_text(content, "UTF-8", newline="\n")
print("✅ Created .gitignore")
elif gitignore_path.exists():
gitignore_content = gitignore_path.read_bytes() gitignore_content = gitignore_path.read_bytes()
if b"frontend-build" in gitignore_content: if b"frontend-build" in gitignore_content:
print("✔️ .gitignore (frontend-build already ignored)") print("✔️ .gitignore (frontend-build already ignored)")
+6
View File
@@ -0,0 +1,6 @@
// auto-upgrade@fastapi-vue-setup -- remove this if you edit the plugin
import type { Plugin } from 'vite'
declare function fastapiVue(options?: { paths?: string[] }): Plugin
export default fastapiVue
+1 -1
View File
@@ -30,7 +30,7 @@ export default function fastapiVue({ paths = ['/api'] } = {}) {
clearScreen: false, clearScreen: false,
server: { proxy }, server: { proxy },
build: { build: {
outDir: '../MODULE_NAME/frontend-build', outDir: '../MODULE_DIR/frontend-build',
emptyOutDir: true, emptyOutDir: true,
}, },
}), }),
+10 -4
View File
@@ -10,21 +10,27 @@ from pathlib import Path
MIN_NODE_VERSION = 20 MIN_NODE_VERSION = 20
class _Formatter(logging.Formatter):
"""Prefix formatter, intentionally different from fastapi_vue.logging.
class _PrefixFormatter(logging.Formatter): INFO and below pass through unprefixed so messages can use their own
"""Formatter that adds prefix based on log level.""" markings (>>>, ###); WARNING and above get an emoji prefix.
"""
def format(self, record: logging.LogRecord) -> str: def format(self, record: logging.LogRecord) -> str:
if record.levelno >= logging.ERROR:
return f"🛑 {record.getMessage()}"
if record.levelno >= logging.WARNING: if record.levelno >= logging.WARNING:
return f"⚠️ {record.getMessage()}" return f"💣 {record.getMessage()}"
return record.getMessage() return record.getMessage()
_handler = logging.StreamHandler() _handler = logging.StreamHandler()
_handler.setFormatter(_PrefixFormatter()) _handler.setFormatter(_Formatter())
logger = logging.getLogger("fastapi-vue") logger = logging.getLogger("fastapi-vue")
logger.addHandler(_handler) logger.addHandler(_handler)
logger.setLevel(logging.INFO) logger.setLevel(logging.INFO)
logger.propagate = False # own handler; do not double-print via a configured root
def _check_node_version(node_path: str) -> None: def _check_node_version(node_path: str) -> None:
+1 -1
View File
@@ -133,7 +133,7 @@ async def ready(url: str, path: str = "", max_attempts: int = 50) -> None:
for attempt in range(max_attempts): for attempt in range(max_attempts):
if await http_get_server(f"{url}{path}", timeout=1.0) is not None: if await http_get_server(f"{url}{path}", timeout=1.0) is not None:
logger.info("✓ Backend ready!") logger.info("🟢 Backend ready!")
return return
if attempt == max_attempts - 1: if attempt == max_attempts - 1:
logger.error("Backend at %s didn't start in time", url) logger.error("Backend at %s didn't start in time", url)