Compare commits

...
27 Commits
Author SHA1 Message Date
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
LeoVasanko d1e82b5955 Treat dict log_config without a version key as an overlay on uvicorn defaults.
A partial log_config (no "version" key) is deep-merged over uvicorn's
default config before our patching, so users pass only their
customizations, e.g. log_config={"loggers": {"myapp": {"level":
"DEBUG"}}}.  Previously such dicts crashed at startup: dictConfig
requires a version, and our root entry referenced a "default" handler
that might not exist; that reference is now only added when the handler
exists.  The stock default formatter is replaced only when untouched,
so an overlaid fmt survives.  Dicts with version and non-dict configs
behave as before.  Documented in the README logging section, with tests
under fastapi-vue/tests (each patched config validated through
dictConfig).
2026-09-16 03:57:44 +00:00
LeoVasanko 5730e5dd01 Document logging principles under the server section. 2026-09-16 02:52:46 +00:00
LeoVasanko 3a9c5d1674 Set root logger level by dev mode, drop kanta logging config.
The root logger entry patch_log_config adds now uses INFO in dev and
WARNING in production (Python's default), so third-party library INFO
noise stays silent in production while subloggers remain free to
define their own level overrides.  A user-supplied root level still
wins (setdefault).

staticfiles now logs via its own module logger instead of
uvicorn.error: the startup stats line shows in dev (root INFO) and is
hidden in production, and it no longer passes through the
uvicorn-quiet filter that silently ate it.

The kanta logger/handler/formatter block is removed: kanta configures
its own event loggers at import time as of its logging rework, and its
diagnostics follow the root logger like any other library.
2026-09-16 02:22:35 +00:00
LeoVasanko 22d9fe350b Migrate pre-1.6 DEVMODE to fastapi_vue.env with in-place patching, run uv sync after dependency changes. 2026-09-14 04:22:24 +00:00
LeoVasanko 42adf6120d Replace startupbox in code box with an actual screenshot. 2026-09-14 03:38:36 +00:00
LeoVasanko 5ab12187bd Add new screenshot of a simulated My App run. 2026-09-14 03:05:17 +00:00
LeoVasanko 5d1ba2f58c Add README images. 2026-09-14 02:26:50 +00:00
LeoVasanko cf2957ab8d Add images for docs. 2026-09-14 02:19:03 +00:00
LeoVasanko fbfd2ba1bb Rewritten fastapi-vue-setup README. 2026-09-14 01:48:26 +00:00
LeoVasanko ff33df0ebc Rewritten fastapi-vue README. Add listen URLs on startup box default template. 2026-09-13 22:59:48 +00:00
LeoVasanko dec5191157 Add fastapi_vue.env accessor with FASTAPI_VUE prefix 2026-09-13 21:24:51 +00:00
LeoVasanko baab37ae2d Startup box improvements. 2026-09-13 06:04:10 +00:00
LeoVasanko ab2d721723 Devutil ready() now logs and raises RuntimeError on timeout. 2026-09-13 05:36:40 +00:00
LeoVasanko 2d7e119161 server: listen on both loopback families for localhost.
Bind sockets ourselves and pass them to uvicorn (Server.serve(sockets) or the
ChangeReload/Multiprocess supervisors, same API uvicorn.run uses). localhost
expands to 127.0.0.1 and ::1 so clients reach the server regardless of how
localhost resolves; individual bind failures degrade with a warning. Reload
mode now serves all endpoints instead of only the first, multi-endpoint
configs no longer run lifespan twice, and unix sockets are cleaned up on
exit.
2026-09-13 05:36:40 +00:00
LeoVasanko d4f835428b Update vite-plugin-fastapi.js formatting to match modern vue/prettier/biome defaults. 2026-09-13 04:32:02 +00:00
LeoVasanko 1a22ec9dce devutil template made fully compatible and ruff-clean across python 3.11-3.14 target projects. 2026-09-13 04:12:13 +00:00
LeoVasanko 02b3890cd3 Proper error messages on npm and port check failures, avoid leaking asyncio tasks. 2026-09-12 05:50:52 +00:00
LeoVasanko 4c0be1bfa7 Forward extra args to create-vue, e.g. for no-interactive use. 2026-09-12 05:15:50 +00:00
LeoVasanko beca6806ef Tidy up HTTP client; nowadays we have UTF-8 headers. 2026-09-12 04:56:18 +00:00
LeoVasanko c25835f467 Rewritten devutil.ProcessGroup with proper handling of subprocess deaths
- Terminates instantly when vital process dies (e.g. backend doesn't start)
- Derived of TaskGroup, augmenting its functionality with similar semantics
2026-09-12 04:48:03 +00:00
LeoVasanko 38d5402045 Enable colored access log, tracerite and other software when running under systemd/journald which support colors (journalctl -ocat) and can strip them off (any other output mode). Sets environment FORCE_COLOR to full color if conditions are met. This is intended to affect all other modules and child processes alike (set NO_COLOR or FORCE_COLOR to override). 2026-09-03 17:27:48 +00:00
22 changed files with 1032 additions and 293 deletions
+94 -57
View File
@@ -1,96 +1,133 @@
# fastapi-vue-setup ![FastAPI-Vue setup complete](https://raw.githubusercontent.com/LeoVasanko/fastapi-vue-setup/main/docs/banner.webp)
Create or patch a FastAPI + Vue project with an integrated dev/build workflow. # FastAPI-Vue Full Stack Setup
- Development: one command runs Vite + FastAPI (reloads) Build and develop **FastAPI + Vue** as a single project, while keeping production purely Python — **JavaScript tooling is only needed during development!**
- Production: `uv build` bakes the built Vue assets into the Python package (no Node/JS runtime needed to *run* the installed package)
Unlike a template repository or a tutorial, the setup adapts to the application you already have and can keep that integration up to date as the stack evolves. It also fills in the practical gaps around FastAPI itself, providing a production-ready application runtime with a CLI command to start the server with integrated pretty logging and tracebacks — giving your project a running start.
Start a new project with your preferred setup, integrate an existing FastAPI or Vue codebase, or upgrade an already integrated project to the latest simply by running the script.
## Quick start ## Quick start
Install [UV](https://docs.astral.sh/uv/) and any JS runtime (node, deno, or bun). Install [UV](https://docs.astral.sh/uv/getting-started/installation/) and Node ([nvm](https://github.com/nvm-sh/nvm#installing-and-updating)), and create your project:
```
This README uses `my-app` as the example project name:
- project directory: `my-app/`
- Python module: `my_app`
- env prefix: `MY_APP`
- CLI command: `my-app`
Create a new project in `./my-app`:
```sh
uvx fastapi-vue-setup my-app uvx fastapi-vue-setup my-app
``` ```
Once in your source tree, you will typically use `.` for the path. If there is an existing project, `fastapi-vue-setup` will do its best to find and patch a backend module and create or patch a Vue project in `frontend/`. The integration can be upgraded by running a new version of `fastapi-vue-setup` on it, preserving earlier default ports and user customizations. Inside the new project, start the development server with live reloads and debug aids:
```sh
uv run scripts/devserver.py
```
## In your project Or build a release package, and run anywhere:
```sh
uv build # a dist Python package
uvx --with dist/my_app-0.1.0.tar.gz my-app --help
```
️ Everything below is meant to be run within your project source tree. Or install as a proper executable:
```sh
uv tool install dist/my_app-0.1.0.tar.gz
my-app --help
```
The setup creates a CLI entry for your package, so that it becomes a command to run, not a Python module nor `fastapi myapp...`. The CLI main can be customized, although --listen should be kept for devserver compatibility. <img src="https://raw.githubusercontent.com/LeoVasanko/fastapi-vue-setup/main/docs/hello.webp" alt='"You did it" with Vue-FastAPI connection.' width="500">
You can choose the JS runtime with environment `JS_RUNTIME` (e.g. `node`, `deno`, `bun`, or path to one). This is used by the build and the devserver scripts. By default any available runtime on the system is chosen. ## Working in your project
### Setup and upgrades
The setup command handles new projects, existing FastAPI or Vue projects, and projects previously configured by an older version. Use a project directory to create or migrate it, or `.` when already inside the source tree:
```sh
uvx fastapi-vue-setup [project-dir] [options]
```
| Option | Purpose |
| -------------------------- | ------------------------------------------------------------------------------------------------------------ |
| `--module-name NAME` | Override the Python module name, normally detected automatically |
| `--ports BACKEND,VITE,DEV` | Set the production backend, development frontend and development backend ports; defaults to `3100,3100,3200` |
| `--health PATH` | Endpoint used to wait for the development backend to become ready; use `--health=""` to disable the check |
| `--dry`, `--dry-run` | Preview the changes without modifying the project |
| `--version` | Print the setup version |
| `-- ARGS` | Pass the remaining arguments to `create-vue` when a frontend needs to be created non-interactively |
Existing projects are inspected rather than replaced: the Python module, FastAPI application, CLI entry point and Vue project are reused where found, with missing pieces created as needed. Running a newer setup version on the project upgrades the generated integration while retaining configured ports and health checks unless explicitly overridden.
️ Generated files marked `auto-upgrade@fastapi-vue-setup` may be refreshed automatically on later runs. Remove that marker when taking ownership of a generated file; where an updated version is still useful, the setup writes a `.new.py` file for manual merging instead of overwriting your changes. Internal files under `scripts/fastapi-vue/` belong to the setup itself and are updated automatically.
### Main CLI (my-app)
The setup gives your application its own CLI command. This becomes the normal way to start it, rather than invoking via FastAPI CLI or by other means, except perhaps via `uv run` or `uvx` to run the latest version directly and avoid installation completely.
The generated `__main__.py` is yours to customize. Keep its `--listen` option if you want it to remain compatible with the development server that also passes arguments to your CLI entry point.
### Development server (Vite + FastAPI) ### Development server (Vite + FastAPI)
```sh ```sh
uv run scripts/devserver.py [args] scripts/devserver.py [args]
``` ```
Arguments are forwarded to the main CLI, except that `--listen` controls where Vite listens, and `--backend` is passed to main CLI as `--listen`. Windows users have to use `uv run scripts/devserver.py`, while Linux and Mac users can just run the script directly.
This runs the Vite development server and FastAPI together, with reloads on both sides and the environment configured so they can communicate directly. Vite serves the Vue app and proxies specific paths to the FastAPI backend. The paths default to `/api/` only, and can be configured in your `vite.config.js` (or `vite.config.ts` if you chose TypeScript). In dev mode the browser only connects via Vite, and trying to load frontend assets from the backend is blocked.
- `--listen` set Vite listening port
- `--backend` set FastAPI port (forwarded as `--listen`)
- `--help` see help; any other arguments are passed directly to your CLI
JavaScript tooling is used only from the source tree, by the devserver and build commands. An available runtime is selected automatically; set `JS_RUNTIME` to `node`, `deno`, `bun`, or an executable path to choose one explicitly.
### Production ### Production
Build the Python package (this compiles the Vue frontend) and run the production server: Building the Python package also builds the Vue frontend, ensuring the source repository has a fresh build. If you wish to run production mode in your source repo, with a fresh build:
```sh ```
uv build && uv run my-app [args] uv build && uv run my-app [args]
``` ```
Once happy with it, publish the package Publishing alike follows `uv build` and involves either copying the `dist/*.tar.gz` archive to where you need it, or `uv publish` to make it a public release that can be run directly by `uvx my-app`.
```sh ️ Other Python build and installation methods like `pip install` work equally well, we just prefer using UV.
uv build && uv publish
```
Afterwards, you can easily run it anywhere, no JS runtimes required: ## Project Layout
```sh A newly created project typically looks like this:
uvx my-app [args]
```
️ Instead of `uvx` you may consider `uv tool install`, oldskool `pip install` or whatever best suits you.
### Vite plugin
The generated Vite plugin lives in `frontend/vite-plugin-fastapi.js` and defaults to proxying `/api`.
It reads `MY_APP_BACKEND_URL` to know where to proxy; if unset it falls back to your configured default backend port.
## Project layout (typical)
``` ```
my-app/ my-app/
├── frontend/ # Vue app (Vite) ├── frontend/ # Vue source (Node, Vite)
│ ├── src/ │ ├── src/
│ ├── vite-plugin-fastapi.js │ ├── vite-plugin-fastapi.js # Helper plugin
│ ├── vite.config.js # Loads the plugin with app setup
│ └── package.json │ └── package.json
├── my_app/ # Python package ├── my_app/ # Python package
│ ├── __main__.py # CLI entrypoint │ ├── __init__.py
│ ├── __main__.py # Application CLI
│ ├── app.py # FastAPI app │ ├── app.py # FastAPI app
│ └── frontend-build/ # built assets (included in distributions) │ └── frontend-build/ # Generated production frontend
├── pyproject.toml ├── scripts/
└── scripts/ │ ├── devserver.py # Vite + FastAPI development server
── devserver.py # Run Vite and FastAPI together in dev mode ── fastapi-vue/ # Generated build/dev support
└── fastapi-vue/ # Dev utilities (only on the source tree) │ ├── buildhook.py
├── buildhook.py ├── buildutil.py
── buildutil.py ── devutil.py
└── devutil.py └── pyproject.toml
``` ```
## The fastapi-vue runtime module Existing projects retain their own layout wherever possible; this is only the default structure.
The backend runs the FastAPI app and serves the frontend build using the companion package in [fastapi-vue/README.md](fastapi-vue/README.md). Your project will depend on Fastapi and this lightweight module. ## Runtime and build tooling
️ Development functionality is in `scripts/fastapi-vue/` directly in your source tree, and is not to be confused with this runtime module. Only the runtime is installed with your package. There are deliberately two separate pieces to the integration.
The files under `scripts` written by the setup script belong to the source tree. They handle development and package building, and are not part of the Python package.
The installed application instead depends on the lightweight [fastapi_vue](https://pypi.org/project/fastapi-vue/) runtime module. It serves the built frontend and provides the server runner, logging and traceback integration used by the generated application.
This keeps the development tooling where it belongs while leaving the distributed application as a normal, self-contained Python package.
️ The version numbering between the runtime and setup packages is synchronized, and the setup script always bumps the version in `pyproject.toml` to ensure compatible updates.
<img src="https://raw.githubusercontent.com/LeoVasanko/fastapi-vue-setup/main/docs/my-app.webp" alt="My App startup box and log items" width="500">
BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 76 KiB

BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 19 KiB

BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 11 KiB

+80 -17
View File
@@ -1,13 +1,13 @@
# fastapi-vue # FastAPI-Vue Runtime
Runtime helpers for FastAPI + Vite/Vue projects. Runtime utilities for making FastAPI apps standalone, with their own CLI entry point and facilities that make the FastAPI + Vue stack pleasant to use.
## Overview ️ Use [fastapi-vue-setup](https://pypi.org/project/fastapi-vue-setup/) to set up your project. Everything below is configured automatically by it.
This package provides: ## Main Components
- `fastapi_vue.Frontend`: serves built SPA assets (with SPA support, caching, and optional zstd) - **Frontend**: Serves static files with proper caching, compression and SPA support
- `fastapi_vue.server.run`: a small Uvicorn runner with convenient `listen` endpoint parsing - **Server**: Runs the FastAPI app from your own CLI entry point with uvicorn facilities vastly augmented
## Quickstart ## Quickstart
@@ -34,16 +34,16 @@ app = FastAPI(lifespan=lifespan)
frontend.route(app, "/") frontend.route(app, "/")
``` ```
## Frontend If SPA mode is disabled, we only route the paths that actually exist, leaving anything else to your own handlers that come after and may themselves wish to catch all that remains.
`Frontend` serves a directory with: ## Frontend (fastapi_vue.Frontend)
- RAM caching, with zstd compression when smaller than original - Designed to serve at `/`, living together with your other routes
- Browser caching: ETag + Last-Modified, Immutable assets - SPA routing: serves `index.html` for paths not otherwise handled
- Favicon mapping (serve PNG or other images there instead) - RAM caching with zstd compression
- SPA routing (serve browsers index.html at all paths not otherwise handled) - Browser caching with ETag, Last-Modified and immutable assets
Dev-mode behavior with `FastAPI(debug=True)`: requests error HTTP 409 with a message telling you to use the Vite dev server instead. Avoids accidentally using outdated `frontend-build` during development. With `FastAPI(debug=True)`, frontend requests return HTTP 409 with a message directing you to the Vite dev server. This prevents accidentally serving an outdated frontend build during development.
- `directory`: Path on local filesystem - `directory`: Path on local filesystem
- `index`: Index file name (default: `index.html`) - `index`: Index file name (default: `index.html`)
@@ -53,11 +53,13 @@ Dev-mode behavior with `FastAPI(debug=True)`: requests error HTTP 409 with a mes
- `favicon`: Optional path or glob (e.g. `/assets/logo*.png`) - `favicon`: Optional path or glob (e.g. `/assets/logo*.png`)
- `zstdlevel`: Compression level (default: 18) - `zstdlevel`: Compression level (default: 18)
Even when your page has a meta tag giving favicon location, browsers still try loading `/favicon.ico` whenever looking at something else. We find it more convenient to simply serve the image where the browser expects it, with correct MIME type. This also allows having a default favicon for your application that can be easily overriden at the reverse proxy (Caddy, Nginx) to serve the company branding if needed in deployment. Browsers commonly request `/favicon.ico` even when another icon is specified in HTML. The favicon option lets you serve an SVG or PNG there instead. This also provides a convenient application default that a deployment reverse proxy such as Caddy or Nginx can override with company branding.
## Server runner ## Server runner (fastapi_vue.server)
When you need more flexibility than `fastapi` CLI can provide (e.g. CLI arguments to your own program), you may use this convenience to run FastAPI app with Uvicorn startup on given `listen` endpoints. Runs in the same process if possible but delegates to `uvicorn.run()` for auto-reloads and multiple workers. This would typically be called from your CLI main, which can set its own env variables to pass information to the FastAPI instances that run (Python imports only work in same-process mode). When you need more flexibility than the `fastapi` CLI provides—for example, to support arguments in your own CLI—you can use the bundled server runner.
It starts the FastAPI app, running directly in the current process when possible and delegating to Uvicorn supervisors for reloads and multiple workers. The `server.run` is modeled after `uvicorn.run` that you would otherwise have to use to run FastAPI.
```python ```python
from fastapi_vue import server from fastapi_vue import server
@@ -65,4 +67,65 @@ from fastapi_vue import server
server.run("my_app.app:app", listen=["localhost:8000"]) server.run("my_app.app:app", listen=["localhost:8000"])
``` ```
- As a deployment option, environment `FORWARDED_ALLOW_IPS` controls `X-Forwarded` trusted IPs (default: `127.0.0.1,::1`). Endpoints are plain strings: `host:port`, a bare port (localhost only), `:port` (all interfaces), or a unix socket path. Multiple endpoints can be served simultaneously. This also avoids Uvicorn's localhost limitation, where localhost may bind only to either 127.0.0.1 or ::1.
A single `reload` argument replaces Uvicorn's separate reload arguments and may directly specify paths to watch.
A startup box with the app name, version and connect URL is printed before serving. Pass a `startup_box` template (`{name}`, `{version}`, `{listen}`, `{url}`, ...) to customize it, None to disable, or use `server.print_startup_box` on its own.
<img src="https://raw.githubusercontent.com/LeoVasanko/fastapi-vue-setup/main/docs/my-app.webp" alt="My App startup box and log items" width="500">
Other arguments are generally passed to `uvicorn.run`, although some like `log_config` receive our modifications.
### 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.
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
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.
Set the application prefix before `server.run()`, at top of your CLI main:
```python
os.environ["FASTAPI_VUE"] = "MY_APP"
```
### 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 -1
View File
@@ -1,5 +1,7 @@
"""FastAPI Vue integration - serve Vue frontend from FastAPI.""" """FastAPI Vue integration - serve Vue frontend from FastAPI."""
from .environ import env, teleport
from .logging import setup_logging
from .staticfiles import Frontend from .staticfiles import Frontend
__all__ = ["Frontend"] __all__ = ["Frontend", "env", "setup_logging", "teleport"]
+129
View File
@@ -0,0 +1,129 @@
"""Access to the project's fastapi-vue environment variables.
The project entry point (generated __main__.py) sets FASTAPI_VUE to the
project-specific prefix (e.g. "MY_APP"). Project settings are then passed
as "<PREFIX>_*" environment variables; this module is the single place
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 re
from typing import Any, TypeVar
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:
"""Lazy accessors for the project's "<PREFIX>_*" environment variables.
Evaluated on each access. Value accessors return None when FASTAPI_VUE
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
def prefix(self) -> str | None:
"""Return the project prefix from the FASTAPI_VUE environment variable."""
return os.environ.get(PREFIX_VARIABLE) or None
def _get(self, name: str) -> str | None:
prefix = self.prefix
return os.environ.get(f"{prefix}_{name}") if prefix else None
@property
def dev(self) -> bool:
"""Check whether running under the devserver (<PREFIX>_DEV=1)."""
return self._get("DEV") == "1"
@property
def vite_url(self) -> str | None:
"""Return the vite devserver URL (<PREFIX>_VITE_URL), if set."""
return self._get("VITE_URL")
@property
def backend_url(self) -> str | None:
"""Return the backend URL (<PREFIX>_BACKEND_URL), if set."""
return self._get("BACKEND_URL")
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)
+95 -32
View File
@@ -10,6 +10,7 @@ from __future__ import annotations
import io import io
import logging import logging
import os
import re import re
import sys import sys
from contextlib import suppress from contextlib import suppress
@@ -19,7 +20,7 @@ from typing import TYPE_CHECKING, Literal
import tracerite import tracerite
from starlette.middleware.errors import ServerErrorMiddleware from starlette.middleware.errors import ServerErrorMiddleware
from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse, Response from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse, Response
from uvicorn.config import Config from uvicorn.config import LOGGING_CONFIG, Config
from uvicorn.lifespan.on import LifespanOn from uvicorn.lifespan.on import LifespanOn
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -30,9 +31,12 @@ if TYPE_CHECKING:
from uvicorn.lifespan.on import LifespanSendMessage from uvicorn.lifespan.on import LifespanSendMessage
from .accesslog import AccessLogMiddleware from .accesslog import AccessLogMiddleware
from .environ import env
ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m") 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"
@@ -43,6 +47,21 @@ def strip_ansi(text: str) -> str:
return ANSI_ESCAPE_RE.sub("", 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: "🔷",
@@ -81,12 +100,14 @@ 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"}
) )
if install:
patch_lifespan_logging() patch_lifespan_logging()
patch_server_error_middleware() patch_server_error_middleware()
if access: if access:
@@ -94,7 +115,7 @@ class Formatter(logging.Formatter):
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 = sys.stdout.isatty() self.use_colors = use_color(sys.stdout)
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
@@ -103,7 +124,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
@@ -282,13 +310,53 @@ def patch_server_error_middleware() -> None:
ServerErrorMiddleware.error_response = error_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 def _merge_log_config(base: dict, overlay: dict) -> dict:
"""Deep-merge *overlay* onto *base*; dicts merge recursively, others replace."""
for key, value in overlay.items():
if isinstance(value, dict) and isinstance(base.get(key), dict):
_merge_log_config(base[key], value)
else:
base[key] = value
return base
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.
Users presumably base their config on uvicorn's default dict, but any A dict without a ``version`` key is treated as a partial config: it is
shape is tolerated: pieces that do not fit the config's structure are merged over uvicorn's default dict, so only the customizations are
silently skipped. Non-dict configs (e.g. an ini file path) pass through needed (e.g. ``{"loggers": {"kanta": {"level": "DEBUG"}}}``). A dict
untouched. with ``version`` is a complete config used as-is; pieces that do not
fit its structure are silently skipped. Non-dict configs (e.g. an ini
file path) pass through untouched.
Always adds an unreferenced NullHandler whose Formatter instantiation Always adds an unreferenced NullHandler whose Formatter instantiation
loads tracerite in every process uvicorn applies the config in, filters loads tracerite in every process uvicorn applies the config in, filters
@@ -296,20 +364,26 @@ def patch_log_config(log_config, *, access_log: bool = True): # noqa: ANN001, A
routine INFO lines, an emoji-level-prefix Formatter in place of routine INFO lines, an emoji-level-prefix Formatter in place of
uvicorn's stock ``default`` formatter (a user-supplied one wins), a root uvicorn's stock ``default`` formatter (a user-supplied one wins), a root
logger entry so ``logging.info()`` et al. print through the default logger entry so ``logging.info()`` et al. print through the default
handler, and a no-prefix ``kanta`` logger entry (likewise). The handler when one exists, at INFO in dev and WARNING in production
(matching Python's default). The
``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()``.
""" """
if not isinstance(log_config, dict): if not isinstance(log_config, dict):
return log_config return log_config
config = deepcopy(log_config) config = deepcopy(log_config)
if "version" not in config:
config = _merge_log_config(deepcopy(LOGGING_CONFIG), config)
if install:
with suppress(Exception): with suppress(Exception):
config["formatters"]["fastapi_vue"] = {"()": "fastapi_vue.logging.Formatter"} config["formatters"]["fastapi_vue"] = {"()": "fastapi_vue.logging.Formatter"}
config["handlers"]["fastapi_vue"] = { config["handlers"]["fastapi_vue"] = {
@@ -330,18 +404,22 @@ def patch_log_config(log_config, *, access_log: bool = True): # noqa: ANN001, A
# default formatter; a user-supplied default formatter is left alone. # default formatter; a user-supplied default formatter is left alone.
with suppress(Exception): with suppress(Exception):
default = config["formatters"]["default"] default = config["formatters"]["default"]
if default.get("()") in (None, "uvicorn.logging.DefaultFormatter"): if default == LOGGING_CONFIG["formatters"]["default"]:
config["formatters"]["default"] = { config["formatters"]["default"] = {
"()": "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
# logging.info() et al.; route root through uvicorn's default handler. # logging.info() et al.; route root through uvicorn's default handler.
# Level is WARNING in production so third-party loggers stay quiet, as
# 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") root.setdefault("level", "INFO" if (env.dev if dev is None else dev) else "WARNING")
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:
root_handlers.append("default") root_handlers.append("default")
@@ -349,23 +427,8 @@ def patch_log_config(log_config, *, access_log: bool = True): # noqa: ANN001, A
# watchfiles logs "N changes detected" to its own logger at INFO; only the # watchfiles logs "N changes detected" to its own logger at INFO; only the
# WARNING "Reloading..." line (uvicorn.error) should show. # WARNING "Reloading..." line (uvicorn.error) should show.
with suppress(Exception): with suppress(Exception):
config.setdefault("loggers", {}).setdefault("watchfiles.main", {}).setdefault("level", "WARNING") 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: if access_log:
+125 -17
View File
@@ -4,6 +4,7 @@ import asyncio
import importlib.metadata import importlib.metadata
import logging import logging
import os import os
import socket
from contextlib import suppress from contextlib import suppress
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -11,29 +12,49 @@ from typing import Any
import tracerite import tracerite
import uvicorn import uvicorn
from uvicorn import Config, Server from uvicorn import Config, Server
from uvicorn.main import STARTUP_FAILURE
from uvicorn.supervisors import ChangeReload, Multiprocess
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
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
# Install force color to aid tracerite and any external software to use full color when available
# Define NO_COLOR or FORCE_COLOR beforehand to avoid this
if "FORCE_COLOR" not in os.environ and use_color():
os.environ["FORCE_COLOR"] = "3"
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_WILDCARD_HOSTS = frozenset({"0.0.0.0", "::"}) # noqa: S104 _WILDCARD_HOSTS = frozenset({"0.0.0.0", "::"}) # noqa: S104
_LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1"})
def _bind_hosts(host: str) -> list[str]:
"""Addresses bound for a configured host; localhost binds both loopbacks."""
return sorted(_LOOPBACK_HOSTS) if host == "localhost" else [host]
def _connect_url(endpoints: list[dict]) -> str: def _connect_url(endpoints: list[dict]) -> str:
"""Return a URL the user can connect to for the first TCP endpoint. """Return a URL the user can connect to for the first TCP endpoint.
When running under the devserver (<PREFIX>_VITE_URL is set), the vite
devserver URL is shown instead, as that is where the page is served.
Wildcard binds (0.0.0.0, ::) are shown as localhost, as that is the 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. address a user can actually open. Unix-socket-only setups show plain
http://localhost (the typical reverse-proxy target).
""" """
if vite_url := env.vite_url:
return vite_url
for endpoint in endpoints: for endpoint in endpoints:
host = endpoint.get("host") host = endpoint.get("host")
if host is None: if host is None:
@@ -43,16 +64,34 @@ def _connect_url(endpoints: list[dict]) -> str:
elif ":" in host: # IPv6 literal elif ":" in host: # IPv6 literal
host = f"[{host}]" host = f"[{host}]"
return f"http://{host}:{endpoint['port']}" return f"http://{host}:{endpoint['port']}"
return "" return "http://localhost"
def _print_startup_box(template: str, app: str, endpoints: list[dict]) -> None: def _listen_addresses(endpoints: list[dict]) -> str:
"""Return space-separated listen addresses as bound (host:port or uds path).
localhost is expanded to both loopbacks, matching the actual binds.
"""
parts = []
for ep in endpoints:
if "uds" in ep:
parts.append(ep["uds"])
continue
for addr in _bind_hosts(ep["host"]):
shown = f"[{addr}]" if ":" in addr else addr # bracket IPv6 literals
parts.append(f"{shown}:{ep['port']}")
return " ".join(dict.fromkeys(parts))
def print_startup_box(template: str, app: str, endpoints: list[dict]) -> None:
"""Format the startup box template and print it. """Format the startup box template and print it.
Available fields: ``{module}`` (top-level package of the app path), Available fields: ``{module}`` (top-level package of the app path),
``{name}`` (module with spaces instead of underscores), ``{Name}`` ``{name}`` (module with spaces instead of underscores), ``{Name}``
(also capitalized), ``{version}`` (from installed package metadata, (also capitalized), ``{version}`` (from installed package metadata,
"dev" when not installed) and ``{url}``. "dev" when not installed), ``{listen}`` (space-separated listen
addresses as bound, localhost expanded to both loopbacks) and ``{url}``
(vite devserver URL when set, else the first connectable backend URL).
""" """
module = app.split(":", 1)[0].split(".", 1)[0] module = app.split(":", 1)[0].split(".", 1)[0]
name = module.replace("_", " ") name = module.replace("_", " ")
@@ -65,6 +104,7 @@ def _print_startup_box(template: str, app: str, endpoints: list[dict]) -> None:
"name": name, "name": name,
"Name": name.title(), "Name": name.title(),
"version": version, "version": version,
"listen": _listen_addresses(endpoints),
"url": _connect_url(endpoints), "url": _connect_url(endpoints),
} }
print_box(template.format_map(values)) print_box(template.format_map(values))
@@ -78,7 +118,7 @@ def run( # noqa: PLR0913
reload: bool | Path = False, reload: bool | Path = False,
workers: int | None = None, workers: int | None = None,
access_log: bool = True, access_log: bool = True,
startup_box: str | None = "{Name} {version}\n{url}", startup_box: str | None = "{Name} {version} @ {listen}\n{url}",
log_config: Any = uvicorn.config.LOGGING_CONFIG, # noqa: ANN401 log_config: Any = uvicorn.config.LOGGING_CONFIG, # noqa: ANN401
**uvicorn_config: Any, # noqa: ANN401 **uvicorn_config: Any, # noqa: ANN401
) -> None: ) -> None:
@@ -95,7 +135,7 @@ def run( # noqa: PLR0913
access_log: Enable our colored HTTP/WebSocket access logging middleware access_log: Enable our colored HTTP/WebSocket access logging middleware
(uvicorn's own access logging is always bypassed). (uvicorn's own access logging is always bypassed).
startup_box: Template for the startup box printed to stderr before startup_box: Template for the startup box printed to stderr before
serving (see _print_startup_box for fields), None to not print it. serving (see print_startup_box for fields), None to not print it.
log_config: Logging config passed to uvicorn. Dict configs are patched log_config: Logging config passed to uvicorn. Dict configs are patched
best-effort (see fastapi_vue.logging.patch_log_config): tracerite best-effort (see fastapi_vue.logging.patch_log_config): tracerite
loading and WebSocket chatter filtering are always installed, and loading and WebSocket chatter filtering are always installed, and
@@ -108,8 +148,10 @@ 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)
if isinstance(reload, Path): if isinstance(reload, Path):
uvicorn_config["reload_dirs"] = [str(reload)] uvicorn_config["reload_dirs"] = [str(reload)]
@@ -137,6 +179,67 @@ def run( # noqa: PLR0913
asyncio.run(serve(endpoints, **conf)) asyncio.run(serve(endpoints, **conf))
def _bind_sockets(endpoints: list[dict]) -> list[socket.socket]:
"""Bind sockets for all endpoints, expanding localhost to both loopbacks.
localhost is bound as 127.0.0.1 and ::1 explicitly, so resolver quirks
(notably Windows resolving localhost to ::1 only) cannot make the server
unreachable. Addresses that cannot be bound (e.g. IPv6 unavailable) are
skipped with a warning; exits only if nothing could be bound.
"""
sockets: list[socket.socket] = []
seen: set = set()
for ep in endpoints:
if "uds" in ep:
uds = ep["uds"]
if uds in seen:
continue
seen.add(uds)
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
sock.bind(uds)
Path(uds).chmod(0o666)
except OSError as e:
logger.warning("Could not bind unix socket %s: %s", uds, e)
sock.close()
continue
sock.set_inheritable(True)
sockets.append(sock)
continue
host, port = ep["host"], ep["port"]
for addr in _bind_hosts(host):
if (addr, port) in seen:
continue
seen.add((addr, port))
family = socket.AF_INET6 if ":" in addr else socket.AF_INET
sock = socket.socket(family, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
if family == socket.AF_INET6:
with suppress(OSError):
sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
try:
sock.bind((addr, port))
except OSError as e:
logger.warning("Could not bind %s:%d: %s", addr, port, e)
sock.close()
continue
sock.set_inheritable(True)
sockets.append(sock)
if not sockets:
logger.error("Could not bind any endpoint")
raise SystemExit(STARTUP_FAILURE)
return sockets
def _remove_uds_files(endpoints: list[dict]) -> None:
"""Remove unix socket files we created (mirrors uvicorn.run cleanup)."""
for ep in endpoints:
if "uds" in ep:
Path(ep["uds"]).unlink(missing_ok=True)
async def serve(endpoints: list[dict], **kwargs: Any) -> None: # noqa: ANN401 async def serve(endpoints: list[dict], **kwargs: Any) -> None: # noqa: ANN401
"""Serve the given endpoints in current process/loop. Does not spawn extra processes.""" """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} forbidden = {"reload", "workers"} & {k for k, v in kwargs.items() if v}
@@ -145,16 +248,21 @@ async def serve(endpoints: list[dict], **kwargs: Any) -> None: # noqa: ANN401
"Options %s have no effect in simple mode (multiple endpoints)", "Options %s have no effect in simple mode (multiple endpoints)",
", ".join(sorted(forbidden)), ", ".join(sorted(forbidden)),
) )
await asyncio.gather(*(Server(Config(**kwargs, **ep)).serve() for ep in endpoints)) try:
await Server(Config(**kwargs)).serve(sockets=_bind_sockets(endpoints))
finally:
_remove_uds_files(endpoints)
def serve_multiprocess(endpoints: list[dict], **kwargs: Any) -> None: # noqa: ANN401 def serve_multiprocess(endpoints: list[dict], **kwargs: Any) -> None: # noqa: ANN401
"""Serve using uvicorn.run() for reload/workers support. Only first endpoint is used.""" """Serve using uvicorn supervisors for reload/workers support."""
if len(endpoints) > 1: config = Config(**kwargs)
eps = [ep["uds"] if "uds" in ep else f"{ep['host']}:{ep['port']}" for ep in endpoints] server = Server(config)
logger.warning( sockets = _bind_sockets(endpoints)
"Current mode supports only one endpoint. Listening: %s, skipped: %s", try:
eps[0], if config.should_reload:
" ".join(eps[1:]), ChangeReload(config, target=server.run, sockets=sockets).run()
) else:
uvicorn.run(**kwargs, **endpoints[0]) Multiprocess(config, sockets=sockets).run()
finally:
_remove_uds_files(endpoints)
+5 -2
View File
@@ -19,7 +19,9 @@ from starlette.exceptions import HTTPException
from starlette.routing import Route from starlette.routing import Route
from zstandard import ZstdCompressor from zstandard import ZstdCompressor
logger = logging.getLogger("uvicorn.error") # Use FastAPI logging style from .environ import env
logger = logging.getLogger(__name__)
__all__ = ["Frontend"] __all__ = ["Frontend"]
@@ -287,7 +289,8 @@ class Frontend:
def _devmode_respond(_request: Request, _name: str = "") -> JSONResponse: def _devmode_respond(_request: Request, _name: str = "") -> JSONResponse:
"""Return error response directing to Vite server.""" """Return error response directing to Vite server."""
at = f" at {env.vite_url}" if env.vite_url else ""
return JSONResponse( return JSONResponse(
status_code=409, status_code=409,
content={"detail": "[devmode] Use Vite devserver instead."}, content={"detail": f"[devmode] Use Vite devserver{at} instead."},
) )
+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",
]
+1
View File
@@ -0,0 +1 @@
"""Tests for the fastapi_vue package."""
+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()
+60
View File
@@ -0,0 +1,60 @@
"""Tests for fastapi_vue.logging.patch_log_config overlay behavior."""
import logging.config
from fastapi_vue.logging import patch_log_config
def test_empty_dict_overlays_uvicorn_defaults() -> None:
"""An empty dict is a partial config: merged over uvicorn's defaults."""
config = patch_log_config({})
assert config["version"] == 1
assert config["disable_existing_loggers"] is False
assert config["root"]["handlers"] == ["default"]
assert "uvicorn" in config["loggers"]
logging.config.dictConfig(config) # must be a valid, complete config
def test_partial_logger_customization() -> None:
"""The documented use case: only the customization, no boilerplate."""
config = patch_log_config({"loggers": {"kanta": {"level": "DEBUG"}}})
assert config["loggers"]["kanta"] == {"level": "DEBUG"}
assert config["loggers"]["uvicorn"]["handlers"] == ["default"]
assert config["loggers"]["watchfiles.main"]["level"] == "WARNING"
logging.config.dictConfig(config)
assert logging.getLogger("kanta").level == logging.DEBUG
def test_overlay_root_level_wins() -> None:
"""User-supplied root level is kept; our handler wiring still applies."""
config = patch_log_config({"root": {"level": "ERROR"}})
assert config["root"]["level"] == "ERROR"
assert config["root"]["handlers"] == ["default"]
logging.config.dictConfig(config)
assert logging.getLogger().level == logging.ERROR
def test_overlay_formatter_customization_keeps_stock_siblings() -> None:
"""A user formatter replaces ours; the access formatter still works."""
config = patch_log_config({"formatters": {"default": {"fmt": "%(name)s %(message)s"}}})
assert config["formatters"]["default"] == {
"()": "uvicorn.logging.DefaultFormatter", # stock class, user's fmt
"fmt": "%(name)s %(message)s",
"use_colors": None,
}
assert "access" in config["formatters"]
logging.config.dictConfig(config)
def test_full_config_used_as_is() -> None:
"""A dict with version is complete: no uvicorn loggers appear."""
config = patch_log_config({"version": 1})
assert "uvicorn" not in config.get("loggers", {})
# No "default" handler exists, so root must not reference one.
assert "handlers" not in config["root"]
logging.config.dictConfig(config)
def test_non_dict_passes_through() -> None:
"""Non-dict configs (e.g. an ini file path) are returned untouched."""
assert patch_log_config("logging.ini") == "logging.ini"
+186 -31
View File
@@ -9,6 +9,7 @@ Options:
--module-name NAME Python module name (auto-detected from pyproject.toml) --module-name NAME Python module name (auto-detected from pyproject.toml)
--ports DEFAULT,VITE,DEV Port configuration (default: 3100,3100,3200) --ports DEFAULT,VITE,DEV Port configuration (default: 3100,3100,3200)
--dry Show what would be done without making changes --dry Show what would be done without making changes
-- ARGS Extra arguments forwarded to create-vue (e.g. -- --default)
""" """
import argparse import argparse
@@ -110,12 +111,14 @@ def ruff_format_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.""" """Add packages using uv.
cmd = ["uv", "add", "-q", "-U"]
Uses --frozen so only pyproject.toml is edited, without locking or
syncing - those happen in a single uv sync step after all changes.
"""
cmd = ["uv", "add", "-q", "--frozen"]
if group: if group:
cmd.extend(["--group", group]) cmd.extend(["--group", group])
else:
cmd.append("--no-sync")
cmd.extend(packages) cmd.extend(packages)
result = subprocess.run(cmd, cwd=cwd, check=False) # noqa: S603 result = subprocess.run(cmd, cwd=cwd, check=False) # noqa: S603
if result.returncode != 0: if result.returncode != 0:
@@ -472,8 +475,76 @@ def _find_app_in_subpackage(subpkg_dir: Path) -> tuple[Path, str] | None:
return None return None
def _add_devmode_to_main(content: str) -> str: def _migrate_devmode_in_main(content: str) -> str | None:
"""Add DEVMODE variable to an existing main module.""" """Spot-patch the pre-1.6 DEVMODE mechanism to the FASTAPI_VUE env prefix.
Replaces `DEVMODE = os.getenv("PREFIX_DEV") == "1"` with
`os.environ["FASTAPI_VUE"] = "PREFIX"` and remaining DEVMODE references
with env.dev, ensuring env is imported from fastapi_vue.
Returns the patched content, or None if there was nothing to patch.
"""
match = re.search(
r"^DEVMODE\s*=\s*os\.getenv\(\s*[\"']([A-Za-z0-9_]+)_DEV[\"']\s*\)\s*==\s*[\"']1[\"']",
content,
re.MULTILINE,
)
if not match:
return None
content = (
content[: match.start()]
+ f'os.environ["FASTAPI_VUE"] = "{match.group(1)}"'
+ content[match.end() :]
)
# Replace every remaining standalone DEVMODE reference (as in app.py
# migration, string literals are an accepted risk)
content = re.sub(r"(?<![\w.])DEVMODE\b", "env.dev", content)
if not re.search(r"^from fastapi_vue import\b.*\benv\b", content, re.MULTILINE):
if re.search(r"^from fastapi_vue import ", content, re.MULTILINE):
content = re.sub(
r"^from fastapi_vue import ",
"from fastapi_vue import env, ",
content,
count=1,
flags=re.MULTILINE,
)
else:
insert_line = find_import_insertion_line(content)
lines = content.splitlines(keepends=True)
insert_idx = insert_line - 1
import_text = "from fastapi_vue import env\n"
if insert_idx >= len(lines):
content = content.rstrip("\n") + "\n" + import_text
else:
content = "".join(lines[:insert_idx]) + import_text + "".join(lines[insert_idx:])
return content
def _patch_main_devmode(path: Path, *, dry: bool) -> str | None:
"""Apply _migrate_devmode_in_main to a main module in place, if needed.
Done in place even without the auto-upgrade marker: the marker guards
full-file overwrites, while leaving this change to a .new.py merge would
silently break dev mode for every customized pre-1.6 main.
Returns the migrated content if the module was (or would be) patched,
None if there was nothing to patch.
"""
content = path.read_text("UTF-8")
migrated = _migrate_devmode_in_main(content)
if migrated is None:
return None
migrated = ruff_format_content(migrated, path, mode="isort")
if dry:
print(f"✅ Would patch {path} (DEVMODE → FASTAPI_VUE)")
return migrated
path.write_text(migrated, "UTF-8", newline="\n")
print(f"✅ Patched {path} (DEVMODE → FASTAPI_VUE)")
return migrated
def _add_env_prefix_to_main(content: str) -> str:
"""Add FASTAPI_VUE environment prefix setup to an existing main module."""
lines = content.splitlines() lines = content.splitlines()
# Check if os is imported # Check if os is imported
@@ -488,7 +559,7 @@ def _add_devmode_to_main(content: str) -> str:
elif stripped and not stripped.startswith("#"): elif stripped and not stripped.startswith("#"):
break break
# Insert imports and DEVMODE after existing imports # Insert imports and env setup after existing imports
new_lines = [] new_lines = []
if not has_os_import: if not has_os_import:
new_lines.append("import os") new_lines.append("import os")
@@ -496,7 +567,7 @@ def _add_devmode_to_main(content: str) -> str:
[ [
"", "",
"# Added by fastapi-vue-setup", "# Added by fastapi-vue-setup",
'DEVMODE = os.getenv("ENVPREFIX_DEV") == "1"', 'os.environ["FASTAPI_VUE"] = "ENVPREFIX"',
"", "",
] ]
) )
@@ -596,11 +667,34 @@ def render_template(template: str, **kwargs: str) -> str:
return result return result
def patch_app_file(path: Path, main_module_path: str, app_var: str, *, dry: bool = False) -> bool: def needs_app_migration(project_dir: Path) -> bool:
"""Check if the project was set up with fastapi-vue older than 1.6.
Those versions patched app.py with a DEVMODE import from the main module;
1.6+ uses env.dev from fastapi_vue instead. Must be called before the
dependency step rewrites the fastapi-vue requirement in pyproject.toml.
"""
pyproject = project_dir / "pyproject.toml"
if not pyproject.exists():
return False
data = tomlkit.parse(pyproject.read_text("UTF-8"))
for dep in data.get("project", {}).get("dependencies", []):
match = re.match(r"\s*fastapi-vue(?:\[[^\]]*\])?\s*(.*)", str(dep))
if match:
version = re.search(r"(\d+)\.(\d+)", match.group(1))
return version is not None and (int(version[1]), int(version[2])) < (1, 6)
return False
def patch_app_file(
path: Path, main_module_path: str, app_var: str, *, migrate: bool = False, dry: bool = False
) -> bool:
"""Patch an existing app.py with frontend integration. """Patch an existing app.py with frontend integration.
Inserts imports at top (ruff will sort them), route at bottom, Inserts imports at top (ruff will sort them), route at bottom,
and tries to patch lifespan with frontend.load(). and tries to patch lifespan with frontend.load(). With migrate=True,
pre-1.6 patching (DEVMODE import from the main module) is first
rewritten to the current format (env.dev).
Returns True if patched, False if already patched or failed. Returns True if patched, False if already patched or failed.
""" """
@@ -611,24 +705,41 @@ def patch_app_file(path: Path, main_module_path: str, app_var: str, *, dry: bool
original_content = path.read_text("UTF-8") original_content = path.read_text("UTF-8")
content = original_content content = original_content
# Check what's already patched # Migrate pre-1.6 patching to the current format: DEVMODE via
has_frontend = "from fastapi_vue import Frontend" in content # fastapi_vue.env (the Frontend import stays as-is)
has_devmode = f"from {main_module_path} import DEVMODE" in content if migrate:
old_import = f"from {main_module_path} import DEVMODE"
if old_import in content:
# The import sort at the end merges this with any existing
# `from fastapi_vue import Frontend` line
content = content.replace(old_import, "from fastapi_vue import env")
# Replace every remaining standalone DEVMODE reference (not just
# the debug= parameter); it may also appear inside string literals,
# but that's an accepted risk over AST rewriting
content = re.sub(r"(?<![\w.])DEVMODE\b", "env.dev", content)
# Check what's already patched; plain "Frontend(" so user modifications
# of the integration (renames, different call shape) still count
has_frontend = "Frontend(" in content
has_debug_arg = re.search(r"FastAPI\s*\([^)]*debug\s*=", content) is not None has_debug_arg = re.search(r"FastAPI\s*\([^)]*debug\s*=", content) is not None
has_lifespan = "await frontend.load()" in content has_lifespan = "await frontend.load()" in content
if has_frontend and has_devmode and has_debug_arg and has_lifespan: already_patched = has_frontend and has_debug_arg and has_lifespan
if content == original_content and already_patched:
print(f"✔️ {path} (already patched)") print(f"✔️ {path} (already patched)")
return False return False
route_line = f'frontend.route({app_var}, "/")' route_line = f'frontend.route({app_var}, "/")'
# Add missing imports (using AST to find correct insertion point) # Add missing imports (using AST to find correct insertion point);
# the import sort at the end merges duplicate from-imports
imports = [] imports = []
if not has_frontend: if not has_frontend:
imports.extend(["from pathlib import Path", "from fastapi_vue import Frontend"]) imports.append("from pathlib import Path")
if not has_devmode: if not has_frontend or not re.search(
imports.append(f"from {main_module_path} import DEVMODE") r"^from fastapi_vue import\b.*\benv\b", content, re.MULTILINE
):
imports.append("from fastapi_vue import Frontend, env")
if imports: if imports:
insert_line = find_import_insertion_line(content) insert_line = find_import_insertion_line(content)
lines = content.splitlines(keepends=True) lines = content.splitlines(keepends=True)
@@ -663,14 +774,14 @@ def patch_app_file(path: Path, main_module_path: str, app_var: str, *, dry: bool
lines.append(route_line) lines.append(route_line)
content = "\n".join(lines) content = "\n".join(lines)
# Try to patch FastAPI() call with debug=DEVMODE if no debug arg exists # Try to patch FastAPI() call with debug=env.dev if no debug arg exists
if not has_debug_arg: if not has_debug_arg:
fastapi_pattern = r"(\w+\s*=\s*FastAPI\s*\()([^)]*)\)" fastapi_pattern = r"(\w+\s*=\s*FastAPI\s*\()([^)]*)\)"
for match in re.finditer(fastapi_pattern, content, re.DOTALL): for match in re.finditer(fastapi_pattern, content, re.DOTALL):
args = match.group(2) args = match.group(2)
if "debug" not in args: if "debug" not in args:
# Add debug=DEVMODE as last argument # Add debug=env.dev as last argument
new_args = f"{args}, debug=DEVMODE" if args.strip() else "debug=DEVMODE" new_args = (f"{args}, " if args.strip() else "") + "debug=env.dev"
content = ( content = (
content[: match.start()] content[: match.start()]
+ match.group(1) + match.group(1)
@@ -969,6 +1080,11 @@ def _upgrade_old_vite_plugin(path: Path, module_name: str, *, dry: bool = False)
_new_files_written: list[tuple[Path, Path]] = [] _new_files_written: list[tuple[Path, Path]] = []
def _strip_upgrade_marker(content: str) -> str:
"""Remove the auto-upgrade marker line, for content comparison."""
return "\n".join(line for line in content.splitlines() if UPGRADE_MARKER not in line)
def write_file( def write_file(
path: Path, path: Path,
content: str, content: str,
@@ -1211,7 +1327,9 @@ def ensure_python_project(project_dir: Path, *, dry: bool = False) -> bool:
return True return True
def ensure_frontend(project_dir: Path, *, dry: bool = False) -> bool: def ensure_frontend(
project_dir: Path, *, vue_args: list[str] | None = None, dry: bool = False
) -> bool:
"""Ensure frontend directory exists with a Vue project, run create-vue if needed.""" """Ensure frontend directory exists with a Vue project, run create-vue if needed."""
frontend_dir = project_dir / "frontend" frontend_dir = project_dir / "frontend"
package_json = frontend_dir / "package.json" package_json = frontend_dir / "package.json"
@@ -1234,6 +1352,10 @@ def ensure_frontend(project_dir: Path, *, dry: bool = False) -> bool:
"bun": [js_tool, "create", "vue@latest", "frontend"], "bun": [js_tool, "create", "vue@latest", "frontend"],
} }
create_cmd = create_vue_commands[js_name] create_cmd = create_vue_commands[js_name]
if vue_args:
# npm needs a `--` separator so it doesn't eat the arguments;
# create-vue runs non-interactively when given feature flags (e.g. --default)
create_cmd = [*create_cmd, *(["--"] if js_name == "npm" else []), *vue_args]
if dry: if dry:
print(f"🎨 Would run: {' '.join(create_cmd)}") print(f"🎨 Would run: {' '.join(create_cmd)}")
@@ -1241,6 +1363,7 @@ def ensure_frontend(project_dir: Path, *, dry: bool = False) -> bool:
print("🎨 No frontend/ found, creating Vue project...") print("🎨 No frontend/ found, creating Vue project...")
print(f">>> {' '.join(create_cmd)}") print(f">>> {' '.join(create_cmd)}")
if not vue_args:
print("(Follow the prompts to configure your Vue app)") print("(Follow the prompts to configure your Vue app)")
print() print()
result = subprocess.run(create_cmd, cwd=project_dir, check=False) # noqa: S603 result = subprocess.run(create_cmd, cwd=project_dir, check=False) # noqa: S603
@@ -1285,7 +1408,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
print(f"🔧 Setting up project: {project_dir}") print(f"🔧 Setting up project: {project_dir}")
# Step 1: Ensure frontend exists (do this first so cancellation doesn't leave partial setup) # Step 1: Ensure frontend exists (do this first so cancellation doesn't leave partial setup)
if not ensure_frontend(project_dir, dry=dry): if not ensure_frontend(project_dir, vue_args=args.vue_args, dry=dry):
return 1 return 1
# Step 2: Ensure Python project exists # Step 2: Ensure Python project exists
@@ -1308,7 +1431,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
# Check if project already has a CLI entrypoint in pyproject.toml # Check if project already has a CLI entrypoint in pyproject.toml
existing_cli_module = _find_existing_cli_module_path(project_dir, module_name) existing_cli_module = _find_existing_cli_module_path(project_dir, module_name)
# Determine main module path for DEVMODE import # Determine main module path (for migrating old DEVMODE imports)
main_module_path = existing_cli_module or f"{module_name}.__main__" main_module_path = existing_cli_module or f"{module_name}.__main__"
if existing_cli_module: if existing_cli_module:
print(f"️ Using existing CLI: {existing_cli_module}") print(f"️ Using existing CLI: {existing_cli_module}")
@@ -1455,7 +1578,9 @@ def cmd_setup(args: argparse.Namespace) -> int:
# === Handle app module === # === Handle app module ===
if app_file: if app_file:
# Existing app: patch with import, route, and try to patch lifespan # Existing app: patch with import, route, and try to patch lifespan
patch_app_file(app_file, main_module_path, app_var, dry=dry) patch_app_file(
app_file, main_module_path, app_var, migrate=needs_app_migration(project_dir), dry=dry
)
else: else:
# No app: create full app.py # No app: create full app.py
# Create __init__.py if missing # Create __init__.py if missing
@@ -1483,7 +1608,19 @@ def cmd_setup(args: argparse.Namespace) -> int:
main_content = render_template(template, **tpl_vars) main_content = render_template(template, **tpl_vars)
if main_file.exists(): if main_file.exists():
# File exists: update if it has the auto-upgrade marker, otherwise use fallback # Spot-patch the pre-1.6 DEVMODE mechanism in place first - the
# auto-upgrade marker guards full-file overwrites, but leaving this
# change to a .new.py merge would silently break dev mode
migrated = _patch_main_devmode(main_file, dry=dry)
existing = migrated if migrated is not None else main_file.read_text("UTF-8")
# Update if it has the auto-upgrade marker, otherwise use fallback -
# unless the markerless file is otherwise up to date (e.g. the
# DEVMODE spot-patch was the only change), then no fallback is needed
if UPGRADE_MARKER not in existing and _strip_upgrade_marker(
existing
) == _strip_upgrade_marker(ruff_format_content(main_content, main_file)):
print(f"✔️ {main_file} (already up to date)")
else:
write_file( write_file(
main_file, main_file,
main_content, main_content,
@@ -1501,7 +1638,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
) )
else: else:
# Existing CLI entrypoint: write our template as .new.py beside the existing module # Existing CLI entrypoint: write our template as .new.py beside the existing module
# and also patch the existing module with DEVMODE if needed # and also patch the existing module with FASTAPI_VUE setup if needed
_write_fallback_file( _write_fallback_file(
main, main,
main_fallback, main_fallback,
@@ -1510,9 +1647,10 @@ def cmd_setup(args: argparse.Namespace) -> int:
executable=False, executable=False,
) )
if main.exists(): if main.exists():
content = main.read_text("UTF-8") migrated = _patch_main_devmode(main, dry=dry)
if "DEVMODE" not in content: content = migrated if migrated is not None else main.read_text("UTF-8")
new_content = _add_devmode_to_main(content) if "FASTAPI_VUE" not in content:
new_content = _add_env_prefix_to_main(content)
new_file = main.with_suffix(".new.py") new_file = main.with_suffix(".new.py")
_write_fallback_file( _write_fallback_file(
main, main,
@@ -1603,9 +1741,16 @@ def cmd_setup(args: argparse.Namespace) -> int:
fastapi_vue_req = f"fastapi-vue~={mmp[1]}.{mmp[2]}.{mmp[3]}" if mmp else "fastapi-vue" fastapi_vue_req = f"fastapi-vue~={mmp[1]}.{mmp[2]}.{mmp[3]}" if mmp else "fastapi-vue"
if dry: if dry:
print(f"📦 Would add: fastapi[standard], {fastapi_vue_req}") print(f"📦 Would add: fastapi[standard], {fastapi_vue_req}")
print("📦 Would run: uv sync")
else: else:
print("📦 Dependencies") print("📦 Dependencies")
uv_add_packages(["fastapi[standard]", fastapi_vue_req], cwd=project_dir) uv_add_packages(["fastapi[standard]", fastapi_vue_req], cwd=project_dir)
# uv add runs with --frozen, so lock and sync the environment once
# everything is in place; attached to the terminal so the user sees
# the updates, and non-fatal - setup is complete either way
result = subprocess.run(["uv", "sync"], cwd=project_dir, check=False) # noqa: S607
if result.returncode != 0:
print("⚠️ uv sync failed - run it manually to update the environment")
print() print()
print_boxed("Setup complete!") print_boxed("Setup complete!")
@@ -1663,6 +1808,8 @@ Examples:
fastapi-vue-setup . Set up integration in current directory fastapi-vue-setup . Set up integration in current directory
fastapi-vue-setup . --dry Preview what would be done fastapi-vue-setup . --dry Preview what would be done
fastapi-vue-setup . --ports=8000,5173,8080 Change default ports (backend, vite dev, backend dev) fastapi-vue-setup . --ports=8000,5173,8080 Change default ports (backend, vite dev, backend dev)
fastapi-vue-setup my-app -- --default Non-interactive create-vue (extra args after --
are forwarded to create-vue, e.g. --default, --ts)
""", """,
) )
parser.add_argument( parser.add_argument(
@@ -1685,7 +1832,15 @@ Examples:
) )
parser.add_argument("--dry", "--dry-run", action="store_true", help="Show what would be done") parser.add_argument("--dry", "--dry-run", action="store_true", help="Show what would be done")
args = parser.parse_args() # Everything after a standalone `--` is forwarded verbatim to create-vue
argv = sys.argv[1:]
if "--" in argv:
split = argv.index("--")
ours, vue_args = argv[:split], argv[split + 1 :]
else:
ours, vue_args = argv, []
args = parser.parse_args(ours)
args.vue_args = vue_args
if args.project_dir is None: if args.project_dir is None:
parser.print_help() parser.print_help()
+1
View File
@@ -49,3 +49,4 @@ ignore = ["CPY", "D203", "D213", "COM812", "PLR2004"]
"template/**" = ["F821"] # Undefined names are template placeholders "template/**" = ["F821"] # Undefined names are template placeholders
"template/scripts/devserver.py" = ["N806"] # MODULE_NAME is a template variable "template/scripts/devserver.py" = ["N806"] # MODULE_NAME is a template variable
"fastapi_vue_setup.py" = ["PLR", "C901", "T201", "RUF001"] "fastapi_vue_setup.py" = ["PLR", "C901", "T201", "RUF001"]
"**/tests/**" = ["S101"] # Asserts are the point of tests
+3 -3
View File
@@ -5,10 +5,10 @@ import argparse
import os import os
from pathlib import Path from pathlib import Path
from fastapi_vue import server from fastapi_vue import env, server
DEFAULT_PORT = TEMPLATE_DEFAULT_PORT DEFAULT_PORT = TEMPLATE_DEFAULT_PORT
DEVMODE = os.getenv("ENVPREFIX_DEV") == "1" os.environ["FASTAPI_VUE"] = "ENVPREFIX"
def main() -> None: def main() -> None:
@@ -26,7 +26,7 @@ def main() -> None:
listen=args.listen, listen=args.listen,
default_port=DEFAULT_PORT, default_port=DEFAULT_PORT,
server_header=False, server_header=False,
reload=Path(__file__).parent if DEVMODE else False, reload=Path(__file__).parent if env.dev else False,
) )
+2 -3
View File
@@ -5,8 +5,7 @@ from contextlib import asynccontextmanager
from pathlib import Path from pathlib import Path
from fastapi import FastAPI from fastapi import FastAPI
from fastapi_vue import Frontend from fastapi_vue import Frontend, env
from MAIN_MODULE import DEVMODE
# Vue Frontend static files # Vue Frontend static files
frontend = Frontend(Path(__file__).with_name("frontend-build")) frontend = Frontend(Path(__file__).with_name("frontend-build"))
@@ -19,7 +18,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator:
yield yield
app = FastAPI(title="PROJECT_TITLE", debug=DEVMODE, lifespan=lifespan) app = FastAPI(title="PROJECT_TITLE", debug=env.dev, lifespan=lifespan)
# Add API routes here... # Add API routes here...
+5 -5
View File
@@ -8,11 +8,11 @@
* - Disables Vite's screen clearing on startup * - Disables Vite's screen clearing on startup
* *
* Options: * Options:
* paths - Array of paths to proxy (default: ["/api"]) * paths - Array of paths to proxy (default: ['/api'])
*/ */
export default function fastapiVue({ paths = ["/api"] } = {}) { export default function fastapiVue({ paths = ['/api'] } = {}) {
const backendUrl = process.env.ENVPREFIX_BACKEND_URL || "http://localhost:TEMPLATE_DEV_PORT" const backendUrl = process.env.ENVPREFIX_BACKEND_URL || 'http://localhost:TEMPLATE_DEV_PORT'
// Build proxy configuration for each path // Build proxy configuration for each path
const proxy = {} const proxy = {}
@@ -25,12 +25,12 @@ export default function fastapiVue({ paths = ["/api"] } = {}) {
} }
return { return {
name: "vite-plugin-fastapi-MODULE_NAME", name: 'vite-plugin-fastapi-MODULE_NAME',
config: () => ({ config: () => ({
clearScreen: false, clearScreen: false,
server: { proxy }, server: { proxy },
build: { build: {
outDir: "../MODULE_NAME/frontend-build", outDir: '../MODULE_NAME/frontend-build',
emptyOutDir: true, emptyOutDir: true,
}, },
}), }),
+9 -5
View File
@@ -5,8 +5,8 @@
import argparse import argparse
import asyncio import asyncio
import os import os
import subprocess
import sys import sys
from contextlib import suppress
from pathlib import Path from pathlib import Path
import tracerite import tracerite
@@ -48,11 +48,11 @@ async def run_devserver(
os.environ["ENVPREFIX_DEV"] = "1" os.environ["ENVPREFIX_DEV"] = "1"
async with ProcessGroup() as pg: async with ProcessGroup() as pg:
pg.create_task(check_ports_free(viteurl, backurl))
npm_i = await pg.spawn(*npm_install, cwd=front) npm_i = await pg.spawn(*npm_install, cwd=front)
await check_ports_free(viteurl, backurl) await pg.spawn(*MODULE_NAME, *(extra_args or []), vital=True)
await pg.spawn(*MODULE_NAME, *(extra_args or []))
await pg.wait(npm_i, ready(backurl, path=HEALTH)) await pg.wait(npm_i, ready(backurl, path=HEALTH))
await pg.spawn(*vite, cwd=front) await pg.spawn(*vite, cwd=front, vital=True)
def main() -> None: def main() -> None:
@@ -75,8 +75,12 @@ def main() -> None:
help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})", help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})",
) )
args, extra_args = parser.parse_known_args() args, extra_args = parser.parse_known_args()
with suppress(KeyboardInterrupt): try:
asyncio.run(run_devserver(args.listen, args.backend, extra_args)) asyncio.run(run_devserver(args.listen, args.backend, extra_args))
except* KeyboardInterrupt:
pass # user stopped the devserver: normal exit
except* (subprocess.SubprocessError, RuntimeError):
raise SystemExit(1) from None # logged in devutil already; exit 1
HELP_EPILOG = """ HELP_EPILOG = """
+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:
+71 -94
View File
@@ -1,107 +1,83 @@
# ruff: noqa: INP001 # ruff: noqa: INP001
"""Utilities meant for devserver script, used only in source repository with dev deps.""" """Utilities meant for devserver script, used only in source repository with dev deps."""
from __future__ import annotations
import asyncio import asyncio
import subprocess
import sys import sys
from asyncio.subprocess import Process
from contextlib import suppress from contextlib import suppress
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING, Any, Self from subprocess import CalledProcessError
from typing import TYPE_CHECKING, Any
from urllib.parse import urlsplit from urllib.parse import urlsplit
from buildutil import find_dev_tool, find_install_tool, logger from buildutil import find_dev_tool, find_install_tool, logger
from fastapi_vue.hostutil import parse_endpoint from fastapi_vue.hostutil import parse_endpoint
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Coroutine from collections.abc import Awaitable
class ProcessGroup: class ProcessGroup(asyncio.TaskGroup):
"""Manage async subprocesses with automatic cleanup, like TaskGroup for processes.""" """TaskGroup with structured ownership of async subprocesses."""
def __init__(self) -> None: def __init__(self, *, terminate_timeout: float = 10) -> None:
"""Initialize empty process tracking.""" """Set the grace period before terminate() escalates to kill()."""
self._procs: list[asyncio.subprocess.Process] = [] super().__init__()
self._cmds: dict[int, str] = {} # pid -> command name self._terminate_timeout = terminate_timeout
self._cmds: dict[Process, tuple[str, ...]] = {}
async def spawn( async def spawn(self, *cmd: str, cwd: str | None = None, vital: bool = False) -> Process:
self, """Spawn and own a subprocess. If a vital process exits, the group cancels."""
*cmd: str,
cwd: str | None = None,
) -> asyncio.subprocess.Process:
"""Spawn a subprocess and track it."""
cmd_name = Path(cmd[0]).stem
logger.info(">>> %s", " ".join([cmd_name, *cmd[1:]]))
proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd)
self._procs.append(proc)
self._cmds[proc.pid] = cmd_name
return proc
async def wait( async def run() -> None:
self, name = Path(cmd[0]).stem
*waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]", logger.info(">>> %s", " ".join([name, *cmd[1:]]))
) -> None:
"""Wait for processes/coroutines to complete, raise SystemExit on failure."""
async def wait_proc(proc: asyncio.subprocess.Process) -> None:
returncode = await proc.wait()
if returncode != 0:
cmd_name = self._cmds.get(proc.pid, "unknown")
raise subprocess.CalledProcessError(returncode, cmd_name)
tasks = [
wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w for w in waitables
]
try: try:
await asyncio.gather(*tasks) proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd)
except subprocess.CalledProcessError as e: self._cmds[proc] = cmd
logger.warning("%s failed with exit status %d", e.cmd, e.returncode) started.set_result(proc)
raise SystemExit(1) from None except Exception as e: # noqa: BLE001
started.set_exception(e)
async def __aenter__(self) -> Self:
"""Enter the async context manager."""
return self
async def __aexit__(self, exc_type: type[BaseException] | None, *_: object) -> None:
"""Wait for one process to exit, terminate others, then wait for all."""
await self._cleanup(immediate=exc_type is not None)
async def _cleanup(self, *, immediate: bool = False) -> None:
running = [p for p in self._procs if p.returncode is None]
if not running:
return return
if not immediate:
# Wait for any one process to exit
with suppress(asyncio.CancelledError):
await asyncio.wait(
[asyncio.create_task(p.wait()) for p in running],
return_when=asyncio.FIRST_COMPLETED,
)
# Terminate remaining processes
for p in self._procs:
if p.returncode is None:
with suppress(ProcessLookupError):
p.terminate()
# Wait for all to finish (with overall timeout), shielded from cancellation
still_running = [p for p in self._procs if p.returncode is None]
if still_running:
with suppress(asyncio.CancelledError):
try: try:
await asyncio.shield( returncode = await proc.wait()
asyncio.wait_for( finally:
asyncio.gather(*[p.wait() for p in still_running]),
timeout=10,
),
)
except TimeoutError:
for p in self._procs:
if p.returncode is None:
with suppress(ProcessLookupError): with suppress(ProcessLookupError):
p.kill() proc.terminate()
await p.wait() try:
await asyncio.wait_for(proc.wait(), self._terminate_timeout)
except TimeoutError:
with suppress(ProcessLookupError):
proc.kill()
await proc.wait()
if vital:
logger.warning("Vital process %s exited", name)
raise CalledProcessError(returncode, cmd)
started = asyncio.get_running_loop().create_future()
self.create_task(run())
return await asyncio.shield(started)
async def wait(self, *waitables: Process | Awaitable) -> tuple[Any, ...]:
"""Wait concurrently and return results in argument order."""
async def task(w: Process | Awaitable) -> Any: # noqa: ANN401
if not isinstance(w, Process):
return await w
if retcode := await w.wait():
cmd = self._cmds[w]
logger.warning("Process %s exited with status %d", Path(cmd[0]).stem, retcode)
raise CalledProcessError(retcode, cmd)
return retcode
async with asyncio.TaskGroup() as group:
tasks = [group.create_task(task(w)) for w in waitables]
return tuple(task.result() for task in tasks)
async def http_get_server(url: str, timeout: float) -> str | None: # noqa: ASYNC109 async def http_get_server(url: str, timeout: float) -> str | None: # noqa: ASYNC109
@@ -127,40 +103,41 @@ async def http_get_server(url: str, timeout: float) -> str | None: # noqa: ASYN
writer.close() writer.close()
except (OSError, EOFError, ValueError, TimeoutError): except (OSError, EOFError, ValueError, TimeoutError):
return None return None
for line in data.decode("latin-1").split("\r\n"): for line in data.decode(errors="replace").split("\r\n"):
if line.lower().startswith("server:"): if line.lower().startswith("server:"):
return line.split(":", 1)[1].strip() return line[7:].strip()
return "" return ""
async def check_ports_free(*urls: str) -> None: async def check_ports_free(*urls: str) -> None:
"""Verify URLs are not responding (ports are free). Raise SystemExit if any respond.""" """Verify URLs are not responding (ports are free).
async def check(url: str) -> None: Meant to run as a task inside a TaskGroup. Logs the conflict and raises
server = await http_get_server(url, timeout=0.1) RuntimeError (handled like a failed process) if any URL responds.
"""
servers = await asyncio.gather(*(http_get_server(url, timeout=0.1) for url in urls))
for url, server in zip(urls, servers, strict=True):
if server is not None: if server is not None:
logger.warning("Conflicting %s already running at %s", server or "server", url) logger.error("Conflicting %s already running at %s", server or "server", url)
raise SystemExit(1) raise RuntimeError(url)
await asyncio.gather(*[check(url) for url in urls])
async def ready(url: str, path: str = "", max_attempts: int = 50) -> None: async def ready(url: str, path: str = "", max_attempts: int = 50) -> None:
"""Wait for the server to be ready by polling an endpoint. """Wait for the server to be ready by polling an endpoint.
Use empty path to disable the check and make this return immediately. Use empty path to disable the check and make this return immediately.
Raises SystemExit(1) if server doesn't start in time. Logs, then raises RuntimeError if the server doesn't start in time.
""" """
if not path: if not path:
return return
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.warning("Backend didn't start in time") logger.error("Backend at %s didn't start in time", url)
raise SystemExit(1) raise RuntimeError(url)
await asyncio.sleep(0.1) await asyncio.sleep(0.1)