Compare commits

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

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

The middleware is installed by patching Config.load/load_app, hooked
into AccessFormatter instantiation so reload/worker subprocesses
(which re-apply log_config without re-calling run()) are covered.
2026-08-31 04:16:43 +00:00
LeoVasanko 52ca3f4f49 Update project URLs. 2026-08-31 02:18:08 +00:00
LeoVasanko ebc13c0ee4 Add ruff ignores for the check phase to avoid useless diagnostics. 2026-08-26 01:07:24 +00:00
LeoVasanko 7a64d8f73b Disable server: uvicorn header. 2026-08-25 23:42:21 +00:00
LeoVasanko a5b7d88a89 Fix typo in devserver.py comment
'Tell the everyone' -> 'Tell everyone via environment'.
2026-08-25 23:41:46 +00:00
LeoVasanko 03aa31cedb Implement smarter reload handing on fastapi_vue.server(), simplify __main__, reload the main folder regardless of CWD. 2026-06-29 18:49:18 +00:00
LeoVasanko ec6a084969 Fix hardcoded project dir in template __main__.py (regression in commit 1b24a7). 2026-06-29 18:18:04 +00:00
LeoVasanko 15a6710c17 [lint] Format template after change. 2026-03-08 17:35:02 +00:00
LeoVasanko dcaf415032 Fix a typing issue with the fastapi-vue staticfiles module causing problems with SPA mode handler. 2026-03-08 17:30:42 +00:00
LeoVasanko 79316eebd4 Rename build-frontend.py to buildhook.py. 2026-03-08 15:40:49 +00:00
LeoVasanko 682d70cb23 Cleanup for ALL ruff checks, and re-ruff to target project settings when installing templates, avoiding formatting errors after patching. 2026-03-08 15:25:52 +00:00
LeoVasanko 1aaf040db7 Make devserver backend health check endpoint configurable, preserved in upgrades and optional. 2026-03-07 23:55:10 +00:00
LeoVasanko aac27da67c Suppress CancelledError along with KeyboardInterrupt as a normal exit from server.run(). 2026-02-18 17:50:34 +00:00
LeoVasanko 79e645a263 Add parse_endpoints helper function that handles complicated listen structures alike server.run() already did, returning a simple list of every endpoint. 2026-02-18 17:05:10 +00:00
LeoVasanko 8b8a24a6b2 Determine app module path correctly even when in a submodule. A bit cleaner dev options. 2026-02-11 20:41:36 +00:00
LeoVasanko cb6017cedf Find custom CLI main before looking up existing DEFAULT_PORT. Cleaner messages when Vue doesn't need patching. 2026-02-11 20:27:35 +00:00
LeoVasanko fab4108e92 Cleaner devserver help message. 2026-02-11 20:06:36 +00:00
LeoVasanko 637f737d4c Restore app module isort by reusing the full format function. 2026-02-11 19:48:45 +00:00
LeoVasanko 0b986717eb Cleaner message 2026-02-11 18:40:58 +00:00
LeoVasanko 9469d57f90 Better patching with existing custom main modules. Print all .new.py needing migration at the end. 2026-02-11 18:35:28 +00:00
LeoVasanko 1ce0fffe3e Use dev port as default fallback in vite-plugin-fastapi, instead of default port. No effect on invocations via devserver script. 2026-02-11 18:15:57 +00:00
LeoVasanko f483cf978c Ruff format 2026-02-10 20:56:06 +00:00
LeoVasanko 0e12e3a531 Improved error message. 2026-02-10 19:51:05 +00:00
LeoVasanko 67bea9a2a7 Remove unnecessary bool() 2026-02-10 19:46:32 +00:00
LeoVasanko a05c8236f6 Format Python modules installed with ruff to target project style. 2026-02-10 19:34:35 +00:00
LeoVasanko fc3d6dc912 README 2026-02-10 19:00:41 +00:00
LeoVasanko c63a375ab2 Use ## to allow command be pasted as a whole. 2026-02-10 17:14:14 +00:00
LeoVasanko c98994fd1a Dry run cleanup, also allow --dry-run. 2026-02-10 17:06:42 +00:00
LeoVasanko 6beaa88418 Improved setup complete message. 2026-02-10 17:00:24 +00:00
LeoVasanko 26a77dcf84 Cleaner messages while running the setup. 2026-02-10 16:39:45 +00:00
LeoVasanko 38c9ffe00b Add --version, rename --dry-run to just --dry. 2026-02-10 16:29:30 +00:00
LeoVasanko b87984af9a Consistent form of CLI arguments. 2026-02-10 16:16:32 +00:00
LeoVasanko d3b6fccce0 The devserver now takes --listen for vite endpoint rather than a positional argument. Changed FRONTEND_URL to VITE_URL and streamlined terminology elsewhere. 2026-02-10 16:09:07 +00:00
LeoVasanko d2ee7395c5 health?from=frontend for logs 2026-02-10 16:01:06 +00:00
LeoVasanko b94608ffe3 Run ruff within local venv (otherwise required system path which did not include the venv with uv tool install). 2026-02-09 16:03:30 +00:00
LeoVasanko ef6bce8f3f Use the CLI script rather than python -m for running the main app from devserver. 2026-02-09 15:55:08 +00:00
LeoVasanko b8bf46132a Reload arguments passed differently to avoid uvicorn warning in prod mode. 2026-02-06 22:43:00 +00:00
LeoVasanko ca85c2fcfb Ruff 2026-02-06 22:37:03 +00:00
LeoVasanko 440808361a Patch existing projects better. 2026-02-06 20:09:17 +00:00
LeoVasanko 05fd5c562e Missing import 2026-02-06 19:35:04 +00:00
LeoVasanko 5f0e74713f Improved dry mode message. 2026-02-06 19:17:06 +00:00
LeoVasanko 2d117a9d5e Improved messages. 2026-02-06 19:07:50 +00:00
LeoVasanko 4bf102579a Added automatic handling of FastAPI debug mode. Use templating more completely to enable project name as env prefix etc. 2026-02-06 19:05:36 +00:00
LeoVasanko fbed7873e5 [fastapi_vue] Further cleanup, wildcards for favicon. 2026-02-06 18:09:22 +00:00
LeoVasanko 008b3593db [fastapi_vue] StaticFiles cleanup: use app.debug rather than env variables; default assets caching. 2026-02-06 17:21:40 +00:00
LeoVasanko 3820c6f3c1 Fix argument passing to vite under deno. 2026-02-05 20:11:22 +00:00
LeoVasanko c300a1a291 Simplified uvicorn running in multiprocess mode, avoiding a crash on Windows when it couldn't setup threads (running from a worker thread). 2026-02-02 21:02:34 +00:00
LeoVasanko bdfaa22b0c Another file extension showing, swapping to stem for logging. 2026-02-02 20:47:32 +00:00
LeoVasanko 47e5ec279c Hide tool.CMD .EXE etc. from logs, makes Windows cleaner. 2026-02-02 20:42:18 +00:00
LeoVasanko 093b83737e Improved status messages, added a note of not installing main CLI entry point. 2026-02-02 20:35:30 +00:00
LeoVasanko a84041a923 Refactor with new fastapi-vue utility module to run the server, used in CLI main and devserver to run the main app. Default ports made part of templating, so that --ports can be easily specified to fastapi-vue-setup to customize (respects earlier choices when no --ports is specified). New overall default ports 3100, 3200. 2026-02-02 19:56:31 +00:00
LeoVasanko 9a9a00380f Add a release script to simplify building both. 2026-02-02 19:53:06 +00:00
LeoVasanko 0e4f8f65f2 Moved fastapi-vue project to fastapi-vue-setup repository, using common version numbering (starting with v0.7.0). 2026-02-02 19:51:45 +00:00
LeoVasanko 5391c81e8b Cleanup fixes, and-- cleanup. 2026-02-02 16:43:39 +00:00
LeoVasanko c106ce1a44 Add missing files. 2026-02-02 16:27:34 +00:00
LeoVasanko b7b1920846 Make devserver verify that no servers are already running before starting them. 2026-02-02 16:26:19 +00:00
LeoVasanko 8ccfa1f658 Simplified devserver script, fixed termination on failed npm install. 2026-02-02 15:56:19 +00:00
LeoVasanko c23401de88 Avoid external venv or of fastapi-vue-setup itself leaking into target project. Cleaner and faster uv processing. 2026-02-02 13:46:34 +00:00
LeoVasanko c271f66615 Work around some Microsoft idiocy to attain Windows compatibility. 2026-02-02 13:28:36 +00:00
LeoVasanko 458d798022 Refactored scripts to remove any extra code from build-frontend.py and devserver.py (for better user editability). Individual helper functions can still be imported from scripts/fastapi-vue/ for any customization needs. Removed devmode support from CLI entry point simplifying it. Instead running via FastAPI CLI in dev mode (but it is still easy to modify to use a custom CLI entry). 2026-01-30 18:15:52 +00:00
30 changed files with 3425 additions and 789 deletions
+87 -91
View File
@@ -1,137 +1,133 @@
# fastapi-vue-setup
![FastAPI-Vue setup complete](https://raw.githubusercontent.com/LeoVasanko/fastapi-vue-setup/main/docs/banner.webp)
Tool to create or patch FastAPI project with a Vue frontend, with integrated build and development systems. The Python package will not need any JS runtime because it includes a prebuilt Vue frontend in it. For development (Vite and FastAPI auto reloads) and building the package one of npm, deno or bun is required (node is recommended due to bugs in deno and bun).
# FastAPI-Vue Full Stack Setup
## Features
Build and develop **FastAPI + Vue** as a single project, while keeping production purely Python — **JavaScript tooling is only needed during development!**
- **No JavaScript**: Your Python package can be installed and used without any JS runtime
- **Integrated build system**: Vue frontend builds into Python package during `uv build`
- **Development server**: Single command runs Vite + FastAPI with hot-reload
- **Optimized static serving**: Caching, zstd compression and SPA support
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.
## Installation
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.
Install [uv](https://docs.astral.sh/uv/getting-started/installation/), then:
## Quick start
```sh
uv tool install fastapi-vue-setup
fastapi-vue-setup --help
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:
```
Or run directly:
```sh
uvx fastapi-vue-setup my-app
```
## Usage
Inside the new project, start the development server with live reloads and debug aids:
```sh
uv run scripts/devserver.py
```
The script may be used to create an all new project folder or to patch or update an existing project to use this framework. It autodetects the project folder given and performs the appropriate actions.
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
```
### Create a new project
Or install as a proper executable:
```sh
uv tool install dist/my_app-0.1.0.tar.gz
my-app --help
```
<img src="https://raw.githubusercontent.com/LeoVasanko/fastapi-vue-setup/main/docs/hello.webp" alt='"You did it" with Vue-FastAPI connection.' width="500">
## 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
fastapi-vue-setup my-app
uvx fastapi-vue-setup [project-dir] [options]
```
This will:
| 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 |
1. Run `uv init my-app`
2. Run `create-vue frontend` (interactive - choose your Vue options)
3. Patch the project with FastAPI integration
4. Install dependencies via `uv add`
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.
### Patch an existing project
️ 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.
You should have your pyproject.toml at the current working directory, and Vue with its package.json under `frontend/`. If either one is missing, new applications will be initialised. Otherwise we only patch what can be patched without breaking your existing projects.
### 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)
```sh
fastapi-vue-setup .
scripts/devserver.py [args]
```
### CLI Options
️ 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
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:
```
fastapi-vue-setup [project-dir] [options]
Options:
--module-name NAME Python module name (auto-detected from pyproject.toml)
--dry-run Preview changes without modifying files
uv build && uv run my-app [args]
```
## Port Configuration
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`.
In development, you access the Vite dev server at `http://localhost:5173`. Vite proxies `/api/*` requests to FastAPI at port 5180. Ports and hosts of Vite and FastAPI are configurable by `devserver.py` arguments.
️ Other Python build and installation methods like `pip install` work equally well, we just prefer using UV.
In production, FastAPI serves both the API and static files at `http://localhost:5080`. Configurable by `host:port` argument with defaults set in `__main__.py`
## Project Layout
## Main CLI
If your project didn't already have `__main__.py`, we create one that runs the FastAPI app with richer configuration than what the FastAPI CLI offers. Running your module starts it in production mode, and optionally host:port may be given as argument to specify where it listens.
If you are running behind a reverse proxy like [Caddy](https://caddyserver.com/) on localhost, your app will trust the proxy headers it sends. However, if you need to configure another proxy host or IP, set `FORWARDED_ALLOW_IPS` env variable before running the server.
The devserver script depends on this CLI entry for running the backend. You will have to modify the `devserver.py` script if your app has its own incompatible main module. Note that we set FastAPI debug mode and Uvicorn reload when configured via `FASTAPI_VUE_BACKEND_URL` env variable (set by `devserver.py`), while for normal production use these stay disabled. The same variable also controls static files serving (disabled in dev mode).
## Vite Plugin Configuration
The `fastapiVue()` plugin in `vite.config.ts` accepts options to customize proxy behavior:
```js
import fastapiVue from "./vite-plugin-fastapi.js";
export default defineConfig({
plugins: [
vue(),
// Default: proxies only /api
fastapiVue(),
// Or specify custom paths to proxy to backend
fastapiVue({ paths: ["/api", "/auth", "/ws"] }),
],
});
```
The plugin reads environment `FASTAPI_VUE_BACKEND_URL` (default: `http://localhost:5180`) to determine where to proxy requests. This is set automatically by `devserver.py`.
## Project Structure
A newly created project typically looks like this:
```
my-app/
├── frontend/ # Vue application
├── frontend/ # Vue source (Node, Vite)
│ ├── src/
│ ├── vite-plugin-fastapi.js
│ ├── vite.config.js
│ ├── vite-plugin-fastapi.js # Helper plugin
│ ├── vite.config.js # Loads the plugin with app setup
│ └── package.json
├── my_app/ # Python module (files included in sdist)
├── my_app/ # Python package
│ ├── __init__.py
│ ├── __main__.py # CLI entrypoint
│ ├── app.py # FastAPI application
│ └── frontend-build/ # Built frontend (gitignored)
│ ├── __main__.py # Application CLI
│ ├── app.py # FastAPI app
│ └── frontend-build/ # Generated production frontend
├── scripts/
│ ├── devserver.py # CLI dev server (only in source tree)
│ └── fastapi-vue/
│ ├── build-frontend.py
── util.py
│ ├── devserver.py # Vite + FastAPI development server
│ └── fastapi-vue/ # Generated build/dev support
│ ├── buildhook.py
── buildutil.py
│ └── devutil.py
└── pyproject.toml
```
The project directory tree looks roughly like this after project creation or patching. The script finds your existing app module and other files and patches them with minimal changes to enable the Vue-FastAPI interconnection. New Python and Vue projects are created automatically if none exist.
Existing projects retain their own layout wherever possible; this is only the default structure.
## Development Workflow
## Runtime and build tooling
```bash
# Start dev server (runs both Vite and FastAPI)
uv run scripts/devserver.py
There are deliberately two separate pieces to the integration.
# Build for production
uv build
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.
# Run production server
uv run my-app
```
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.
## Frontend serving
This keeps the development tooling where it belongs while leaving the distributed application as a normal, self-contained Python package.
Your FastAPI app will use [fastapi-vue](https://git.zi.fi/LeoVasanko/fastapi-vue) to serve the frontend files. Refer to that package's documentation for further configuration.
️ 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

+97
View File
@@ -0,0 +1,97 @@
# FastAPI-Vue Runtime
Runtime utilities for making FastAPI apps standalone, with their own CLI entry point and facilities that make the FastAPI + Vue stack pleasant to use.
️ Use [fastapi-vue-setup](https://pypi.org/project/fastapi-vue-setup/) to set up your project. Everything below is configured automatically by it.
## Main Components
- **Frontend**: Serves static files with proper caching, compression and SPA support
- **Server**: Runs the FastAPI app from your own CLI entry point with uvicorn facilities vastly augmented
## Quickstart
Serve built frontend assets from `frontend-build/`:
```python
from pathlib import Path
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi_vue import Frontend
frontend = Frontend(Path(__file__).with_name("frontend-build"), spa=True)
@asynccontextmanager
async def lifespan(app: FastAPI):
await frontend.load()
yield
app = FastAPI(lifespan=lifespan)
# Add API routes here...
# Final catch-all route for frontend files (keep at end of file)
frontend.route(app, "/")
```
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 (fastapi_vue.Frontend)
- Designed to serve at `/`, living together with your other routes
- SPA routing: serves `index.html` for paths not otherwise handled
- RAM caching with zstd compression
- Browser caching with ETag, Last-Modified and immutable assets
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
- `index`: Index file name (default: `index.html`)
- `spa`: Serve index at any path (default: `False`)
- `catch_all`: Register a single catch-all handler instead of a route to each file; default for SPA
- `cached`: Path prefixes treated as immutable (default: `/assets/`)
- `favicon`: Optional path or glob (e.g. `/assets/logo*.png`)
- `zstdlevel`: Compression level (default: 18)
️ 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 (fastapi_vue.server)
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
from fastapi_vue import server
server.run("my_app.app:app", listen=["localhost:8000"])
```
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.
> As a deployment option, environment `FORWARDED_ALLOW_IPS` controls `X-Forwarded` trusted IPs (default: `127.0.0.1,::1` works for typical setups).
### Logging and exceptions
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 (fastapi_vue.env)
We use environment variables to pass values between program components, from devserver script setting dev mode and telling backend and frontend URLs, to your CLI, which in turn runs the FastAPI app that may also need access to this information. The variables are prefixed by the current application name to avoid conflicts. The CLI entry point should set one like `os.environ["FASTAPI_VUE"] = "MY_APP"`, before using `server.run`
The following properties read the environment and return `None` when variables haven't been set:
- `fastapi_vue.env.prefix` — the prefix itself
- `fastapi_vue.env.dev` — running in development mode, from e.g. `MY_APP_DEV=1`
- `fastapi_vue.env.vite_url`, `fastapi_vue.env.backend_url` — URLs set by the devserver
+6
View File
@@ -0,0 +1,6 @@
"""FastAPI Vue integration - serve Vue frontend from FastAPI."""
from .environ import env
from .staticfiles import Frontend
__all__ = ["Frontend", "env"]
+418
View File
@@ -0,0 +1,418 @@
"""HTTP/WebSocket access logging ASGI middleware."""
from __future__ import annotations
import http
import itertools
import logging
import time
from ipaddress import IPv6Address
from typing import TYPE_CHECKING, cast
from .termwidth import pad_display
if TYPE_CHECKING:
from uvicorn._types import (
ASGI3Application,
ASGIReceiveCallable,
ASGIReceiveEvent,
ASGISendCallable,
ASGISendEvent,
Scope,
WWWScope,
)
logger = logging.getLogger("fastapi_vue.access")
# Terminal color codes
_RESET = "\033[0m"
_STATUS_INFO = "\033[32m" # 1xx (green)
_STATUS_OK = "\033[1;92m" # 2xx (bright green)
_STATUS_REDIRECT = "\033[32m" # 3xx (green)
_STATUS_CLIENT_ERR = "\033[0;31m" # 4xx (red)
_STATUS_SERVER_ERR = "\033[1;91m" # 5xx (bold bright red)
_METHOD_READ = "\033[0;34m" # GET, HEAD, OPTIONS (blue)
_METHOD_WRITE = "\033[1;94m" # POST, PUT, DELETE, PATCH (bold bright blue)
_HOST = "\033[38;5;242m" # hostname (dark grey)
_PATH = "\033[38;5;250m" # path (white)
_TIMING = "\033[38;5;242m" # timing/devmode (dark grey)
_WS_OPEN = "\033[38;5;226m" # WebSocket connect (brightest yellow)
_WS_CLOSE = "\033[38;5;142m" # WebSocket disconnect (dimmer yellow)
def _format_duration(duration: float) -> str:
ms = int(duration * 1000)
if ms < 2000:
return f"{ms}ms"
total_seconds = ms // 1000
if total_seconds < 60:
return f"{total_seconds}s"
if total_seconds < 3600:
minutes, seconds = divmod(total_seconds, 60)
return f"{minutes}m{seconds}s"
hours, remainder = divmod(total_seconds, 3600)
minutes = remainder // 60
return f"{hours}h{minutes}m"
def _status_color(status: int) -> str:
if status < 200:
return _STATUS_INFO
if status < 300:
return _STATUS_OK
if status < 400:
return _STATUS_REDIRECT
if status < 500:
return _STATUS_CLIENT_ERR
return _STATUS_SERVER_ERR
def _method_color(method: str) -> str:
return _METHOD_READ if method in ("GET", "HEAD", "OPTIONS") else _METHOD_WRITE
def _format_extra_timing(extra: str = "", duration: float | None = None) -> tuple[str, str]:
timing = _format_duration(duration) if duration is not None else ""
return (f"{extra} " if extra else "", f"{_TIMING}{timing}{_RESET}" if timing else "")
def _format_ipv6_network(ip: str) -> str:
try:
ip = ip.strip("[]")
if "%" in ip:
ip = ip.split("%")[0]
addr = IPv6Address(ip)
if addr.is_loopback:
return "::1"
if addr.is_unspecified:
return "::"
if addr.ipv4_mapped:
return str(addr.ipv4_mapped)
if addr.is_link_local:
return str(addr)
network_int = int(addr) >> 64
groups: list[str] = []
for _ in range(4):
groups.insert(0, format(network_int & 0xFFFF, "x"))
network_int >>= 16
result = ":".join(groups) + "::"
return str(IPv6Address(result + "0")).removesuffix("::")
except ValueError:
return ip
def _format_client_ip(ip: str) -> str:
if not ip or ip == "-":
return "-"
stripped = ip.strip("[]")
if ":" in stripped:
return _format_ipv6_network(ip)
return ip
def _header(scope: WWWScope, name: str) -> str | None:
name_bytes = name.lower().encode("latin-1")
for key, value in scope["headers"]:
if key.lower() == name_bytes:
return value.decode("latin-1")
return None
def _client_host(scope: WWWScope) -> str:
client = scope["client"]
return client[0] if client else "-"
def _path(scope: WWWScope) -> str:
path = scope["path"]
query = scope["query_string"]
if query:
return f"{path}?{query.decode('latin-1')}"
return path
# WebSocket connection counter (mod 100)
_ws_counter = itertools.count()
def _next_ws_id() -> str:
return f"{next(_ws_counter) % 100:02d}"
WS_CLOSE_CODES = {
1000: "ok",
1001: "going away",
1002: "protocol error",
1003: "unsupported",
1005: "no status",
1006: "abnormal",
1007: "invalid data",
1008: "policy violation",
1009: "too large",
1010: "extension required",
1011: "server error",
1012: "restarting",
1013: "try again",
1014: "bad gateway",
1015: "tls error",
}
def _http_access_log_extra(
scope: WWWScope,
status: int,
duration: float,
extra: str = "",
method: str | None = None,
) -> dict[str, object]:
client_addr = _client_host(scope)
full_path = _path(scope)
method = method if method is not None else cast("str", scope.get("method", "-"))
method = cast("str", scope.get("state", {}).get("access_log_method") or method)
try:
status_phrase = http.HTTPStatus(status).phrase
except ValueError:
status_phrase = ""
extra, timing = _format_extra_timing(extra, duration)
return {
"client": _format_client_ip(client_addr).ljust(19),
"status": f"{_status_color(status)}{str(status).rjust(3)}{_RESET}",
"method": (
f"{_METHOD_READ}{pad_display('🔌', 7)}{_RESET}"
if method == "🔌"
else f"{_method_color(method)}{pad_display(method, 7)}{_RESET}"
),
"host": f"{_HOST}{_header(scope, 'host') or '-'}{_RESET}",
"path": f"{_PATH}{full_path}{_RESET}",
"extra": extra,
"timing": timing,
"client_addr": client_addr,
"status_code": f"{status} {status_phrase}",
"request_line": f"{method} {full_path} HTTP/{scope.get('http_version', '-')}",
"http_version": scope.get("http_version", "-"),
"full_path": full_path,
}
def _ws_open_extra(
scope: WWWScope,
ws_id: str,
origin: str | None,
extra: str = "",
) -> dict[str, object]:
client_addr = _client_host(scope)
path = scope.get("path", "")
full_path = _path(scope)
origin_host = origin.split("://", 1)[-1] if origin else None
extra, timing = _format_extra_timing(extra)
host = _header(scope, "host")
path = f"{_PATH}{path}{_RESET}"
if origin_host and origin_host != host:
path += f" {_RESET}from {_HOST}{origin_host}{_RESET}"
return {
"client": _format_client_ip(client_addr).ljust(19),
"status": f"{_WS_OPEN} {ws_id}{_RESET}",
"method": f"{_METHOD_READ}{pad_display('🔌', 7)}{_RESET}",
"host": f"{_HOST}{host}{_RESET}" if host else "",
"path": path,
"extra": extra,
"timing": timing,
"client_addr": client_addr,
"status_code": "",
"request_line": f"WebSocket {path}",
"http_version": scope.get("http_version", "-"),
"full_path": full_path,
}
def _ws_close_extra(
scope: WWWScope,
ws_id: str,
close_code: int | None,
duration: float,
extra: str = "",
) -> dict[str, object]:
client_addr = _client_host(scope)
path = scope.get("path", "-")
full_path = _path(scope)
if close_code is None:
code, status_text = "----", "unknown"
else:
code = str(close_code)
status_text = WS_CLOSE_CODES.get(close_code, f"code {close_code}")
extra, timing = _format_extra_timing(extra, duration)
return {
"client": " " * 19,
"status": f"{_WS_CLOSE} {ws_id}{_RESET}",
"method": f"{_TIMING}{pad_display('closed', 7)}{_RESET}",
"host": "",
"path": f"{code} {status_text}",
"extra": extra,
"timing": timing,
"client_addr": client_addr,
"status_code": f"{code} {status_text}",
"request_line": f"WebSocket {path}",
"http_version": scope.get("http_version", "-"),
"full_path": full_path,
}
def _ws_reject_extra(
scope: WWWScope,
close_code: int | None,
duration: float,
extra: str = "",
) -> dict[str, object]:
"""Open-format line for a connection closed before accept.
Replaces the open line (which never happened), so the client IP, host and
path stay visible. No connection id is printed (ids are only assigned on
accept); the status column shows a dim ``--`` and the close reason rides
in the extra column.
"""
if close_code is None:
code, status_text = "----", "unknown"
else:
code = str(close_code)
status_text = WS_CLOSE_CODES.get(close_code, f"code {close_code}")
fields = _ws_open_extra(scope, "--", _header(scope, "origin"))
reason = f"closed {code} {status_text}"
extra, timing = _format_extra_timing(f"{reason} {extra}" if extra else reason, duration)
fields.update(
{
"status": f"{_WS_CLOSE} --{_RESET}",
"extra": extra,
"timing": timing,
"status_code": f"{code} {status_text}",
}
)
return fields
def _assemble_access_log(fields: dict[str, object]) -> str:
return (
f"{fields['client']} {fields['status']} {fields['method']}"
f"{fields['host']}{fields['path']}{fields['extra']}{fields['timing']}"
)
class AccessLogMiddleware:
"""ASGI middleware logging HTTP and WebSocket access with colored fields."""
def __init__(self, app: ASGI3Application) -> None:
"""Store the wrapped app."""
self.app = app
async def __call__(
self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
) -> None:
"""Dispatch by scope type to HTTP/WebSocket access logging."""
if scope["type"] == "http":
return await self._handle_http(scope, receive, send)
if scope["type"] == "websocket":
return await self._handle_websocket(scope, receive, send)
return await self.app(scope, receive, send)
async def _handle_http(
self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
) -> None:
start = time.perf_counter()
www_scope = cast("WWWScope", scope)
async def wrapped_send(message: ASGISendEvent) -> None:
if message["type"] == "http.response.start":
fields = _http_access_log_extra(
www_scope,
status=message["status"],
duration=time.perf_counter() - start,
extra=www_scope.get("state", {}).get("log_extra", ""),
)
logger.info(
'%s - "%s" %s',
fields["client_addr"],
fields["request_line"],
fields["status_code"],
extra=fields,
)
await send(message)
return await self.app(scope, receive, wrapped_send)
async def _handle_websocket(
self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
) -> None:
start = time.perf_counter()
ws_id: str | None = None # assigned on accept; rejects print "--"
accepted = False
closed = False
www_scope = cast("WWWScope", scope)
origin = _header(www_scope, "origin")
def _extra() -> str:
return www_scope.get("state", {}).get("log_extra", "")
def _close_fields(message: ASGIReceiveEvent | ASGISendEvent) -> dict[str, object]:
if accepted:
assert ws_id is not None # noqa: S101 # guaranteed once accepted
return _ws_close_extra(
www_scope,
ws_id,
message.get("code"),
time.perf_counter() - start,
_extra(),
)
return _ws_reject_extra(
www_scope,
message.get("code"),
time.perf_counter() - start,
_extra(),
)
async def wrapped_send(message: ASGISendEvent) -> None:
nonlocal accepted, closed, ws_id
if message["type"] == "websocket.accept" and not accepted:
accepted = True
ws_id = _next_ws_id()
fields = _ws_open_extra(www_scope, ws_id, origin, _extra())
logger.info(_assemble_access_log(fields), extra=fields)
elif message["type"] == "websocket.http.response.start" and not closed:
closed = True
fields = _http_access_log_extra(
www_scope,
status=message["status"],
duration=time.perf_counter() - start,
extra=_extra(),
method="🔌",
)
logger.info(_assemble_access_log(fields), extra=fields)
elif message["type"] == "websocket.close" and not closed:
closed = True
fields = _close_fields(message)
logger.info(_assemble_access_log(fields), extra=fields)
await send(message)
async def wrapped_receive() -> ASGIReceiveEvent:
nonlocal closed
message = await receive()
if message["type"] == "websocket.disconnect" and not closed:
closed = True
fields = _close_fields(message)
logger.info(_assemble_access_log(fields), extra=fields)
return message
return await self.app(scope, wrapped_receive, wrapped_send)
+46
View File
@@ -0,0 +1,46 @@
"""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.
"""
import os
PREFIX_VARIABLE = "FASTAPI_VUE"
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.
"""
@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()
+119
View File
@@ -0,0 +1,119 @@
"""Parse endpoint strings for uvicorn server configuration."""
import contextlib
import ipaddress
from urllib.parse import urlparse
def _parse_all_interfaces(value: str) -> list[dict] | None:
"""Parse ':port' format to bind all interfaces."""
if not (value.startswith(":") and value != ":"):
return None
port_part = value[1:]
if not port_part.isdigit():
msg = f"Invalid port in '{value}'"
raise SystemExit(msg)
port = int(port_part)
return [{"host": "0.0.0.0", "port": port}, {"host": "::", "port": port}] # noqa: S104
def _parse_unix_socket(value: str) -> list[dict] | None:
"""Parse UNIX domain socket paths."""
if value.startswith("/"):
return [{"uds": value}]
if value.startswith("unix:"):
uds_path = value[5:] or None
if uds_path is None:
msg = "unix: path must not be empty"
raise SystemExit(msg)
return [{"uds": uds_path}]
return None
def _parse_unbracketed_ipv6(value: str, default_port: int) -> list[dict] | None:
"""Parse unbracketed IPv6 addresses."""
if value.count(":") <= 1 or value.startswith("["):
return None
try:
ipaddress.IPv6Address(value)
except ValueError as e:
msg = f"Invalid IPv6 address '{value}': {e}"
raise SystemExit(msg) from e
return [{"host": value, "port": default_port}]
def _parse_host_port(value: str, default_port: int) -> list[dict]:
"""Parse host[:port] or [ipv6][:port] using urlparse."""
parsed = urlparse(f"//{value}") # // prefix lets urlparse treat it as netloc
host = parsed.hostname or "localhost"
port = parsed.port or default_port
# Validate IP literals (optional; hostname passes through)
with contextlib.suppress(ValueError):
ipaddress.ip_address(host)
return [{"host": host, "port": port}]
def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
"""Parse an endpoint string into uvicorn bind configurations.
Args:
value: Endpoint string to parse
default_port: Port to use when not specified
Returns:
List of dicts with uvicorn bind kwargs (host/port or uds).
Two entries may be returned for IPv4 and IPv6 (all interaces).
Supported forms:
- None or empty -> [{host: "localhost", port: default_port}]
- port (numeric) -> [{host: "localhost", port: port}]
- :port -> [{host: "0.0.0.0", port}, {host: "::", port}] (all interfaces)
- host:port -> [{host, port}]
- host -> [{host, port: default_port}]
- [ipv6]:port -> [{host: ipv6, port}]
- ipv6 (unbracketed) -> [{host: ipv6, port: default_port}]
- /path or unix:/path -> [{uds: path}]
"""
if not value:
return [{"host": "localhost", "port": default_port}]
# Port only (numeric) -> localhost:port
if value.isdigit():
return [{"host": "localhost", "port": int(value)}]
# Try specialized parsers in order
result = _parse_all_interfaces(value)
if result is not None:
return result
result = _parse_unix_socket(value)
if result is not None:
return result
result = _parse_unbracketed_ipv6(value, default_port)
if result is not None:
return result
# Fallback: host[:port], [ipv6][:port]
return _parse_host_port(value, default_port)
def parse_endpoints(
listen: str | list[str] | None = None,
default_port: int = 8000,
) -> list[dict]:
"""Parse listen strings into a list of endpoint dicts.
Args:
listen: Endpoint string(s) (see parse_endpoint for formats).
default_port: Port to use when not specified in listen args.
"""
if listen is None:
listen = [f"localhost:{default_port}"]
elif isinstance(listen, str):
listen = [listen]
return [ep for s in listen for ep in parse_endpoint(s, default_port)]
+407
View File
@@ -0,0 +1,407 @@
"""Logging integration: tracerite loading and colored access log formatting.
The access log middleware supplies colored fields (``client``, ``status``,
``method``, ``host``, ``path``, ``extra``, ``timing``) via ``extra=``. When
colors are disabled the ANSI escape codes are stripped from the assembled
output so the same formatting code path produces plain text.
"""
from __future__ import annotations
import io
import logging
import os
import re
import sys
from contextlib import suppress
from copy import deepcopy
from typing import TYPE_CHECKING, Literal
import tracerite
from starlette.middleware.errors import ServerErrorMiddleware
from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse, Response
from uvicorn.config import LOGGING_CONFIG, Config
from uvicorn.lifespan.on import LifespanOn
if TYPE_CHECKING:
from typing import Any
from starlette.requests import Request
from uvicorn._types import LifespanScope
from uvicorn.lifespan.on import LifespanSendMessage
from .accesslog import AccessLogMiddleware
from .environ import env
ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m")
ACCESS_LOG_FMT = "%(client)s %(status)s %(method)s %(host)s%(path)s %(extra)s%(timing)s"
ACCESS_LOGGER = "fastapi_vue.access"
def strip_ansi(text: str) -> str:
"""Remove ANSI escape codes from text."""
return ANSI_ESCAPE_RE.sub("", text)
def use_color(stream: io.TextIOBase = sys.stderr) -> bool:
"""Test if the stream supports color codes."""
if os.environ.get("NO_COLOR"): # Non empty means no (no-color.org)
return False
if os.environ.get("FORCE_COLOR", "") not in {"", "0"}: # force-color.org, node
return True
if hasattr(stream, "isatty") and stream.isatty():
return True
with suppress(KeyError, ValueError, OSError): # Journald does color (-ocat)
dev, ino = map(int, os.environ["JOURNAL_STREAM"].split(":", 1))
st = os.fstat(stream.fileno())
return st.st_dev == dev and st.st_ino == ino
return False
_LEVEL_EMOJI = {
logging.DEBUG: "🐛",
logging.INFO: "🔷",
logging.WARNING: "",
logging.ERROR: "🛑",
logging.CRITICAL: "🚨",
}
def _level_prefix(record: logging.LogRecord) -> str:
emoji = _LEVEL_EMOJI.get(record.levelno)
return f"{emoji} " if emoji else f"{record.levelname}: "
class Formatter(logging.Formatter):
"""Formatter for both access records and ordinary log messages.
Records with the middleware's access fields (``client`` etc.) are
formatted from those; anything else gets an emoji level prefix
(``LEVEL: `` fallback for unknown levels) in place of uvicorn's
``levelprefix``.
Instantiation always loads tracerite, and with ``access=True`` also
installs the access-log middleware: ``dictConfig`` builds formatters while
uvicorn applies ``log_config``, which happens before the app is loaded —
including in reload/worker subprocesses that re-import the config without
calling ``fastapi_vue.server.run()`` again. Patching the server error
middleware here likewise propagates it to those subprocesses.
"""
def __init__(
self,
fmt: str | None = None,
datefmt: str | None = None,
style: Literal["%", "{", "$"] = "%",
use_colors: bool | None = None, # noqa: FBT001 # mirrors logging.Formatter
*,
access: bool = False,
) -> None:
"""Load tracerite, optionally install the access log, detect color support."""
tracerite.load()
tracerite.load_suppressions(
extra={"starlette.routing": "until", "fastapi.routing": "until"}
)
patch_lifespan_logging()
patch_server_error_middleware()
if access:
install_access_log()
if use_colors in (True, False):
self.use_colors = use_colors
else:
self.use_colors = use_color(sys.stdout)
super().__init__(fmt=fmt, datefmt=datefmt, style=style)
def formatMessage(self, record: logging.LogRecord) -> str: # noqa: N802
"""Format access records via middleware fields, others with an emoji prefix."""
if "client" not in record.__dict__:
return _level_prefix(record) + record.getMessage()
formatted = super().formatMessage(record)
if not self.use_colors:
formatted = strip_ansi(formatted)
return formatted
class WebSocketChatterFilter(logging.Filter):
"""Drop stock uvicorn WebSocket handshake/chatter records.
Stock uvicorn logs WS handshakes (``'%s - "WebSocket %s" ...'``) and the
websockets library's "connection open/closed" chatter to ``uvicorn.error``,
ungated by ``access_log``. Our middleware logs WebSockets itself.
"""
_PREFIXES = ('%s - "WebSocket ', "connection open", "connection closed", "connection rejected")
def filter(self, record: logging.LogRecord) -> bool:
"""Keep records not matching stock WebSocket chatter prefixes."""
msg = record.msg
if not isinstance(msg, str):
return True
return not msg.startswith(self._PREFIXES)
class UvicornQuietFilter(logging.Filter):
"""Silence uvicorn's routine chatter (startup/shutdown lines, etc.).
Handler-side, not a logger level: uvicorn's ``configure_logging``
re-applies ``log_level`` to its loggers after ``dictConfig``, which would
override a level lifted in the config dict.
"""
def filter(self, record: logging.LogRecord) -> bool:
"""Drop uvicorn records below WARNING."""
return not (record.name.startswith("uvicorn") and record.levelno < logging.WARNING)
_installed = False
def install_access_log() -> None:
"""Wrap apps loaded by uvicorn with AccessLogMiddleware, once per process.
The guard is deliberately module-level: reload/worker subprocesses
re-import this module, resetting it so the patch is re-applied there.
"""
global _installed # noqa: PLW0603 # deliberately module-level, see docstring
if _installed:
return
_installed = True
original_load = Config.load
def load(self): # noqa: ANN001, ANN202
original_load(self)
if not isinstance(self.loaded_app, AccessLogMiddleware):
self.loaded_app = AccessLogMiddleware(self.loaded_app)
Config.load = load # type: ignore[method-assign]
_lifespan_patched = False
def patch_lifespan_logging() -> None:
"""Patch uvicorn's LifespanOn to log lifespan failures with exc_info.
Starlette formats lifespan exceptions into a plain-text ASGI message,
which uvicorn logs as-is without exc_info, while the exc_info-carrying
log in ``LifespanOn.main()`` is skipped when a failure message was sent.
This suppresses the text message and always logs the exception with
exc_info, so tracerite (or any exc_info-aware handler) renders the
traceback. Monkeypatches uvicorn internals; written against uvicorn 0.52.
"""
global _lifespan_patched # noqa: PLW0603 # once-per-process, resets in subprocesses
if _lifespan_patched:
return
_lifespan_patched = True
original_send = LifespanOn.send
async def send(self: LifespanOn, message: LifespanSendMessage) -> None:
# Drop the pre-formatted traceback text; main() logs the exception itself.
if message["type"] in ("lifespan.startup.failed", "lifespan.shutdown.failed"):
message = dict(message) # type: ignore[assignment]
message.pop("message", None)
await original_send(self, message)
async def main(self: LifespanOn) -> None:
"""Mirror upstream LifespanOn.main, but always log failures with exc_info."""
try:
app = self.config.loaded_app
scope: LifespanScope = {
"type": "lifespan",
"asgi": {"version": self.config.asgi_version, "spec_version": "2.0"},
"state": self.state,
}
await app(scope, self.receive, self.send)
except BaseException:
self.asgi = None
self.error_occurred = True
if self.startup_failed or self.shutdown_failed or self.config.lifespan != "auto":
phase = "shutdown" if self.shutdown_failed else "startup"
self.logger.exception("Uncaught exception during application %s", phase)
else:
self.logger.info("ASGI 'lifespan' protocol appears unsupported.")
finally:
self.startup_event.set()
self.shutdown_event.set()
LifespanOn.send = send # type: ignore[method-assign]
LifespanOn.main = main # type: ignore[method-assign]
_server_error_patched = False
DEBUG_INGRESS = """This page is shown for your guidance because the application is \
running in debug mode and has crashed handling this request."""
def _generate_html(exc: Exception) -> str:
return tracerite.html_page(
exc,
title="FastAPI debugger",
heading="500 Server Error",
ingress=DEBUG_INGRESS,
)
def _generate_plain_text(exc: Exception) -> str:
buffer = io.StringIO()
tracerite.tty_traceback(exc, file=buffer)
return buffer.getvalue()
def _generate_json(exc: Exception) -> dict[str, Any]:
chain = tracerite.extract_chain(exc)
return {"detail": "Internal Server Error", "traceback": chain}
def patch_server_error_middleware() -> None:
"""Patch Starlette's ServerErrorMiddleware to format debug errors with tracerite.
Starlette's debug responses use its own static HTML traceback template.
This replaces ``debug_response`` with tracerite renderers (source
context, locals, chained exceptions), adds ``accept: application/json``
handling, and returns a JSON body also for non-debug errors when
requested. Only apps running with ``debug=True`` produce traceback
responses. Monkeypatches Starlette internals; written against
starlette 1.6.
"""
global _server_error_patched # noqa: PLW0603 # once-per-process, resets in subprocesses
if _server_error_patched:
return
_server_error_patched = True
def debug_response(
self: ServerErrorMiddleware, # noqa: ARG001
request: Request,
exc: Exception,
) -> Response:
accept = request.headers.get("accept", "")
if "text/html" in accept:
return HTMLResponse(_generate_html(exc), status_code=500)
if "application/json" in accept:
return JSONResponse(_generate_json(exc), status_code=500)
return PlainTextResponse(_generate_plain_text(exc), status_code=500)
def error_response(
self: ServerErrorMiddleware, # noqa: ARG001
request: Request,
exc: Exception, # noqa: ARG001 # signature mirrors Starlette's
) -> Response:
if "application/json" in request.headers.get("accept", ""):
return JSONResponse({"detail": "Internal Server Error"}, status_code=500)
return PlainTextResponse("Internal Server Error", status_code=500)
ServerErrorMiddleware.debug_response = debug_response # type: ignore[method-assign]
ServerErrorMiddleware.error_response = error_response # type: ignore[method-assign]
def _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 patch_log_config(log_config, *, access_log: bool = True): # noqa: ANN001, ANN201
"""Patch a uvicorn log_config dict for our logging, best-effort.
A dict without a ``version`` key is treated as a partial config: it is
merged over uvicorn's default dict, so only the customizations are
needed (e.g. ``{"loggers": {"kanta": {"level": "DEBUG"}}}``). A dict
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
loads tracerite in every process uvicorn applies the config in, filters
on the default handler dropping stock uvicorn's WebSocket chatter and
routine INFO lines, an emoji-level-prefix Formatter in place of
uvicorn's stock ``default`` formatter (a user-supplied one wins), a root
logger entry so ``logging.info()`` et al. print through the default
handler 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
detected" line is dropped while the WARNING "Reloading..." line (logged
to ``uvicorn.error``) still shows; a user-supplied level wins.
With ``access_log``, additionally rewires the ``access`` formatter to
our Formatter and attaches its handler to our ``fastapi_vue.access``
logger. We must not
attach handlers to ``uvicorn.access``: uvicorn gates its own
protocol-level access logging on ``uvicorn.access.hasHandlers()``.
"""
if not isinstance(log_config, dict):
return log_config
config = deepcopy(log_config)
if "version" not in config:
config = _merge_log_config(deepcopy(LOGGING_CONFIG), config)
with suppress(Exception):
config["formatters"]["fastapi_vue"] = {"()": "fastapi_vue.logging.Formatter"}
config["handlers"]["fastapi_vue"] = {
"class": "logging.NullHandler",
"formatter": "fastapi_vue",
}
with suppress(Exception):
filters = config.setdefault("filters", {})
filters["ws_chatter"] = {"()": "fastapi_vue.logging.WebSocketChatterFilter"}
filters["uvicorn_quiet"] = {"()": "fastapi_vue.logging.UvicornQuietFilter"}
handler_filters = config["handlers"]["default"].setdefault("filters", [])
for name in ("ws_chatter", "uvicorn_quiet"):
if name not in handler_filters:
handler_filters.append(name)
# Emoji level prefixes for ordinary logs, replacing uvicorn's stock
# default formatter; a user-supplied default formatter is left alone.
with suppress(Exception):
default = config["formatters"]["default"]
if default == LOGGING_CONFIG["formatters"]["default"]:
config["formatters"]["default"] = {
"()": "fastapi_vue.logging.Formatter",
"fmt": "%(message)s",
"use_colors": None,
}
# uvicorn's default config leaves the root logger handlerless, eating
# logging.info() et al.; route root through uvicorn's default handler.
# 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):
root = config.setdefault("root", {})
root.setdefault("level", "INFO" if env.dev else "WARNING")
if "default" in config.get("handlers", {}):
root_handlers = root.setdefault("handlers", [])
if "default" not in root_handlers:
root_handlers.append("default")
# watchfiles logs "N changes detected" to its own logger at INFO; only the
# WARNING "Reloading..." line (uvicorn.error) should show.
with suppress(Exception):
config.setdefault("loggers", {}).setdefault("watchfiles.main", {}).setdefault(
"level", "WARNING"
)
if access_log:
with suppress(Exception):
config["formatters"]["access"] = {
"()": "fastapi_vue.logging.Formatter",
"fmt": ACCESS_LOG_FMT,
"use_colors": None,
"access": True,
}
with suppress(Exception):
if "access" in config["handlers"]:
config.setdefault("loggers", {})[ACCESS_LOGGER] = {
"handlers": ["access"],
"level": "INFO",
"propagate": False,
}
return config
+266
View File
@@ -0,0 +1,266 @@
"""Uvicorn server runner with multi-endpoint support."""
import asyncio
import importlib.metadata
import logging
import os
import socket
from contextlib import suppress
from pathlib import Path
from typing import Any
import tracerite
import uvicorn
from uvicorn import Config, Server
from uvicorn.main import STARTUP_FAILURE
from uvicorn.supervisors import ChangeReload, Multiprocess
from .environ import env
from .hostutil import parse_endpoints
from .logging import (
install_access_log,
patch_lifespan_logging,
patch_log_config,
patch_server_error_middleware,
use_color,
)
from .startupbox import print_box
tracerite.load() # Early load on CLI load (import server); uvicorn workers reload via log config
# Install force color to aid tracerite and any external software to use full color when available
# Define NO_COLOR or FORCE_COLOR beforehand to avoid this
if "FORCE_COLOR" not in os.environ and use_color():
os.environ["FORCE_COLOR"] = "3"
logger = logging.getLogger(__name__)
_WILDCARD_HOSTS = frozenset({"0.0.0.0", "::"}) # noqa: S104
_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:
"""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
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:
host = endpoint.get("host")
if host is None:
continue
if host in _WILDCARD_HOSTS:
host = "localhost"
elif ":" in host: # IPv6 literal
host = f"[{host}]"
return f"http://{host}:{endpoint['port']}"
return "http://localhost"
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.
Available fields: ``{module}`` (top-level package of the app path),
``{name}`` (module with spaces instead of underscores), ``{Name}``
(also capitalized), ``{version}`` (from installed package metadata,
"dev" when not installed), ``{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]
name = module.replace("_", " ")
try:
version = importlib.metadata.version(module)
except importlib.metadata.PackageNotFoundError:
version = ""
values = {
"module": module,
"name": name,
"Name": name.title(),
"version": version,
"listen": _listen_addresses(endpoints),
"url": _connect_url(endpoints),
}
print_box(template.format_map(values))
def run( # noqa: PLR0913
app: str,
*,
listen: str | list[str] | None = None,
default_port: int = 8000,
reload: bool | Path = False,
workers: int | None = None,
access_log: bool = True,
startup_box: str | None = "{Name} {version} @ {listen}\n{url}",
log_config: Any = uvicorn.config.LOGGING_CONFIG, # noqa: ANN401
**uvicorn_config: Any, # noqa: ANN401
) -> None:
"""Run uvicorn server(s) for the given app.
Args:
app: The ASGI application path (e.g., "myapp.main:app")
listen: Endpoint string(s) (see parse_endpoint for formats).
default_port: Port to use when not specified in listen args.
reload: Enable auto-reload. If a Path is given, reload watches that
directory. True enables reload without setting a reload directory.
False disables reload and clears any reload_dirs.
workers: Number of worker processes (requires uvicorn.run, single endpoint only).
access_log: Enable our colored HTTP/WebSocket access logging middleware
(uvicorn's own access logging is always bypassed).
startup_box: Template for the startup box printed to stderr before
serving (see print_startup_box for fields), None to not print it.
log_config: Logging config passed to uvicorn. Dict configs are patched
best-effort (see fastapi_vue.logging.patch_log_config): tracerite
loading and WebSocket chatter filtering are always installed, and
when access_log is enabled the access formatting is rewired too.
**uvicorn_config: Additional uvicorn config options (overrides all other settings).
"""
endpoints = parse_endpoints(listen, default_port)
if not endpoints:
msg = "No endpoints to serve; check listen configuration"
raise ValueError(msg)
if startup_box:
print_startup_box(startup_box, app, endpoints)
if isinstance(reload, Path):
uvicorn_config["reload_dirs"] = [str(reload)]
elif not reload:
uvicorn_config.pop("reload_dirs", None)
if access_log:
install_access_log()
patch_lifespan_logging()
patch_server_error_middleware()
uvicorn_config["access_log"] = False # We always bypass uvicorn's own access logging
uvicorn_config["log_config"] = patch_log_config(log_config, access_log=access_log)
conf: dict[str, object] = {"app": app, "reload": bool(reload), "workers": workers}
proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1")
if proxy:
conf["proxy_headers"] = True
conf["forwarded_allow_ips"] = proxy
conf.update(uvicorn_config)
with suppress(KeyboardInterrupt, asyncio.CancelledError):
if reload or workers:
serve_multiprocess(endpoints, **conf)
else:
asyncio.run(serve(endpoints, **conf))
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
"""Serve the given endpoints in current process/loop. Does not spawn extra processes."""
forbidden = {"reload", "workers"} & {k for k, v in kwargs.items() if v}
if forbidden:
logger.warning(
"Options %s have no effect in simple mode (multiple endpoints)",
", ".join(sorted(forbidden)),
)
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
"""Serve using uvicorn supervisors for reload/workers support."""
config = Config(**kwargs)
server = Server(config)
sockets = _bind_sockets(endpoints)
try:
if config.should_reload:
ChangeReload(config, target=server.run, sockets=sockets).run()
else:
Multiprocess(config, sockets=sockets).run()
finally:
_remove_uds_files(endpoints)
+20
View File
@@ -0,0 +1,20 @@
"""Startup banner: a rounded unicode box printed to stderr (not via logging)."""
import sys
from .termwidth import display_width, pad_display
def print_box(text: str) -> None:
"""Print text in a rounded unicode box sized to its longest line.
The text is split on newlines as-is (not trimmed). Padding accounts
for ANSI colors and unicode display width (see termwidth.display_width).
"""
lines = text.split("\n")
width = max(map(display_width, lines))
border = "" * (width + 2)
out = [f"{border}"]
out.extend(f"{pad_display(line, width)}" for line in lines)
out.append(f"{border}")
sys.stderr.write("\n".join(out) + "\n")
+296
View File
@@ -0,0 +1,296 @@
"""FastAPI static file serving with zstd compression and SPA support."""
from __future__ import annotations
import fnmatch
import logging
import mimetypes
import time
from base64 import urlsafe_b64encode
from functools import partial
from pathlib import Path, PurePath, PurePosixPath
from wsgiref.handlers import format_date_time
from blake3 import blake3
from fastapi import FastAPI, Request, Response
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import JSONResponse, RedirectResponse
from starlette.exceptions import HTTPException
from starlette.routing import Route
from zstandard import ZstdCompressor
from .environ import env
logger = logging.getLogger(__name__)
__all__ = ["Frontend"]
class Assets:
"""Default cached value to /assets/."""
@staticmethod
def parse(cached: str | list[str] | Assets) -> list[str]:
match cached:
case Assets():
return ["/assets/"]
case str():
return [cached]
case list():
return cached
case _:
msg = f"Invalid cached value: {cached!r}"
raise ValueError(msg)
class Frontend:
"""Static file server with automatic zstd compression and caching.
Features:
- Automatic zstd compression for compressible files
- ETag-based caching of immutable assets
- SPA (Single Page Application) support
- /favicon.ico with correct MIME type (image/png etc)
- Dev mode: indexes files but returns error directing to Vite server
Args:
directory: Path to the directory containing static files
index: Name of the index file (default: "index.html")
spa: Enable SPA mode - serve index.html for unknown routes (default: False)
cached: Path prefixes that are immutable (default: "/assets/")
favicon: Wildcard path to favicon. E.g. /assets/logo*.png matches Vite output
zstdlevel: Zstd compression level (default: 18)
"""
def __init__( # noqa: PLR0913
self,
directory: Path | str,
*,
index: str = "index.html",
spa: bool = False,
catch_all: bool | None = None,
cached: str | list[str] | Assets | None = None,
favicon: str | None = None,
zstdlevel: int = 18,
) -> None:
"""Initialize Frontend with given configuration."""
self.www: dict[str, tuple[bytes, bytes | None, dict]] = {}
self.base: Path = Path(directory)
self.index = index
self.spa = spa
self._catch_all = spa if catch_all is None else catch_all
self.cached_paths = Assets.parse(cached if cached is not None else Assets())
self.zstdlevel = zstdlevel
self.favicon = favicon
self._app: FastAPI | None = None
self._mount_path: str = ""
self._ridx: int = 0
self._routes: list[Route] = []
def _index_only(self) -> set[str]:
"""Index file paths without loading content (for dev mode)."""
paths: set[str] = set()
if not self.base.exists():
return paths
queue = [PurePath()]
while queue:
current = self.base / queue.pop(0)
for p in current.iterdir():
rel = p.relative_to(self.base)
if p.is_dir():
queue.append(rel)
continue
name = "/" + rel.as_posix()
name = name.removesuffix(self.index)
paths.add(name)
if self.favicon:
p = PurePosixPath(self.favicon)
base = str(p.with_suffix(""))
ext = p.suffix
if any(path.startswith(base) and path.endswith(ext) for path in paths):
paths.add("/favicon.ico")
return paths
def _load(self) -> dict[str, tuple[bytes, bytes | None, dict]]:
"""Load static files from disk with compression."""
www: dict[str, tuple[bytes, bytes | None, dict]] = {}
if not self.base.exists():
logger.error(
"Missing %s - no frontend (try uv build)",
self.base,
)
else:
paths = [PurePath()]
while paths:
current = self.base / paths.pop(0)
for p in current.iterdir():
rel = p.relative_to(self.base)
if p.is_dir():
paths.append(rel)
continue
# Read file
name = "/" + rel.as_posix()
mime = mimetypes.guess_type(name)[0] or "application/octet-stream"
name = name.removesuffix(self.index)
data = p.read_bytes()
etag = urlsafe_b64encode(blake3(data).digest(9)).decode()
if mime.startswith("text/"):
mime += "; charset=UTF-8"
mtime = p.stat().st_mtime
cached = any(name.startswith(prefix) for prefix in self.cached_paths)
headers = {
"etag": f'"{etag}"',
"last-modified": format_date_time(mtime),
"cache-control": ("max-age=31536000, immutable" if cached else "no-cache"),
"content-type": mime,
}
zstd = ZstdCompressor(self.zstdlevel).compress(data)
if len(zstd) >= len(data):
zstd = None
www[name] = data, zstd, headers
if self.favicon and (m := fnmatch.filter(www, self.favicon)):
data, zstd, headers = www[m[0]]
if "immutable" in headers.get("cache-control", ""):
headers = {**headers, "cache-control": "max-age=86400"}
www["/favicon.ico"] = data, zstd, headers
if not www:
msg = "Frontend files missing, check your installation.\n"
www["/"] = (
msg.encode(),
None,
{
"etag": "error",
"content-type": "text/plain",
"cache-control": "no-store",
},
)
return www
async def load(self, *, debug: bool | None = None, log: bool = True) -> None:
"""Load or reload static files from disk.
In debug mode, returns 409 instead of files (avoid accidental use of stale builds)
If debug is None, uses app.debug (app passed to frontend.route)
"""
if debug is None:
debug = getattr(self._app, "debug", False)
if debug:
# Dev mode: just index paths, no content loading
self._devmode_paths = await run_in_threadpool(self._index_only)
self._register_routes()
return
start = time.perf_counter()
self.www = await run_in_threadpool(self._load)
self._register_routes()
duration = time.perf_counter() - start
if not log:
return
compfiles = [(len(d), len(z)) for d, z, _ in self.www.values() if z]
raw = sum(v[0] for v in compfiles)
comp = sum(v[1] for v in compfiles)
ratio = comp / raw * 100 if raw else 100.0
if log and self.www:
logger.info(
"%s: %d files in %.1f ms | zstd %d files %.2f->%.2f MB (%.0f %%)",
self.base.name,
len(self.www),
1000 * duration,
len(compfiles),
1e-6 * raw,
1e-6 * comp,
ratio,
)
if self.favicon and "/favicon.ico" not in self.www:
logger.warning("Favicon not found: %s", self.favicon)
def route(self, app: FastAPI, mount_path: str = "/") -> None:
"""Register frontend routes with a FastAPI app.
In SPA/catch-all mode, this must only be called only after all other routes.
The calling position determines routing priority, although in regular mode the
routes are actually added only after load() is called.
Args:
app: FastAPI application instance
mount_path: Path where the frontend should be mounted (default: "/")
"""
self._app = app
self._mount_path = mount_path.rstrip("/")
self._ridx = len(app.routes)
if self._catch_all:
# Register catch-all immediately (works without load)
path = self._mount_path + "{path:path}"
app.api_route(path, methods=["GET", "HEAD"], name="frontend", response_model=None)(
self.handle
)
def _register_routes(self) -> None:
"""Register individual routes for each loaded file (non-catch_all mode)."""
if self._app is None or self._catch_all:
return
# Remove previously registered routes (for reload support)
for route in list(self._routes):
if route in self._app.routes:
self._app.routes.remove(route)
self._routes.clear()
# Get paths and select handler based on mode (checked once, not per request)
debug = getattr(self._app, "debug", False)
paths = self._devmode_paths if debug else self.www.keys()
handler = _devmode_respond if debug else self._respond
# Insert at the position where route() was called
self._app.routes[self._ridx : self._ridx] = self._routes = [
Route(
self._mount_path + p,
endpoint=handler if debug else partial(handler, name=p),
methods=["GET", "HEAD"],
name=f"frontend{p.replace('/', '_')}",
)
for p in paths
]
def _respond(self, request: Request, name: str) -> Response:
"""Serve a static file with ETag and compression support."""
data, zstd, headers = self.www[name]
if request.headers.get("if-none-match") == headers["etag"]:
return Response(status_code=304, headers=headers)
if zstd and "zstd" in request.headers.get("accept-encoding", ""):
return Response(
content=zstd,
headers={**headers, "content-encoding": "zstd"},
)
return Response(content=data, headers=headers)
def handle(self, request: Request, path: str) -> Response | RedirectResponse:
"""SPA catch-all handler with directory redirects and fallback to index."""
name = path.removesuffix(self.index)
debug = getattr(self._app, "debug", False)
files = self._devmode_paths if debug else self.www
if name not in files:
# Friendly redirect for directories missing trailing slash
if name and f"{name}/" in files:
return RedirectResponse(request.url.path + "/")
# SPA support: serve / for unknown paths if the browser wants HTML
if self.spa and "text/html" in request.headers.get("accept", ""):
name = "/"
# 404 for everything else
if name not in files:
raise HTTPException(status_code=404)
return (_devmode_respond if debug else self._respond)(request, name)
def _devmode_respond(_request: Request, _name: str = "") -> JSONResponse:
"""Return error response directing to Vite server."""
at = f" at {env.vite_url}" if env.vite_url else ""
return JSONResponse(
status_code=409,
content={"detail": f"[devmode] Use Vite devserver{at} instead."},
)
+37
View File
@@ -0,0 +1,37 @@
"""Terminal display width calculation for unicode and ANSI-colored text."""
import re
import unicodedata
ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;:]*[A-Za-z]")
def _is_wide(char: str) -> bool:
"""Return True for characters rendered as two terminal columns."""
if unicodedata.east_asian_width(char) in {"F", "W"}:
return True
cp = ord(char)
return unicodedata.category(char) == "So" and (
0x2600 <= cp <= 0x27BF or 0x1F300 <= cp <= 0x1F9FF or 0x1FA00 <= cp <= 0x1FAFF
)
def display_width(text: str) -> int:
"""Calculate the display width of a string in terminal columns.
ANSI escape codes are ignored. Wide characters (East Asian F/W and
emoji) count as two columns, combining marks and format characters
(e.g. emoji variation selectors) as zero.
"""
plain = ANSI_ESCAPE_RE.sub("", text)
width = 0
for char in plain:
if unicodedata.category(char) in {"Mn", "Mc", "Me", "Cf"}:
continue
width += 2 if _is_wide(char) else 1
return width
def pad_display(text: str, width: int) -> str:
"""Pad text with trailing spaces to the given display width (no truncation)."""
return text + " " * max(width - display_width(text), 0)
+25
View File
@@ -0,0 +1,25 @@
[project]
name = "fastapi-vue"
dynamic = ["version"]
description = "Serves Vue assets on a FastAPI app. Use fastapi-vue-setup tool to add Vue build to your package."
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"fastapi>=0.115.0",
"zstandard>=0.23.0",
"blake3>=1.0.8",
"tracerite>=2.6.5",
]
[project.urls]
Homepage = "https://vasanko.com/coders/fastapi-vue"
Repository = "https://git.zi.fi/LeoVasanko/fastapi-vue-setup"
Issues = "https://github.com/LeoVasanko/fastapi-vue-setup"
[build-system]
requires = ["hatchling", "hatch-vcs"]
build-backend = "hatchling.build"
[tool.hatch.version]
source = "vcs"
raw-options.root = ".."
+1
View File
@@ -0,0 +1 @@
"""Tests for the fastapi_vue package."""
+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"
+911 -281
View File
File diff suppressed because it is too large Load Diff
+26 -3
View File
@@ -9,12 +9,14 @@ description = "Tool to create or patch FastAPI+Vue projects with integrated buil
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"ruff>=0.14.13",
"tomlkit>=0.12.0",
]
[project.urls]
Homepage = "https://git.zi.fi/LeoVasanko/fastapi-vue-setup"
Repository = "https://github.com/LeoVasanko/fastapi-vue-setup"
Homepage = "https://vasanko.com/coders/fastapi-vue"
Repository = "https://git.zi.fi/LeoVasanko/fastapi-vue-setup"
Issues = "https://github.com/LeoVasanko/fastapi-vue-setup"
[project.scripts]
fastapi-vue-setup = "fastapi_vue_setup:main"
@@ -26,4 +28,25 @@ source = "vcs"
include = ["fastapi_vue_setup.py", "template/**/*", "_version.py"]
[dependency-groups]
dev = ["ruff"]
dev = ["ruff", "fastapi-vue"]
[tool.uv.sources]
fastapi-vue = { path = "fastapi-vue", editable = true }
[tool.uv.workspace]
members = [
"f",
]
[tool.ruff]
line-length = 100
[tool.ruff.lint]
select = ["ALL"]
ignore = ["CPY", "D203", "D213", "COM812", "PLR2004"]
[tool.ruff.lint.per-file-ignores]
"template/**" = ["F821"] # Undefined names are template placeholders
"template/scripts/devserver.py" = ["N806"] # MODULE_NAME is a template variable
"fastapi_vue_setup.py" = ["PLR", "C901", "T201", "RUF001"]
"**/tests/**" = ["S101"] # Asserts are the point of tests
+28
View File
@@ -0,0 +1,28 @@
#!/usr/bin/env -S uv run
"""Build and release both fastapi-vue and fastapi-vue-setup packages."""
import shutil
import subprocess
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
DIST = ROOT / "dist"
FASTAPI_VUE = ROOT / "fastapi-vue"
def main() -> None:
"""Build both packages to dist directory."""
# Clear the dist directory
if DIST.exists():
shutil.rmtree(DIST)
DIST.mkdir()
# Build fastapi-vue (subdirectory) to root dist
subprocess.run(["uv", "build", "--out-dir", str(DIST)], cwd=FASTAPI_VUE, check=True) # noqa: S603, S607
# Build fastapi-vue-setup (root)
subprocess.run(["uv", "build", "--out-dir", str(DIST)], cwd=ROOT, check=True) # noqa: S603, S607
if __name__ == "__main__":
main()
+1
View File
@@ -0,0 +1 @@
"""Backend package with FastAPI application and Vue frontend integration."""
+19 -40
View File
@@ -1,54 +1,33 @@
# auto-upgrade@fastapi-vue-setup - remove this if you modify this file
"""Command-line entry point for running the backend server."""
import argparse
import asyncio
import os
from pathlib import Path
import uvicorn
from fastapi_vue.hostutil import parse_endpoint
from uvicorn import Config, Server
from fastapi_vue import env, server
from .APP_MODULE import APP_VAR
DEFAULT_PORT = 5080
DEFAULT_PORT = TEMPLATE_DEFAULT_PORT
os.environ["FASTAPI_VUE"] = "ENVPREFIX"
def run_server(endpoints: list[dict], *, proxy="", devmode=False):
conf: dict[str, object] = {"app": "MODULE_NAME.APP_MODULE:APP_VAR"}
if proxy:
conf["proxy_headers"] = True
conf["forwarded_allow_ips"] = proxy
if devmode:
conf["reload"] = True
conf["reload_dirs"] = ["MODULE_NAME"]
APP_VAR.debug = True
if len(endpoints) > 1:
# Run separate servers for multiple endpoints
async def serve_all():
async with asyncio.TaskGroup() as tg:
for ep in endpoints:
tg.create_task(Server(Config(**conf, **ep)).serve())
asyncio.run(serve_all())
else:
uvicorn.run(**conf, **endpoints[0])
def main():
def main() -> None:
"""Run the backend server with optional arguments."""
parser = argparse.ArgumentParser(description="Run the MODULE_NAME server.")
parser.add_argument(
"endpoint",
nargs="?",
help=(
f"Endpoint (default: localhost:{DEFAULT_PORT}). "
"Forms: host:port | :port | [ipv6]:port | ip | host | unix:/path.sock"
),
"-l",
"--listen",
action="append",
help=(f"Endpoint (default: localhost:{DEFAULT_PORT})."),
)
args = parser.parse_args()
proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1")
devmode = bool(os.getenv("FASTAPI_VUE_FRONTEND_URL"))
endpoints = parse_endpoint(args.endpoint, DEFAULT_PORT)
run_server(endpoints, proxy=proxy, devmode=devmode)
server.run(
"APP_MODULE:APP_VAR",
listen=args.listen,
default_port=DEFAULT_PORT,
server_header=False,
reload=Path(__file__).parent if env.dev else False,
)
if __name__ == "__main__":
+9 -5
View File
@@ -1,21 +1,24 @@
"""FastAPI application module with Vue frontend integration."""
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI
from fastapi_vue import Frontend
from fastapi_vue import Frontend, env
# Vue Frontend static files
frontend = Frontend(Path(__file__).with_name("frontend-build"), cached=["/assets/"])
frontend = Frontend(Path(__file__).with_name("frontend-build"))
@asynccontextmanager
async def lifespan(app: FastAPI):
async def lifespan(_app: FastAPI) -> AsyncGenerator:
"""Manage app startup and shutdown resources."""
await frontend.load()
yield
app = FastAPI(title="PROJECT_TITLE", lifespan=lifespan)
app = FastAPI(title="PROJECT_TITLE", debug=env.dev, lifespan=lifespan)
# Add API routes here...
@@ -23,7 +26,8 @@ app = FastAPI(title="PROJECT_TITLE", lifespan=lifespan)
# Health check endpoint for the Vue demo app to verify the backend is running
@app.get("/api/health")
async def health_check():
async def health_check() -> dict:
"""Return backend status for health monitoring."""
return {"status": "ok"}
+9 -6
View File
@@ -1,17 +1,19 @@
/**
* FastAPI-Vue Vite Plugin
* auto-upgrade@fastapi-vue-setup -- remove this if you edit the plugin
*
* Configures Vite for FastAPI backend integration:
* - Proxies /api/* requests to the FastAPI backend
* - Builds to the Python module's frontend-build directory
* - Disables Vite's screen clearing on startup
*
* Environment variables (with defaults):
* FASTAPI_VUE_BACKEND_URL=http://localhost:5180 - Backend API URL for proxying
* Options:
* paths - Array of paths to proxy (default: ['/api'])
*/
const backendUrl = process.env.FASTAPI_VUE_BACKEND_URL || "http://localhost:5180"
export default function fastapiVue({ paths = ['/api'] } = {}) {
const backendUrl = process.env.ENVPREFIX_BACKEND_URL || 'http://localhost:TEMPLATE_DEV_PORT'
export default function fastapiVue({ paths = ["/api"] } = {}) {
// Build proxy configuration for each path
const proxy = {}
for (const path of paths) {
@@ -23,11 +25,12 @@ export default function fastapiVue({ paths = ["/api"] } = {}) {
}
return {
name: "fastapi-vite",
name: 'vite-plugin-fastapi-MODULE_NAME',
config: () => ({
clearScreen: false,
server: { proxy },
build: {
outDir: "../MODULE_NAME/frontend-build",
outDir: '../MODULE_NAME/frontend-build',
emptyOutDir: true,
},
}),
+60 -242
View File
@@ -1,276 +1,94 @@
#!/usr/bin/env -S uv run
# auto-upgrade@fastapi-vue-setup - remove this if you modify this file
"""Run Vite development server for frontend and FastAPI backend with auto-reload.
Usage:
uv run scripts/devserver.py [host:port] [--backend host:port]
The optional host:port argument sets where the Vite frontend listens.
Supported forms: host[:port], :port (all interfaces), or just port.
The --backend option sets where the FastAPI backend listens (default: localhost:5180).
Environment:
JS_RUNTIME Path or name of JS runtime to use (deno, npm/node or bun).
FASTAPI_VUE_FRONTEND_URL Set by this script for the backend to know where Vite is.
"""
"""Run Vite development server for Vue app and FastAPI backend with auto-reload."""
import argparse
import asyncio
import contextlib
import os
import subprocess
import sys
from pathlib import Path
from sys import stderr
import httpx
from fastapi_vue.hostutil import parse_endpoint
import tracerite
exec((Path(__file__).parent / "fastapi-vue/util.py").read_text("UTF-8")) # noqa: S102
# Import util.py from scripts/fastapi-vue (not a package, so we adjust sys.path)
sys.path.insert(0, str(Path(__file__).with_name("fastapi-vue")))
from devutil import (
ProcessGroup,
check_ports_free,
logger,
ready,
setup_cli,
setup_vite,
)
DEFAULT_VITE_PORT = 5173
DEFAULT_BACKEND_PORT = 5180
FRONTEND_PATH = Path(__file__).parent.parent / "frontend"
EPILOG = """
scripts/devserver.py # Default ports on localhost
scripts/devserver.py 3000 # Vite on localhost:3000
scripts/devserver.py :3000 --backend 8000 # *:3000, localhost:8000
"""
BUN_BUG = """\
┃ ⚠️ Bun cannot correctly proxy API requests to the backend.
┃ Bug report: https://github.com/oven-sh/bun/issues/9882
┃ Consider using deno or npm instead for development.
"""
def resolve_frontend_tools(
vite_port: int, all_ifaces: bool
) -> tuple[list[str], list[str], str]:
"""Resolve frontend install and dev commands.
Returns (install_cmd, dev_cmd, tool_name).
Raises SystemExit if tools are not available.
"""
if not (FRONTEND_PATH / "package.json").exists():
stderr.write(f"┃ ⚠️ Frontend source not found at {FRONTEND_PATH}\n")
raise SystemExit(1)
result = find_js_runtime() # noqa # type: ignore
if result is None:
if not os.environ.get("JS_RUNTIME"):
stderr.write("┃ ⚠️ deno, npm or bun needed to run the frontend server.\n")
raise SystemExit(1)
tool, name = result
install_args = {
"deno": ("install", "--quiet", "--allow-scripts=npm:vue-demi"),
"npm": ("install", "--silent"),
"bun": ("install", "--silent"),
}
dev_args = {
"deno": ("run", "dev", "--"),
"npm": ("--silent", "run", "dev", "--"),
"bun": ("run", "dev", "--"),
}
install_cmd = [tool, *install_args[name]]
dev_cmd = [
tool,
*dev_args[name],
"--clearScreen=false",
f"--port={vite_port}",
]
if all_ifaces:
dev_cmd.append("--host")
if name == "bun":
stderr.write(BUN_BUG)
return install_cmd, dev_cmd, name
async def wait_for_backend(host: str, port: int):
"""Wait for the backend to be ready by polling the health endpoint."""
max_attempts = 50
url = f"http://{host}:{port}"
async with httpx.AsyncClient() as client:
for attempt in range(max_attempts):
try:
await client.get(url, timeout=1.0)
stderr.write("✓ Backend ready!\n")
return True
except httpx.RequestError:
if attempt == max_attempts - 1:
stderr.write("┃ ⚠️ Backend didn't start in time\n")
return False
await asyncio.sleep(0.1)
return False
async def _terminate_process(proc: asyncio.subprocess.Process, name: str) -> None:
"""Gracefully terminate a subprocess."""
if proc.returncode is not None:
return
try:
proc.terminate()
except ProcessLookupError:
return
try:
await asyncio.wait_for(proc.wait(), timeout=2)
except TimeoutError:
try:
proc.kill()
except ProcessLookupError:
return
await proc.wait()
DEFAULT_VITE_PORT = TEMPLATE_VITE_PORT
DEFAULT_DEV_PORT = TEMPLATE_DEV_PORT
HEALTH = TEMPLATE_HEALTH
async def run_devserver(
vite_port: int,
all_ifaces: bool,
backend_host: str,
backend_port: int,
listen: str,
backend: str,
extra_args: list[str] | None = None,
) -> None:
"""Run the development server with install, backend, and frontend."""
install_cmd, dev_cmd, tool_name = resolve_frontend_tools(vite_port, all_ifaces)
"""Start Vite and FastAPI dev servers with hot reload."""
reporoot = Path(__file__).parent.parent
front = reporoot / "frontend"
if not (front / "package.json").exists():
logger.warning("Frontend source not found at %s", front)
raise SystemExit(1)
# Tell the backend where the Vite dev server is
os.environ["FASTAPI_VUE_FRONTEND_URL"] = f"http://localhost:{vite_port}"
# Tell Vite where the backend is (for proxying /api requests)
os.environ["FASTAPI_VUE_BACKEND_URL"] = f"http://{backend_host}:{backend_port}"
viteurl, npm_install, vite = setup_vite(listen, DEFAULT_VITE_PORT)
backurl, MODULE_NAME = setup_cli("PROJECT_CLI", backend, DEFAULT_DEV_PORT)
backend_cmd = [
"uvicorn",
"MODULE_NAME.app:app",
"--host",
backend_host,
"--port",
str(backend_port),
"--reload",
]
# Tell everyone via environment (vite proxy and backend devmode use these)
os.environ["ENVPREFIX_VITE_URL"] = viteurl
os.environ["ENVPREFIX_BACKEND_URL"] = backurl
os.environ["ENVPREFIX_DEV"] = "1"
cwd = str(Path(__file__).parent.parent)
frontend_cwd = str(FRONTEND_PATH)
backend_proc: asyncio.subprocess.Process | None = None
install_proc: asyncio.subprocess.Process | None = None
frontend_proc: asyncio.subprocess.Process | None = None
try:
# Start install (concurrent with backend)
stderr.write(f">>> {tool_name} {' '.join(install_cmd[1:])}\n")
install_proc = await asyncio.create_subprocess_exec(
*install_cmd, cwd=frontend_cwd
)
await asyncio.sleep(0.1)
# Start backend (concurrent with install)
stderr.write(f">>> {' '.join(backend_cmd)}\n")
backend_proc = await asyncio.create_subprocess_exec(*backend_cmd, cwd=cwd)
# Wait for install to complete and backend to be ready
install_task = asyncio.create_task(install_proc.wait(), name="install")
backend_ready_task = asyncio.create_task(
wait_for_backend(backend_host, backend_port), name="backend_ready"
)
done, pending = await asyncio.wait(
{install_task, backend_ready_task},
return_when=asyncio.FIRST_COMPLETED,
)
for task in done:
if task.get_name() == "install":
if task.result() != 0:
stderr.write("┃ ⚠️ Install failed\n")
raise SystemExit(1)
elif task.get_name() == "backend_ready" and not task.result():
raise SystemExit(1)
if pending:
done2, _ = await asyncio.wait(pending)
for task in done2:
if task.get_name() == "install":
if task.result() != 0:
stderr.write("┃ ⚠️ Install failed\n")
raise SystemExit(1)
elif task.get_name() == "backend_ready" and not task.result():
raise SystemExit(1)
install_proc = None
# Start Vite dev server
stderr.write(f">>> {tool_name} {' '.join(dev_cmd[1:])}\n")
frontend_proc = await asyncio.create_subprocess_exec(*dev_cmd, cwd=frontend_cwd)
# Wait for either process to exit
done, pending = await asyncio.wait(
{
asyncio.create_task(backend_proc.wait(), name="backend"),
asyncio.create_task(frontend_proc.wait(), name="frontend"),
},
return_when=asyncio.FIRST_COMPLETED,
)
for t in done:
t.result()
for t in pending:
t.cancel()
except asyncio.CancelledError:
stderr.write("\n✓ Shutting down...\n")
finally:
if frontend_proc is not None:
await _terminate_process(frontend_proc, "frontend")
if install_proc is not None:
await _terminate_process(install_proc, "install")
if backend_proc is not None:
await _terminate_process(backend_proc, "backend")
async with ProcessGroup() as pg:
pg.create_task(check_ports_free(viteurl, backurl))
npm_i = await pg.spawn(*npm_install, cwd=front)
await pg.spawn(*MODULE_NAME, *(extra_args or []), vital=True)
await pg.wait(npm_i, ready(backurl, path=HEALTH))
await pg.spawn(*vite, cwd=front, vital=True)
def main():
def main() -> None:
"""Parse CLI arguments and run the devserver."""
tracerite.load()
parser = argparse.ArgumentParser(
description="Run Vite and FastAPI development servers",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=EPILOG,
epilog=HELP_EPILOG,
)
parser.add_argument(
"frontend",
nargs="?",
metavar="host:port",
help="Vite frontend endpoint (default: localhost:5173)",
"-l",
"--listen",
metavar="addr",
help=f"Vite (default: localhost:{DEFAULT_VITE_PORT})",
)
parser.add_argument(
"--backend",
metavar="host:port",
help="FastAPI backend endpoint (default: localhost:5180)",
metavar="addr",
help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})",
)
args = parser.parse_args()
args, extra_args = parser.parse_known_args()
try:
asyncio.run(run_devserver(args.listen, args.backend, extra_args))
except* KeyboardInterrupt:
pass # user stopped the devserver: normal exit
except* (subprocess.SubprocessError, RuntimeError):
raise SystemExit(1) from None # logged in devutil already; exit 1
# parse_endpoint returns list of dicts with host/port or uds keys
# Multiple entries means bind all interfaces (IPv4 + IPv6)
vite_endpoints = parse_endpoint(args.frontend, DEFAULT_VITE_PORT)
backend_endpoints = parse_endpoint(args.backend, DEFAULT_BACKEND_PORT)
# Vite doesn't support unix sockets
if "uds" in vite_endpoints[0]:
stderr.write("┃ ⚠️ Unix sockets not supported for frontend\n")
raise SystemExit(1)
if "uds" in backend_endpoints[0]:
stderr.write("┃ ⚠️ Unix sockets not supported for backend\n")
raise SystemExit(1)
HELP_EPILOG = """
Other options are forwarded to PROJECT_CLI [args]
vite_port = vite_endpoints[0]["port"]
all_ifaces = len(vite_endpoints) > 1
backend_host = backend_endpoints[0]["host"]
backend_port = backend_endpoints[0]["port"]
with contextlib.suppress(KeyboardInterrupt):
asyncio.run(run_devserver(vite_port, all_ifaces, backend_host, backend_port))
JS_RUNTIME environment variable can be used to select the JS runtime:
npm, deno, bun, or full path to the runtime executable (node maps to npm).
"""
if __name__ == "__main__":
@@ -1,34 +0,0 @@
"""Hatch build hook for building Vue frontend during package build."""
import subprocess
from pathlib import Path
from sys import stderr
from hatchling.builders.hooks.plugin.interface import BuildHookInterface # type: ignore
exec(Path(__file__).with_name("util.py").read_text("UTF-8")) # noqa: S102
def run(cmd, **kwargs):
"""Run a command and display it."""
display_cmd = [Path(cmd[0]).name, *cmd[1:]]
stderr.write(f"### {' '.join(display_cmd)}\n")
subprocess.run(cmd, check=True, **kwargs)
class CustomBuildHook(BuildHookInterface):
"""Build hook that compiles Vue frontend before packaging."""
def initialize(self, version, build_data):
super().initialize(version, build_data)
stderr.write(">>> Building the frontend\n")
install_cmd, build_cmd = find_build_tool() # noqa # type: ignore
try:
run(install_cmd, cwd="frontend")
stderr.write("\n")
run(build_cmd, cwd="frontend")
except Exception as e:
stderr.write(f"Error occurred while building frontend: {e}\n")
raise
+19
View File
@@ -0,0 +1,19 @@
# ruff: noqa: INP001
"""Hatch build hook for building Vue frontend during package build."""
import sys
from pathlib import Path
from hatchling.builders.hooks.plugin.interface import BuildHookInterface
sys.path.insert(0, str(Path(__file__).parent))
from buildutil import build
class CustomBuildHook(BuildHookInterface): # type: ignore[misc]
"""Hatch build hook that builds Vue frontend during package build."""
def initialize(self, version: str, build_data: dict) -> None: # type: ignore[override]
"""Build frontend before package is built."""
super().initialize(version, build_data)
build("frontend")
+231
View File
@@ -0,0 +1,231 @@
# ruff: noqa: INP001
"""Utilities used at build time and in devserver script. No dependencies."""
import logging
import os
import re
import shutil
import subprocess
from pathlib import Path
MIN_NODE_VERSION = 20
class _PrefixFormatter(logging.Formatter):
"""Formatter that adds prefix based on log level."""
def format(self, record: logging.LogRecord) -> str:
if record.levelno >= logging.WARNING:
return f"⚠️ {record.getMessage()}"
return record.getMessage()
_handler = logging.StreamHandler()
_handler.setFormatter(_PrefixFormatter())
logger = logging.getLogger("fastapi-vue")
logger.addHandler(_handler)
logger.setLevel(logging.INFO)
def _check_node_version(node_path: str) -> None:
"""Check if Node.js version is >= 20.
Raises RuntimeError if version is too old or cannot be determined.
"""
try:
result = subprocess.run( # noqa: S603
[node_path, "--version"],
capture_output=True,
text=True,
check=True,
)
version_str = result.stdout.strip()
# Parse version like "v20.10.0" or "v18.17.1"
match = re.match(r"v(\d+)", version_str)
if match:
major_version = int(match.group(1))
if major_version >= MIN_NODE_VERSION:
return
msg = f"Node.js {version_str} found, but v{MIN_NODE_VERSION}+ required"
raise RuntimeError(msg)
except (subprocess.CalledProcessError, FileNotFoundError, ValueError):
pass
msg = "Could not determine Node.js version"
raise RuntimeError(msg)
def _validate_npm_runtime(tool: str) -> bool:
"""Validate npm runtime by checking Node.js version. Returns True if valid."""
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path is None:
return False
try:
_check_node_version(node_path)
except RuntimeError:
return False
return True
def _find_runtime_from_env(options: list[str]) -> tuple[str, str] | None:
"""Find runtime specified by JS_RUNTIME environment variable."""
js_runtime_env = os.environ.get("JS_RUNTIME")
if not js_runtime_env:
return None
js_runtime = js_runtime_env
js_path = Path(js_runtime)
runtime_name = js_path.name
# Map node to npm
if runtime_name == "node":
runtime_name = "npm"
js_runtime = str(js_path.parent / "npm") if js_path.parent.name else "npm"
for option in options:
if option != runtime_name and not runtime_name.startswith(option):
continue
tool = shutil.which(js_runtime)
if tool is None:
msg = f"JS_RUNTIME={js_runtime_env}: {option} not found"
raise RuntimeError(msg)
if option == "npm":
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path is None:
msg = f"JS_RUNTIME={js_runtime_env}: node not found"
raise RuntimeError(msg)
_check_node_version(node_path)
return tool, option
msg = f"JS_RUNTIME={js_runtime_env} not recognized"
raise RuntimeError(msg)
def _auto_detect_runtime(options: list[str]) -> tuple[str, str]:
"""Auto-detect JavaScript runtime from available options."""
node_version_error: RuntimeError | None = None
for option in options:
tool = shutil.which(option)
if not tool:
continue
if option == "npm" and not _validate_npm_runtime(tool):
try:
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path:
_check_node_version(node_path)
except RuntimeError as e:
node_version_error = e
continue
return tool, option
if node_version_error:
raise node_version_error
msg = "Node.js (v20+), Deno or Bun is required but none was found"
raise RuntimeError(msg)
def find_js_runtime() -> tuple[str, str]:
"""Find a JavaScript runtime from JS_RUNTIME env or auto-detect.
Returns (tool_path, tool_name) where tool_name is "deno", "npm", or "bun".
Raises RuntimeError if no suitable runtime is found.
"""
options = ["npm", "deno", "bun"]
# Check for JS_RUNTIME environment variable
if result := _find_runtime_from_env(options):
return result
# Auto-detect
return _auto_detect_runtime(options)
def find_build_tool() -> tuple[list[str], list[str]]:
"""Find JavaScript runtime and construct install/build commands.
Returns (install_cmd, build_cmd) tuples of command lists.
Raises RuntimeError if no runtime is found.
"""
install = {
"deno": ("install", "--allow-scripts=npm:vue-demi"),
"npm": ("install",),
"bun": ("--bun", "install"),
}
# Run vite directly for deno to avoid npm-run-all2/run-p issues
build = {
"deno": ("run", "-A", "npm:vite", "build"),
"npm": ("run", "build"),
"bun": ("--bun", "run", "build"),
}
tool, name = find_js_runtime()
return [tool, *install[name]], [tool, *build[name]]
def find_dev_tool() -> list[str]:
"""Find JavaScript runtime and construct dev command.
Returns dev_cmd (without vite-specific args).
Raises RuntimeError if no runtime is found.
"""
dev_args = {
"deno": ("run", "-A", "npm:vite"),
"npm": ("--silent", "run", "dev", "--"),
"bun": ("run", "dev", "--"),
}
tool, name = find_js_runtime()
if name == "bun":
logger.warning(
"Bun has a WS proxy bug (github.com/oven-sh/bun/issues/9882). Consider npm.",
)
return [tool, *dev_args[name]]
def find_install_tool() -> list[str]:
"""Find JavaScript runtime and construct install command.
Returns install_cmd.
Raises RuntimeError if no runtime is found.
"""
install_args = {
"deno": ("install", "--quiet", "--allow-scripts=npm:vue-demi"),
"npm": ("install", "--silent"),
"bun": ("install", "--silent"),
}
tool, name = find_js_runtime()
return [tool, *install_args[name]]
def build(folder: str = "frontend") -> None:
"""Build the frontend in the specified folder.
Raises SystemExit(1) on failure.
"""
logger.info(">>> Building %s", folder)
try:
install_cmd, build_cmd = find_build_tool()
except RuntimeError as e:
logger.warning(e)
raise SystemExit(1) from None
def run(cmd: list[str]) -> None:
display_cmd = [Path(cmd[0]).stem, *cmd[1:]]
logger.info("### %s", " ".join(display_cmd))
subprocess.run(cmd, check=True, cwd=folder) # noqa: S603
try:
run(install_cmd)
logger.info("")
run(build_cmd)
except subprocess.CalledProcessError:
raise SystemExit(1) from None
+227
View File
@@ -0,0 +1,227 @@
# ruff: noqa: INP001
"""Utilities meant for devserver script, used only in source repository with dev deps."""
from __future__ import annotations
import asyncio
import sys
from asyncio.subprocess import Process
from contextlib import suppress
from pathlib import Path
from subprocess import CalledProcessError
from typing import TYPE_CHECKING, Any
from urllib.parse import urlsplit
from buildutil import find_dev_tool, find_install_tool, logger
from fastapi_vue.hostutil import parse_endpoint
if TYPE_CHECKING:
from collections.abc import Awaitable
class ProcessGroup(asyncio.TaskGroup):
"""TaskGroup with structured ownership of async subprocesses."""
def __init__(self, *, terminate_timeout: float = 10) -> None:
"""Set the grace period before terminate() escalates to kill()."""
super().__init__()
self._terminate_timeout = terminate_timeout
self._cmds: dict[Process, tuple[str, ...]] = {}
async def spawn(self, *cmd: str, cwd: str | None = None, vital: bool = False) -> Process:
"""Spawn and own a subprocess. If a vital process exits, the group cancels."""
async def run() -> None:
name = Path(cmd[0]).stem
logger.info(">>> %s", " ".join([name, *cmd[1:]]))
try:
proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd)
self._cmds[proc] = cmd
started.set_result(proc)
except Exception as e: # noqa: BLE001
started.set_exception(e)
return
try:
returncode = await proc.wait()
finally:
with suppress(ProcessLookupError):
proc.terminate()
try:
await asyncio.wait_for(proc.wait(), self._terminate_timeout)
except TimeoutError:
with suppress(ProcessLookupError):
proc.kill()
await proc.wait()
if vital:
logger.warning("Vital process %s exited", name)
raise CalledProcessError(returncode, cmd)
started = asyncio.get_running_loop().create_future()
self.create_task(run())
return await asyncio.shield(started)
async def wait(self, *waitables: Process | Awaitable) -> tuple[Any, ...]:
"""Wait concurrently and return results in argument order."""
async def task(w: Process | Awaitable) -> Any: # noqa: ANN401
if not isinstance(w, Process):
return await w
if retcode := await w.wait():
cmd = self._cmds[w]
logger.warning("Process %s exited with status %d", Path(cmd[0]).stem, retcode)
raise CalledProcessError(retcode, cmd)
return retcode
async with asyncio.TaskGroup() as group:
tasks = [group.create_task(task(w)) for w in waitables]
return tuple(task.result() for task in tasks)
async def http_get_server(url: str, timeout: float) -> str | None: # noqa: ASYNC109
"""GET url with plain asyncio streams, return the response Server header.
Returns an empty string when the server responds without a Server header,
and None when the server is unreachable or doesn't answer in time.
"""
parts = urlsplit(url)
host = parts.hostname or "localhost"
port = parts.port or (443 if parts.scheme == "https" else 80)
path = parts.path or "/"
if parts.query:
path += f"?{parts.query}"
try:
async with asyncio.timeout(timeout):
reader, writer = await asyncio.open_connection(host, port)
try:
writer.write(f"GET {path} HTTP/1.0\r\nHost: {host}\r\n\r\n".encode())
await writer.drain()
data = await reader.readuntil(b"\r\n\r\n")
finally:
writer.close()
except (OSError, EOFError, ValueError, TimeoutError):
return None
for line in data.decode(errors="replace").split("\r\n"):
if line.lower().startswith("server:"):
return line[7:].strip()
return ""
async def check_ports_free(*urls: str) -> None:
"""Verify URLs are not responding (ports are free).
Meant to run as a task inside a TaskGroup. Logs the conflict and raises
RuntimeError (handled like a failed process) if any URL responds.
"""
servers = await asyncio.gather(*(http_get_server(url, timeout=0.1) for url in urls))
for url, server in zip(urls, servers, strict=True):
if server is not None:
logger.error("Conflicting %s already running at %s", server or "server", url)
raise RuntimeError(url)
async def ready(url: str, path: str = "", max_attempts: int = 50) -> None:
"""Wait for the server to be ready by polling an endpoint.
Use empty path to disable the check and make this return immediately.
Logs, then raises RuntimeError if the server doesn't start in time.
"""
if not path:
return
for attempt in range(max_attempts):
if await http_get_server(f"{url}{path}", timeout=1.0) is not None:
logger.info("✓ Backend ready!")
return
if attempt == max_attempts - 1:
logger.error("Backend at %s didn't start in time", url)
raise RuntimeError(url)
await asyncio.sleep(0.1)
def setup_vite(
endpoint: str,
default_port: int = 5173,
) -> tuple[str, list[str], list[str]]:
"""Parse frontend endpoint and build commands.
Returns (url, install_cmd, dev_cmd).
Raises SystemExit(1) on invalid config.
"""
endpoints = parse_endpoint(endpoint, default_port)
if "uds" in endpoints[0]:
logger.warning("Unix sockets not supported with vite devserver")
raise SystemExit(1)
port = endpoints[0]["port"]
host = endpoints[0]["host"]
install_cmd = find_install_tool()
dev_cmd = find_dev_tool()
if host != "localhost":
dev_cmd.append("--host" if len(endpoints) > 1 else f"--host={host}")
dev_cmd.append(f"--port={port}")
return f"http://{host}:{port}", install_cmd, dev_cmd
def setup_fastapi(
endpoint: str,
module: str,
default_port: int = 8000,
) -> tuple[str, list[str]]:
"""Parse backend endpoint and build uvicorn command.
Returns (url, uvicorn_cmd).
Raises SystemExit(1) on invalid config.
"""
endpoints = parse_endpoint(endpoint, default_port)
if "uds" in endpoints[0]:
logger.warning("Unix sockets not supported with vite devserver")
raise SystemExit(1)
host = endpoints[0]["host"]
port = endpoints[0]["port"]
reload_dir = module.split(".", maxsplit=1)[0] # Don't reload on frontend changes
cmd = [
sys.executable,
"-m",
"uvicorn",
module,
f"--host={host}",
f"--port={port}",
"--reload",
f"--reload-dir={reload_dir}",
"--forwarded-allow-ips=*",
]
return f"http://{host}:{port}", cmd
def setup_cli(
cli: str,
endpoint: str,
default_port: int = 8000,
) -> tuple[str, list[str]]:
"""Parse backend endpoint and build CLI command.
Returns (url, cli_cmd).
Raises SystemExit(1) on invalid config.
"""
endpoints = parse_endpoint(endpoint, default_port)
if "uds" in endpoints[0]:
logger.warning("Unix sockets not supported with vite devserver")
raise SystemExit(1)
host = endpoints[0]["host"]
port = endpoints[0]["port"]
# Run the package as a module with the current interpreter, instead of
# relying on a PATH-installed CLI entry point.
cmd = [sys.executable, "-m", cli, f"--listen={host}:{port}"]
return f"http://{host}:{port}", cmd
-87
View File
@@ -1,87 +0,0 @@
"""Shared utilities for build and dev scripts."""
import os
import shutil
from pathlib import Path
from sys import stderr
def find_js_runtime() -> tuple[str, str] | None:
"""Find a JavaScript runtime from JS_RUNTIME env or auto-detect.
Returns (tool_path, tool_name) where tool_name is "deno", "npm", or "bun".
Returns None if no runtime is found.
"""
options = ["deno", "npm", "bun"]
# Check for JS_RUNTIME environment variable
if js_runtime_env := os.environ.get("JS_RUNTIME"):
js_runtime = js_runtime_env
js_path = Path(js_runtime)
runtime_name = js_path.name
# Map node to npm
if runtime_name == "node":
runtime_name = "npm"
js_runtime = str(js_path.parent / "npm") if js_path.parent.name else "npm"
for option in options:
if option == runtime_name or runtime_name.startswith(option):
tool = shutil.which(js_runtime)
if tool is None:
stderr.write(f"┃ ⚠️ JS_RUNTIME={js_runtime_env} not found\n")
return None
return tool, option
stderr.write(f"┃ ⚠️ JS_RUNTIME={js_runtime_env} not recognized\n")
return None
# Auto-detect
for option in options:
if tool := shutil.which(option):
return tool, option
return None
def find_build_tool():
"""Find JavaScript runtime and construct install/build commands.
Returns (install_cmd, build_cmd) tuples of command lists.
Raises RuntimeError if no runtime is found.
"""
install = {
"deno": ("install", "--allow-scripts=npm:vue-demi"),
"npm": ("install",),
"bun": ("--bun", "install"),
}
# Run vite directly for deno to avoid npm-run-all2/run-p issues
build = {
"deno": ("run", "-A", "npm:vite", "build"),
"npm": ("run", "build"),
"bun": ("--bun", "run", "build"),
}
result = find_js_runtime()
if result is None:
raise RuntimeError(
"Deno, npm or Bun is required for building but none was found"
)
tool, name = result
return [tool, *install[name]], [tool, *build[name]]
def find_dev_tool():
"""Find JavaScript runtime and construct dev command.
Returns (dev_cmd, tool_name) or (None, None) if not found.
"""
dev_args = {
"deno": ("run", "dev", "--"),
"npm": ("--silent", "run", "dev", "--"),
"bun": ("run", "dev", "--"),
}
result = find_js_runtime()
if result is None:
return None, None
tool, name = result
return [tool, *dev_args[name]], name