Compare commits

..
20 Commits
Author SHA1 Message Date
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
15 changed files with 1029 additions and 202 deletions
+47
View File
@@ -0,0 +1,47 @@
# fastapi-vue
Implements Single-Page-App serving at site root with FastAPI, that the standard StaticFiles module cannot handle. This also caches and zstd compresses the files for lightning-fast operation. This is primarily meant for use with Vue frontend, but technically can host any static files in a similar manner.
## Installation
Script [fastapi-vue-setup](https://git.zi.fi/LeoVasanko/fastapi-vue-setup) should normally be used to convert or create a project with connection between FastAPI and Vue. The target project will depend on this package to serve its static files.
```sh
uvx fastapi-vue-setup --help
```
Refer to instructions below for further configuration.
## Usage
```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi_vue import Frontend
frontend = Frontend(
Path(__file__).with_name("frontend-build"),
spa=True,
cached=["/assets/"],
)
@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, "/")
```
## Configuration
- `directory`: Path to static files directory
- `spa`: Enable SPA mode (serve index.html for unknown routes)
- `cached`: Path prefixes for immutable cache headers (browser won't check for changes)
- `favicon`: Path to serve at `/favicon.ico` (e.g., `"/logo.png"` will be served as `image/png`)
- `zstdlevel`: Compression level (default: 18)
+3
View File
@@ -0,0 +1,3 @@
from .staticfiles import Frontend
__all__ = ["Frontend"]
+67
View File
@@ -0,0 +1,67 @@
import contextlib
import ipaddress
from urllib.parse import urlparse
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).
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)}]
# Leading colon :port -> bind all interfaces (0.0.0.0 + ::)
if value.startswith(":") and value != ":":
port_part = value[1:]
if not port_part.isdigit():
raise SystemExit(f"Invalid port in '{value}'")
port = int(port_part)
return [{"host": "0.0.0.0", "port": port}, {"host": "::", "port": port}] # noqa: S104
# UNIX domain socket (unix:/path or just /path)
if value.startswith("/"):
return [{"uds": value}]
if value.startswith("unix:"):
uds_path = value[5:] or None
if uds_path is None:
raise SystemExit("unix: path must not be empty")
return [{"uds": uds_path}]
# Unbracketed IPv6 (cannot safely contain a port) -> detect by multiple colons
if value.count(":") > 1 and not value.startswith("["):
try:
ipaddress.IPv6Address(value)
except ValueError as e:
raise SystemExit(f"Invalid IPv6 address '{value}': {e}") from e
return [{"host": value, "port": default_port}]
# Use urllib.parse for everything else (host[:port], [ipv6][:port])
parsed = urlparse(f"//{value}") # // prefix lets urlparse treat it as netloc
host = parsed.hostname or "localhost"
port = parsed.port or default_port
# Validate IP literals (optional; hostname passes through)
with contextlib.suppress(ValueError):
ipaddress.ip_address(host)
return [{"host": host, "port": port}]
+78
View File
@@ -0,0 +1,78 @@
import asyncio
import logging
import os
from contextlib import suppress
import uvicorn
from uvicorn import Config, Server
from .hostutil import parse_endpoint
logger = logging.getLogger(__name__)
def run(
app: str,
*,
listen: str | list[str] | None = None,
default_port: int = 8000,
reload: bool = False,
workers: int | None = None,
**uvicorn_config,
):
"""Run uvicorn server(s) for the given app.
Args:
app: The ASGI application path (e.g., "myapp.main:app")
listen: Endpoint string(s) (see parse_endpoint for formats).
default_port: Port to use when not specified in listen args.
reload: Enable auto-reload (requires uvicorn.run, single endpoint only).
workers: Number of worker processes (requires uvicorn.run, single endpoint only).
**uvicorn_config: Additional uvicorn config options (overrides all other settings).
"""
if listen is None:
listen = [f"localhost:{default_port}"]
elif isinstance(listen, str):
listen = [listen]
endpoints: list[dict] = []
for ep in listen:
endpoints.extend(parse_endpoint(ep, default_port))
conf: dict[str, object] = {"app": app, "reload": 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):
if reload or workers:
serve_multiprocess(endpoints, **conf)
else:
asyncio.run(serve(endpoints, **conf))
async def serve(endpoints: list[dict], **kwargs) -> None:
"""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)),
)
await asyncio.gather(*(Server(Config(**kwargs, **ep)).serve() for ep in endpoints))
def serve_multiprocess(endpoints: list[dict], **kwargs) -> None:
"""Serve using uvicorn.run() for reload/workers support. Only first endpoint is used."""
if len(endpoints) > 1:
eps = [
ep["uds"] if "uds" in ep else f"{ep['host']}:{ep['port']}"
for ep in endpoints
]
logger.warning(
"Current mode supports only one endpoint. Listening: %s, skipped: %s",
eps[0],
" ".join(eps[1:]),
)
uvicorn.run(**kwargs, **endpoints[0])
+281
View File
@@ -0,0 +1,281 @@
"""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
logger = logging.getLogger("uvicorn.error") # Use FastAPI logging style
__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 _:
raise ValueError(f"Invalid cached value: {cached!r}")
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: May use wildcards of full path. E.g. /assets/logo*.png matches logo.hash.png created by Vite
zstdlevel: Zstd compression level (default: 18)
"""
def __init__(
self,
directory: Path | str,
*,
index: str = "index.html",
spa: bool = False,
catch_all: bool | None = None,
cached: str | list[str] | Assets = Assets(),
favicon: str | None = None,
zstdlevel: int = 18,
) -> None:
self.www: dict[str, tuple[bytes, bytes | None, dict]] = {}
self.base: Path = Path(directory)
self.index = index
self.spa = spa
self._catch_all = spa if catch_all is None else catch_all
self.cached_paths = Assets.parse(cached)
self.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):
"""Load static files from disk with compression."""
www: dict[str, tuple[bytes, bytes | None, dict]] = {}
if not self.base.exists():
raise ValueError(f"Frontend folder {self.base} not found (try uv build)")
paths = [PurePath()]
while paths:
current = self.base / paths.pop(0)
for p in current.iterdir():
rel = p.relative_to(self.base)
if p.is_dir():
paths.append(rel)
continue
# Read file
name = "/" + rel.as_posix()
mime = mimetypes.guess_type(name)[0] or "application/octet-stream"
name = name.removesuffix(self.index)
data = p.read_bytes()
etag = urlsafe_b64encode(blake3(data).digest(9)).decode()
if mime.startswith("text/"):
mime += "; charset=UTF-8"
mtime = p.stat().st_mtime
cached = any(name.startswith(prefix) for prefix in self.cached_paths)
headers = {
"etag": f'"{etag}"',
"last-modified": format_date_time(mtime),
"cache-control": (
"max-age=31536000, immutable" if cached else "no-cache"
),
"content-type": mime,
}
zstd = ZstdCompressor(self.zstdlevel).compress(data)
if len(zstd) >= len(data):
zstd = None
www[name] = data, zstd, headers
if self.favicon:
if m := fnmatch.filter(www, self.favicon):
data, zstd, headers = www[m[0]]
if "immutable" in headers.get("cache-control", ""):
headers = {**headers, "cache-control": "max-age=86400"}
www["/favicon.ico"] = data, zstd, headers
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):
"""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(
f"{self.base.name}: {len(self.www)} files in {1000 * duration:.1f} ms | "
f"zstd {len(compfiles)} files {1e-6 * raw:.2f}->{1e-6 * comp:.2f} MB ({ratio:.0f} %)"
)
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="/"):
"""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")(self.handle)
def _register_routes(self):
"""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):
"""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):
"""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=""):
"""Return error response directing to Vite server."""
return JSONResponse(
status_code=409,
content={
"detail": "Frontend assets served by Vite in debug mode. You are on backend, connect to frontend instead."
},
)
+23
View File
@@ -0,0 +1,23 @@
[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",
]
[project.urls]
Homepage = "https://git.zi.fi/LeoVasanko/fastapi-vue"
Repository = "https://github.com/LeoVasanko/fastapi-vue"
[build-system]
requires = ["hatchling", "hatch-vcs"]
build-backend = "hatchling.build"
[tool.hatch.version]
source = "vcs"
raw-options.root = ".."
+369 -75
View File
@@ -7,10 +7,12 @@ Usage:
Options: Options:
--module-name NAME Python module name (auto-detected from pyproject.toml) --module-name NAME Python module name (auto-detected from pyproject.toml)
--ports DEFAULT,VITE,DEV Port configuration (default: 3100,3100,3200)
--dry-run Show what would be done without making changes --dry-run Show what would be done without making changes
""" """
import argparse import argparse
import ast
import os import os
import re import re
import shutil import shutil
@@ -21,9 +23,50 @@ from textwrap import indent
import tomlkit import tomlkit
# Track Python files written/patched for ruff formatting
_python_files_to_format: list[Path] = []
# Template directory # Template directory
TEMPLATE_DIR = Path(__file__).parent / "template" TEMPLATE_DIR = Path(__file__).parent / "template"
def ruff_sort_imports(files: list[Path], dry_run: bool = False) -> None:
"""Run ruff to sort imports in the given Python files."""
if not files:
return
py_files = [str(f) for f in files if f.suffix == ".py" and f.exists()]
if not py_files:
return
if dry_run:
print(f"🔧 Would run ruff import sorting on {len(py_files)} files")
return
print("🔧 Ruff isort on modified files")
subprocess.run(
[sys.executable, "-m", "ruff", "check", "--select", "I", "--fix", *py_files],
stdout=subprocess.DEVNULL,
)
def uv_add_packages(
packages: list[str], *, cwd: Path, group: str | None = None
) -> None:
"""Add packages using uv."""
cmd = ["uv", "add", "-q", "-U"]
if group:
cmd.extend(["--group", group])
else:
cmd.append("--no-sync")
cmd.extend(packages)
result = subprocess.run(cmd, cwd=cwd, check=False)
if result.returncode != 0:
label = f" ({group})" if group else ""
print(f"⚠️ Failed to add{label} dependencies")
# Default ports: (default, vite, dev)
# If vite == dev, dev is incremented by 100
DEFAULT_PORTS = (3100, 3100, 3200)
# Marker comment indicating file can be auto-upgraded # Marker comment indicating file can be auto-upgraded
# Users should remove this line to prevent automatic updates # Users should remove this line to prevent automatic updates
UPGRADE_MARKER = "auto-upgrade@fastapi-vue-setup" UPGRADE_MARKER = "auto-upgrade@fastapi-vue-setup"
@@ -52,7 +95,16 @@ PYPROJECT_ADDITIONS = {
# Frontend instantiation block for patching existing apps # Frontend instantiation block for patching existing apps
FRONTEND_BLOCK = """ FRONTEND_BLOCK = """
# Vue Frontend static files # Vue Frontend static files
frontend = Frontend(Path(__file__).with_name("frontend-build"), cached=["/assets/"]) frontend = Frontend(Path(__file__).with_name("frontend-build"))
"""
# Lifespan block for patching apps that don't have one
LIFESPAN_BLOCK = """
@asynccontextmanager
async def lifespan(app: FastAPI):
\"\"\"Manage app startup and shutdown resources.\"\"\"
await frontend.load()
yield
""" """
# TypeScript health check script for Vue components # TypeScript health check script for Vue components
@@ -117,6 +169,98 @@ Next steps:
# ============================================================================= # =============================================================================
def parse_ports(ports_str: str | None) -> tuple[int, int, int]:
"""Parse comma-separated port string into (default, vite, dev) tuple.
If dev == vite, dev is incremented by 100 to avoid conflicts.
"""
if not ports_str:
return DEFAULT_PORTS
parts = ports_str.split(",")
if len(parts) == 1:
default = int(parts[0])
vite = default
dev = default + 100
elif len(parts) == 2:
default = int(parts[0])
vite = int(parts[1])
dev = vite + 100 if vite == default else default + 100
elif len(parts) == 3:
default = int(parts[0])
vite = int(parts[1])
dev = int(parts[2])
else:
raise ValueError(f"Invalid ports format: {ports_str}")
# Auto-adjust dev if it conflicts with vite
if dev == vite:
dev = vite + 100
return default, vite, dev
def find_import_insertion_line(source: str) -> int:
"""Find line number (1-based) for inserting imports, after shebang/docstring."""
try:
tree = ast.parse(source)
except SyntaxError:
return 2 if source.startswith("#!") else 1
# Find first import, or end of docstring if no imports
for node in tree.body:
if isinstance(node, (ast.Import, ast.ImportFrom)):
return node.lineno
if not (isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant)):
break # Non-import, non-docstring statement
# No imports found - insert after docstring or at line 1
if tree.body and isinstance(tree.body[0], ast.Expr):
return tree.body[0].end_lineno + 1
return 2 if source.startswith("#!") else 1
def extract_existing_ports(project_dir: Path) -> tuple[int, int, int] | None:
"""Extract existing port configuration from project files.
Returns (default, vite, dev) or None if not found.
"""
module_name = find_module_name(project_dir)
if not module_name:
return None
default_port = None
vite_port = None
dev_port = None
# Try to extract DEFAULT_PORT from __main__.py
main_file = project_dir / module_name / "__main__.py"
if main_file.exists():
content = main_file.read_text("UTF-8")
match = re.search(r"DEFAULT_PORT\s*=\s*(\d+)", content)
if match:
default_port = int(match.group(1))
# Try to extract ports from devserver.py
devserver_file = project_dir / "scripts" / "devserver.py"
if devserver_file.exists():
content = devserver_file.read_text("UTF-8")
match = re.search(r"DEFAULT_VITE_PORT\s*=\s*(\d+)", content)
if match:
vite_port = int(match.group(1))
match = re.search(r"DEFAULT_DEV_PORT\s*=\s*(\d+)", content)
if match:
dev_port = int(match.group(1))
# Return only if we found at least one port
if default_port is not None or vite_port is not None or dev_port is not None:
return (
default_port or DEFAULT_PORTS[0],
vite_port or DEFAULT_PORTS[1],
dev_port or DEFAULT_PORTS[2],
)
return None
def load_template(path: str) -> str: def load_template(path: str) -> str:
"""Load a template file from the template directory.""" """Load a template file from the template directory."""
return (TEMPLATE_DIR / path).read_text("UTF-8") return (TEMPLATE_DIR / path).read_text("UTF-8")
@@ -342,7 +486,7 @@ def patch_app_file(
) -> bool: ) -> bool:
"""Patch an existing app.py with frontend integration. """Patch an existing app.py with frontend integration.
Inserts import and Frontend instantiation after imports, route at bottom, Inserts imports at top (ruff will sort them), route at bottom,
and tries to patch lifespan with frontend.load(). and tries to patch lifespan with frontend.load().
Returns True if patched, False if already patched or failed. Returns True if patched, False if already patched or failed.
@@ -352,61 +496,143 @@ def patch_app_file(
return False return False
original_content = path.read_text("UTF-8") original_content = path.read_text("UTF-8")
marker = "from fastapi_vue import Frontend" content = original_content
if marker in original_content: # Check what's already patched
has_frontend = "from fastapi_vue import Frontend" in content
has_devmode = f"from {module_name}.__main__ import DEVMODE" in content
has_debug_arg = re.search(r"FastAPI\s*\([^)]*debug\s*=", content) is not None
has_lifespan = "await frontend.load()" in content
if has_frontend and has_devmode and has_debug_arg and has_lifespan:
print(f"✔️ {path} (already patched)") print(f"✔️ {path} (already patched)")
return False return False
# Find where to insert the import (after other imports)
lines = original_content.split("\n")
import_line = "from fastapi_vue import Frontend"
route_line = f'frontend.route({app_var}, "/")' route_line = f'frontend.route({app_var}, "/")'
# Find last import line and check if pathlib is imported # Add missing imports (using AST to find correct insertion point)
last_import_idx = 0 imports = []
has_pathlib = False if not has_frontend:
for i, line in enumerate(lines): imports.extend(["from pathlib import Path", "from fastapi_vue import Frontend"])
stripped = line.strip() if not has_devmode:
if stripped.startswith("import ") or stripped.startswith("from "): imports.append(f"from {module_name}.__main__ import DEVMODE")
last_import_idx = i if imports:
if "pathlib" in stripped or "from pathlib" in stripped: insert_line = find_import_insertion_line(content)
has_pathlib = True lines = content.splitlines(keepends=True)
elif stripped and not stripped.startswith("#") and last_import_idx > 0: # Convert to 0-based index
# Stop at first non-import, non-comment, non-empty line after imports insert_idx = insert_line - 1
break import_text = "\n".join(imports) + "\n"
if insert_idx >= len(lines):
# Append at end
content = content.rstrip("\n") + "\n" + import_text
else:
# Insert at the found position
content = (
"".join(lines[:insert_idx]) + import_text + "".join(lines[insert_idx:])
)
# Check if we found imports to insert after # Insert FRONTEND_BLOCK after last import (only if Frontend wasn't already there)
if last_import_idx == 0 and not lines[0].strip().startswith(("import ", "from ")): if not has_frontend:
print(f"⚠️ Skipping {path} (no imports found to patch)") lines = content.split("\n")
return False last_import_idx = 0
for i, line in enumerate(lines):
stripped = line.strip()
if stripped.startswith("import ") or stripped.startswith("from "):
last_import_idx = i
elif stripped and not stripped.startswith("#") and last_import_idx > 0:
break
lines.insert(last_import_idx + 1, FRONTEND_BLOCK)
content = "\n".join(lines)
# Insert imports after last import, then frontend instantiation # Append route at end (only if not already present)
if not has_pathlib: if route_line not in content:
lines.insert(last_import_idx + 1, "from pathlib import Path") lines = content.split("\n")
last_import_idx += 1 lines.append("")
lines.insert(last_import_idx + 1, import_line) lines.append(
lines.insert(last_import_idx + 2, FRONTEND_BLOCK) "# Serve the Vue frontend (needs to be last if SPA catch-all is used)"
)
lines.append(route_line)
content = "\n".join(lines)
# Append route at end # Try to patch FastAPI() call with debug=DEVMODE if no debug arg exists
lines.append("") if not has_debug_arg:
lines.append("# Serve the Vue frontend (needs to be last if SPA catch-all is used)") fastapi_pattern = r"(\w+\s*=\s*FastAPI\s*\()([^)]*)\)"
lines.append(route_line) for match in re.finditer(fastapi_pattern, content, re.DOTALL):
content = "\n".join(lines) args = match.group(2)
if "debug" not in args:
# Add debug=DEVMODE as last argument
if args.strip():
new_args = f"{args}, debug=DEVMODE"
else:
new_args = "debug=DEVMODE"
content = (
content[: match.start()]
+ match.group(1)
+ new_args
+ ")"
+ content[match.end() :]
)
break # Only patch first FastAPI() call
# Try to patch lifespan function - insert await frontend.load() before yield # Try to patch lifespan function - insert await frontend.load() before yield
lifespan_patched = False lifespan_patched = "await frontend.load()" in content
# Look for yield inside an async def lifespan function # Look for yield inside an async def lifespan function
# Find the yield statement and insert before it # Find the yield statement and insert before it
yield_pattern = r"^([ \t]+)(yield\b)" if not lifespan_patched:
yield_match = re.search(yield_pattern, content, re.MULTILINE) yield_pattern = r"^([ \t]+)(yield\b)"
if yield_match: yield_match = re.search(yield_pattern, content, re.MULTILINE)
indent = yield_match.group(1) if yield_match:
insert_pos = yield_match.start() ws = yield_match.group(1)
load_code = f"{indent}await frontend.load()\n" insert_pos = yield_match.start()
content = content[:insert_pos] + load_code + content[insert_pos:] load_code = f"{ws}await frontend.load()\n"
content = content[:insert_pos] + load_code + content[insert_pos:]
lifespan_patched = True
# No lifespan at all: create one and wire it into FastAPI()
if not lifespan_patched and f"@{app_var}.on_event" not in content:
# Add contextlib import
if "from contextlib import asynccontextmanager" not in content:
insert_line = find_import_insertion_line(content)
lines = content.splitlines(keepends=True)
insert_idx = insert_line - 1
import_text = "from contextlib import asynccontextmanager\n"
if insert_idx >= len(lines):
content = content.rstrip("\n") + "\n" + import_text
else:
content = (
"".join(lines[:insert_idx])
+ import_text
+ "".join(lines[insert_idx:])
)
# Insert lifespan block before the FastAPI() call
fastapi_line_pattern = r"^(\w+\s*=\s*FastAPI\s*\()"
fastapi_match = re.search(fastapi_line_pattern, content, re.MULTILINE)
if fastapi_match:
content = (
content[: fastapi_match.start()]
+ LIFESPAN_BLOCK.lstrip("\n")
+ "\n"
+ content[fastapi_match.start() :]
)
# Add lifespan=lifespan to FastAPI() call
fastapi_pattern = r"(\w+\s*=\s*FastAPI\s*\()([^)]*)\)"
fastapi_match = re.search(fastapi_pattern, content, re.DOTALL)
if fastapi_match and "lifespan" not in fastapi_match.group(2):
args = fastapi_match.group(2)
if args.strip():
new_args = f"{args}, lifespan=lifespan"
else:
new_args = "lifespan=lifespan"
content = (
content[: fastapi_match.start()]
+ fastapi_match.group(1)
+ new_args
+ ")"
+ content[fastapi_match.end() :]
)
lifespan_patched = True lifespan_patched = True
# Check if content actually changed # Check if content actually changed
@@ -415,10 +641,11 @@ def patch_app_file(
return False return False
if dry_run: if dry_run:
print(f"[DRY RUN] Would patch {path}") print(f" Would patch {path}")
return True return True
path.write_text(content, "UTF-8", newline="\n") path.write_text(content, "UTF-8", newline="\n")
_python_files_to_format.append(path)
print(f"✅ Patched {path}") print(f"✅ Patched {path}")
if not lifespan_patched: if not lifespan_patched:
@@ -501,11 +728,11 @@ def patch_vite_config(
# Check if content actually changed # Check if content actually changed
if content == original_content: if content == original_content:
print(f" Skipping {path} (no changes needed)") print(f" Skipping {path} (no changes needed)")
return False return False
if dry_run: if dry_run:
print(f"[DRY RUN] Would patch {path}") print(f" Would patch {path}")
return True return True
path.write_text(content, "UTF-8", newline="\n") path.write_text(content, "UTF-8", newline="\n")
@@ -539,7 +766,7 @@ def patch_frontend_health_check(frontend_dir: Path, dry_run: bool = False) -> bo
target_file = app_vue target_file = app_vue
if target_file is None: if target_file is None:
print(" No Vue file found to patch, skipping frontend health check") print(" No Vue file found to patch, skipping frontend health check")
return False return False
original_content = target_file.read_text("UTF-8") original_content = target_file.read_text("UTF-8")
@@ -584,7 +811,7 @@ def patch_frontend_health_check(frontend_dir: Path, dry_run: bool = False) -> bo
else: else:
# Minimal App.vue - only patch if it contains the default welcome message # Minimal App.vue - only patch if it contains the default welcome message
if "<h1>You did it!</h1>" not in content: if "<h1>You did it!</h1>" not in content:
print(f" Skipping {target_file} (not a default Vue template)") print(f" Skipping {target_file} (not a default Vue template)")
return False return False
# Insert before the </p> tag # Insert before the </p> tag
template_end = content.find("</template>") template_end = content.find("</template>")
@@ -601,11 +828,11 @@ def patch_frontend_health_check(frontend_dir: Path, dry_run: bool = False) -> bo
# Check if content actually changed # Check if content actually changed
if content == original_content: if content == original_content:
print(f" Skipping {target_file} (no changes needed)") print(f" Skipping {target_file} (no changes needed)")
return False return False
if dry_run: if dry_run:
print(f"[DRY RUN] Would patch {target_file}") print(f" Would patch {target_file}")
return True return True
target_file.write_text(content, "UTF-8", newline="\n") target_file.write_text(content, "UTF-8", newline="\n")
@@ -613,6 +840,42 @@ def patch_frontend_health_check(frontend_dir: Path, dry_run: bool = False) -> bo
return True return True
# SHA-256 of old vite-plugin-fastapi.js (before auto-upgrade marker was added)
# with module name replaced by MODULE_NAME in the outDir path
_OLD_VITE_PLUGIN_SHA256 = (
"93713e879c15a25c750a70ce1de684adeaf11b0c723c38da56e5e7ba207f6632"
)
def _upgrade_old_vite_plugin(
path: Path, module_name: str, dry_run: bool = False
) -> None:
"""Remove old vite-plugin-fastapi.js that lacks auto-upgrade marker.
Old versions didn't have the upgrade marker, so write_file skips them as
'customized by user'. We recognize the old version by normalizing the module
name in outDir and comparing the SHA-256 hash.
"""
if not path.exists():
return
content = path.read_text("UTF-8")
if UPGRADE_MARKER in content:
return # Already new format, write_file handles it
import hashlib
normalized = content.replace(
f"../{module_name}/frontend-build", "../MODULE_NAME/frontend-build"
)
digest = hashlib.sha256(normalized.encode()).hexdigest()
if digest != _OLD_VITE_PLUGIN_SHA256:
return # Modified by user, don't touch
if dry_run:
print(f"🔄 Would upgrade old {path}")
return
path.unlink()
print(f"🔄 Removing old {path} (will be replaced)")
# Track .new.py files written during setup (for merge notification) # Track .new.py files written during setup (for merge notification)
_new_files_written: list[tuple[Path, Path]] = [] _new_files_written: list[tuple[Path, Path]] = []
@@ -635,7 +898,7 @@ def write_file(
""" """
exists = path.exists() exists = path.exists()
if exists and not overwrite: if exists and not overwrite:
print(f" Skipping {path} (exists)") print(f" Skipping {path} (exists)")
return False return False
# Check if content is the same # Check if content is the same
@@ -657,13 +920,15 @@ def write_file(
if dry_run: if dry_run:
action = "overwrite" if exists else "create" action = "overwrite" if exists else "create"
print(f"[DRY RUN] Would {action} {path}") print(f" Would {action} {path}")
return True return True
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, "UTF-8", newline="\n") path.write_text(content, "UTF-8", newline="\n")
if executable and sys.platform != "win32": if executable and sys.platform != "win32":
path.chmod(path.stat().st_mode | 0o111) path.chmod(path.stat().st_mode | 0o111)
if path.suffix == ".py":
_python_files_to_format.append(path)
action = "Updated" if exists else "Created" action = "Updated" if exists else "Created"
print(f"{action} {path}") print(f"{action} {path}")
return True return True
@@ -684,7 +949,7 @@ def _write_fallback_file(
return False return False
if dry_run: if dry_run:
print(f"[DRY RUN] Would create {fallback_path} (original customized by user)") print(f" Would create {fallback_path} (original customized by user)")
_new_files_written.append((fallback_path, original_path)) _new_files_written.append((fallback_path, original_path))
return True return True
@@ -692,6 +957,8 @@ def _write_fallback_file(
fallback_path.write_text(content, "UTF-8", newline="\n") fallback_path.write_text(content, "UTF-8", newline="\n")
if executable and sys.platform != "win32": if executable and sys.platform != "win32":
fallback_path.chmod(fallback_path.stat().st_mode | 0o111) fallback_path.chmod(fallback_path.stat().st_mode | 0o111)
if fallback_path.suffix == ".py":
_python_files_to_format.append(fallback_path)
print(f"✅ Created {fallback_path} (original customized by user)") print(f"✅ Created {fallback_path} (original customized by user)")
_new_files_written.append((fallback_path, original_path)) _new_files_written.append((fallback_path, original_path))
return True return True
@@ -825,7 +1092,7 @@ def ensure_python_project(project_dir: Path, dry_run: bool = False) -> bool:
return True return True
if dry_run: if dry_run:
print(f"[DRY RUN] Would run: uv init {project_dir}") print(f"📦 Would run: uv init {project_dir}")
return True return True
print("📦 No pyproject.toml found, initializing Python project...") print("📦 No pyproject.toml found, initializing Python project...")
@@ -869,7 +1136,7 @@ def ensure_frontend(project_dir: Path, dry_run: bool = False) -> bool:
create_cmd = create_vue_commands[js_name] create_cmd = create_vue_commands[js_name]
if dry_run: if dry_run:
print(f"[DRY RUN] Would run: {' '.join(create_cmd)}") print(f"🎨 Would run: {' '.join(create_cmd)}")
return True return True
print("🎨 No frontend/ found, creating Vue project...") print("🎨 No frontend/ found, creating Vue project...")
@@ -913,7 +1180,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
# Create project directory if it doesn't exist # Create project directory if it doesn't exist
if not project_dir.exists(): if not project_dir.exists():
if dry_run: if dry_run:
print(f"[DRY RUN] Would create directory: {project_dir}") print(f" Would create directory: {project_dir}")
else: else:
project_dir.mkdir(parents=True) project_dir.mkdir(parents=True)
print(f"✅ Created {project_dir}") print(f"✅ Created {project_dir}")
@@ -936,17 +1203,40 @@ def cmd_setup(args: argparse.Namespace) -> int:
if not module_name: if not module_name:
# Derive from directory name # Derive from directory name
module_name = project_dir.name.replace("-", "_") module_name = project_dir.name.replace("-", "_")
print(f"📦 Using module name from directory: {module_name}") print(f"📦 Module: {module_name} (from directory name)")
else:
print(f"📦 Module: {module_name}")
# Determine port configuration
# Priority: --ports argument > existing project values > defaults
if args.ports:
default_port, vite_port, dev_port = parse_ports(args.ports)
ports_note = "(--ports)"
else:
existing_ports = extract_existing_ports(project_dir)
if existing_ports:
default_port, vite_port, dev_port = existing_ports
ports_note = "(kept for upgrade)"
else:
default_port, vite_port, dev_port = DEFAULT_PORTS
ports_note = "(--ports to override)"
print(
f"📡 Ports: default={default_port}, vite={vite_port}, dev={dev_port} {ports_note}"
)
# Title for templates # Title for templates
project_title = module_name.replace("_", " ").title() project_title = module_name.replace("_", " ").title()
print(f"📦 Module: {module_name}")
# Template variables # Template variables
tpl_vars = { tpl_vars = {
"MODULE_NAME": module_name, "MODULE_NAME": module_name,
"PROJECT_TITLE": project_title, "PROJECT_TITLE": project_title,
"TEMPLATE_DEFAULT_PORT": str(default_port),
"TEMPLATE_VITE_PORT": str(vite_port),
"TEMPLATE_DEV_PORT": str(dev_port),
"ENVPREFIX": module_name.upper(),
"PROJECT_CLI": module_name,
} }
module_dir = project_dir / module_name module_dir = project_dir / module_name
@@ -984,7 +1274,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
obsolete_util = fastapi_vue_scripts / "util.py" obsolete_util = fastapi_vue_scripts / "util.py"
if obsolete_util.exists(): if obsolete_util.exists():
if dry_run: if dry_run:
print(f"[DRY RUN] Would remove obsolete {obsolete_util}") print(f"🗑️ Would remove obsolete {obsolete_util}")
else: else:
obsolete_util.unlink() obsolete_util.unlink()
print(f"🗑️ Removed obsolete {obsolete_util}") print(f"🗑️ Removed obsolete {obsolete_util}")
@@ -1063,13 +1353,17 @@ def cmd_setup(args: argparse.Namespace) -> int:
overwrite=False, overwrite=False,
dry_run=dry_run, dry_run=dry_run,
) )
# else: no file but has existing entrypoint - don't create (user has custom CLI setup) else:
# No file but has existing entrypoint - don't create (user has custom CLI setup)
print(f"️ Skipping __main__.py (package already has CLI: {existing_cli})")
# === Update vite.config.js/ts === # === Update vite.config.js/ts ===
frontend_dir = project_dir / "frontend" frontend_dir = project_dir / "frontend"
if frontend_dir.exists(): if frontend_dir.exists():
# Install the vite plugin file (always update) # Install the vite plugin file (always update)
plugin_file = frontend_dir / "vite-plugin-fastapi.js" plugin_file = frontend_dir / "vite-plugin-fastapi.js"
# Upgrade old plugin versions that lack the auto-upgrade marker
_upgrade_old_vite_plugin(plugin_file, module_name, dry_run)
template = load_template("frontend/vite-plugin-fastapi.js") template = load_template("frontend/vite-plugin-fastapi.js")
content = render_template(template, **tpl_vars) content = render_template(template, **tpl_vars)
write_file(plugin_file, content, overwrite=True, dry_run=dry_run) write_file(plugin_file, content, overwrite=True, dry_run=dry_run)
@@ -1109,7 +1403,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
if new_content == old_content: if new_content == old_content:
print(f"✔️ {pyproject_path} (already up to date)") print(f"✔️ {pyproject_path} (already up to date)")
elif dry_run: elif dry_run:
print(f"[DRY RUN] Would update {pyproject_path}") print(f" Would update {pyproject_path}")
else: else:
pyproject_path.write_text(new_content, "UTF-8", newline="\n") pyproject_path.write_text(new_content, "UTF-8", newline="\n")
print(f"✅ Updated {pyproject_path}") print(f"✅ Updated {pyproject_path}")
@@ -1122,7 +1416,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
if b"frontend-build" in gitignore_content: if b"frontend-build" in gitignore_content:
print("✔️ .gitignore (frontend-build already ignored)") print("✔️ .gitignore (frontend-build already ignored)")
elif dry_run: elif dry_run:
print(f"[DRY RUN] Would add {gitignore_entry} to .gitignore") print(f" Would add {gitignore_entry} to .gitignore")
else: else:
nl = b"\r\n" if b"\r\n" in gitignore_content else b"\n" nl = b"\r\n" if b"\r\n" in gitignore_content else b"\n"
suffix = b"" if gitignore_content.endswith(nl) else nl suffix = b"" if gitignore_content.endswith(nl) else nl
@@ -1131,25 +1425,19 @@ def cmd_setup(args: argparse.Namespace) -> int:
) )
print(f"✅ Added {gitignore_entry} to .gitignore") print(f"✅ Added {gitignore_entry} to .gitignore")
elif dry_run: elif dry_run:
print(f"[DRY RUN] Would create .gitignore with {gitignore_entry}") print(f" Would create .gitignore with {gitignore_entry}")
else: else:
gitignore_path.write_text(f"{gitignore_entry}\n", "UTF-8", newline="\n") gitignore_path.write_text(f"{gitignore_entry}\n", "UTF-8", newline="\n")
print("✅ Created .gitignore") print("✅ Created .gitignore")
# === Add dependencies using uv === # === Add dependencies using uv ===
uv_add_main = ["uv", "add", "-q", "-U", "--no-sync", "fastapi[standard]", "fastapi-vue"] ruff_sort_imports(_python_files_to_format, dry_run=dry_run)
uv_add_dev = ["uv", "add", "-q", "-U", "--group", "dev", "httpx"]
if dry_run: if dry_run:
print(f"[DRY RUN] Would run: {' '.join(uv_add_main)}") print("📦 Would add: fastapi[standard], fastapi-vue, httpx (dev only)")
print(f"[DRY RUN] Would run: {' '.join(uv_add_dev)}")
else: else:
print("📦 Adding dependencies...") print("📦 Dependencies")
result = subprocess.run(uv_add_main, cwd=project_dir, check=False) uv_add_packages(["fastapi[standard]", "fastapi-vue"], cwd=project_dir)
if result.returncode != 0: uv_add_packages(["httpx"], cwd=project_dir, group="dev")
print("⚠️ Failed to add main dependencies")
result = subprocess.run(uv_add_dev, cwd=project_dir, check=False)
if result.returncode != 0:
print("⚠️ Failed to add dev dependencies")
print() print()
print("=" * 60) print("=" * 60)
@@ -1215,6 +1503,7 @@ Examples:
fastapi-vue-setup my-new-project Create a new project from scratch fastapi-vue-setup my-new-project Create a new project from scratch
fastapi-vue-setup . Set up integration in current directory fastapi-vue-setup . Set up integration in current directory
fastapi-vue-setup . --dry-run Preview what would be done fastapi-vue-setup . --dry-run Preview what would be done
fastapi-vue-setup . --ports 8000,5173,8080 Custom ports (default,vite,dev)
""", """,
) )
parser.add_argument( parser.add_argument(
@@ -1224,6 +1513,11 @@ Examples:
help="Project directory (use . for current directory)", help="Project directory (use . for current directory)",
) )
parser.add_argument("--module-name", help="Python module name (auto-detected)") parser.add_argument("--module-name", help="Python module name (auto-detected)")
parser.add_argument(
"--ports",
metavar="DEFAULT,VITE,DEV",
help="Port configuration as comma-separated values (default: 3100,3100,3200)",
)
parser.add_argument( parser.add_argument(
"--dry-run", action="store_true", help="Show what would be done" "--dry-run", action="store_true", help="Show what would be done"
) )
+5 -1
View File
@@ -9,6 +9,7 @@ description = "Tool to create or patch FastAPI+Vue projects with integrated buil
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.11"
dependencies = [ dependencies = [
"ruff>=0.14.13",
"tomlkit>=0.12.0", "tomlkit>=0.12.0",
] ]
@@ -26,4 +27,7 @@ source = "vcs"
include = ["fastapi_vue_setup.py", "template/**/*", "_version.py"] include = ["fastapi_vue_setup.py", "template/**/*", "_version.py"]
[dependency-groups] [dependency-groups]
dev = ["ruff"] dev = ["ruff", "fastapi-vue"]
[tool.uv.sources]
fastapi-vue = { path = "fastapi-vue", editable = true }
+35
View File
@@ -0,0 +1,35 @@
#!/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():
# 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,
)
# Build fastapi-vue-setup (root)
subprocess.run(
["uv", "build", "--out-dir", str(DIST)],
cwd=ROOT,
check=True,
)
if __name__ == "__main__":
main()
+14 -29
View File
@@ -1,44 +1,29 @@
# auto-upgrade@fastapi-vue-setup - remove this if you modify this file # auto-upgrade@fastapi-vue-setup - remove this if you modify this file
import argparse import argparse
import asyncio
import os import os
from fastapi_vue.hostutil import parse_endpoint from fastapi_vue import server
from uvicorn import Config, Server
DEFAULT_PORT = 5080 DEFAULT_PORT = TEMPLATE_DEFAULT_PORT
DEVMODE = bool(os.getenv("ENVPREFIX_FRONTEND_URL"))
def run_server(endpoints: list[dict], *, proxy=""):
conf: dict[str, object] = {"app": "MODULE_NAME.APP_MODULE:APP_VAR"}
if proxy:
conf["proxy_headers"] = True
conf["forwarded_allow_ips"] = proxy
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())
def main(): def main():
parser = argparse.ArgumentParser(description="Run the MODULE_NAME server.") parser = argparse.ArgumentParser(description="Run the MODULE_NAME server.")
parser.add_argument( parser.add_argument(
"endpoint", "-l",
nargs="?", "--listen",
help=( action="append",
f"Endpoint (default: localhost:{DEFAULT_PORT}). " help=(f"Endpoint (default: localhost:{DEFAULT_PORT})."),
"Forms: host:port | :port | [ipv6]:port | ip | host | unix:/path.sock"
),
) )
args = parser.parse_args() args = parser.parse_args()
proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1") dev = {"reload": True, "reload_dirs": ["MODULE_NAME"]}
try: server.run(
run_server(parse_endpoint(args.endpoint, DEFAULT_PORT), proxy=proxy) "MODULE_NAME.APP_MODULE:APP_VAR",
except KeyboardInterrupt: listen=args.listen,
pass default_port=DEFAULT_PORT,
**(dev if DEVMODE else {}),
)
if __name__ == "__main__": if __name__ == "__main__":
+3 -2
View File
@@ -3,9 +3,10 @@ from pathlib import Path
from fastapi import FastAPI from fastapi import FastAPI
from fastapi_vue import Frontend from fastapi_vue import Frontend
from MODULE_NAME.__main__ import DEVMODE
# Vue Frontend static files # Vue Frontend static files
frontend = Frontend(Path(__file__).with_name("frontend-build"), cached=["/assets/"]) frontend = Frontend(Path(__file__).with_name("frontend-build"))
@asynccontextmanager @asynccontextmanager
@@ -15,7 +16,7 @@ async def lifespan(app: FastAPI):
yield yield
app = FastAPI(title="PROJECT_TITLE", lifespan=lifespan) app = FastAPI(title="PROJECT_TITLE", debug=DEVMODE, lifespan=lifespan)
# Add API routes here... # Add API routes here...
+6 -5
View File
@@ -1,17 +1,18 @@
/** /**
* FastAPI-Vue Vite Plugin * FastAPI-Vue Vite Plugin
* auto-upgrade@fastapi-vue-setup -- remove this if you edit the plugin
* *
* Configures Vite for FastAPI backend integration: * Configures Vite for FastAPI backend integration:
* - Proxies /api/* requests to the FastAPI backend * - Proxies /api/* requests to the FastAPI backend
* - Builds to the Python module's frontend-build directory * - Builds to the Python module's frontend-build directory
* *
* Environment variables (with defaults): * Options:
* FASTAPI_VUE_BACKEND_URL=http://localhost:5180 - Backend API URL for proxying * 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"] } = {}) { export default function fastapiVue({ paths = ["/api"] } = {}) {
const backendUrl = process.env.ENVPREFIX_BACKEND_URL || "http://localhost:TEMPLATE_DEFAULT_PORT"
// Build proxy configuration for each path // Build proxy configuration for each path
const proxy = {} const proxy = {}
for (const path of paths) { for (const path of paths) {
@@ -23,7 +24,7 @@ export default function fastapiVue({ paths = ["/api"] } = {}) {
} }
return { return {
name: "fastapi-vite", name: "vite-plugin-fastapi-MODULE_NAME",
config: () => ({ config: () => ({
server: { proxy }, server: { proxy },
build: { build: {
+21 -27
View File
@@ -4,9 +4,9 @@
import argparse import argparse
import asyncio import asyncio
import contextlib
import os import os
import sys import sys
from contextlib import suppress
from pathlib import Path from pathlib import Path
# Import util.py from scripts/fastapi-vue (not a package, so we adjust sys.path) # Import util.py from scripts/fastapi-vue (not a package, so we adjust sys.path)
@@ -16,40 +16,35 @@ from devutil import ( # type: ignore
check_ports_free, check_ports_free,
logger, logger,
ready, ready,
setup_fastapi, setup_cli,
setup_vite, setup_vite,
) )
DEFAULT_VITE_PORT = TEMPLATE_VITE_PORT
DEFAULT_DEV_PORT = TEMPLATE_DEV_PORT
async def run_devserver(frontend: str, backend: str) -> None:
async def run_devserver(
frontend: str, backend: str, extra_args: list[str] | None = None
) -> None:
reporoot = Path(__file__).parent.parent reporoot = Path(__file__).parent.parent
front = reporoot / "frontend" front = reporoot / "frontend"
if not (front / "package.json").exists(): if not (front / "package.json").exists():
logger.warning("Frontend source not found at %s", front) logger.warning("Frontend source not found at %s", front)
raise SystemExit(1) raise SystemExit(1)
viteurl, npm_install, vite = setup_vite(frontend) viteurl, npm_install, vite = setup_vite(frontend, DEFAULT_VITE_PORT)
backurl, fastapi = setup_fastapi(backend, "MODULE_NAME.APP_MODULE:APP_VAR") backurl, MODULE_NAME = setup_cli("PROJECT_CLI", backend, DEFAULT_DEV_PORT)
# Tell the everyone where the frontend and backend are (vite proxy, etc) # Tell the everyone where the frontend and backend are (vite proxy, etc)
os.environ["FASTAPI_VUE_FRONTEND_URL"] = viteurl os.environ["ENVPREFIX_FRONTEND_URL"] = viteurl
os.environ["FASTAPI_VUE_BACKEND_URL"] = backurl os.environ["ENVPREFIX_BACKEND_URL"] = backurl
async with ProcessGroup() as pg: async with ProcessGroup() as pg:
npm_i = await pg.spawn(*npm_install, cwd=front) npm_i = await pg.spawn(*npm_install, cwd=front)
await check_ports_free(viteurl, backurl) await check_ports_free(viteurl, backurl)
await pg.spawn( await pg.spawn(*MODULE_NAME, *(extra_args or []))
*fastapi,
"--reload",
"--reload-dir=MODULE_NAME", # Don't reload on frontend changes
"--forwarded-allow-ips=*",
cwd=reporoot,
)
# Wait for both install and backend to be ready
await pg.wait(npm_i, ready(backurl, path="/api/health?from=devserver.py")) await pg.wait(npm_i, ready(backurl, path="/api/health?from=devserver.py"))
# Start Vite dev server (ProcessGroup waits for any exit, then terminates others)
await pg.spawn(*vite, cwd=front) await pg.spawn(*vite, cwd=front)
@@ -63,24 +58,23 @@ def main():
"frontend", "frontend",
nargs="?", nargs="?",
metavar="host:port", metavar="host:port",
help="Vite frontend endpoint (default: localhost:5173)", help=f"Vite frontend endpoint (default: localhost:{DEFAULT_VITE_PORT})",
) )
parser.add_argument( parser.add_argument(
"--backend", "--backend",
metavar="host:port", metavar="host:port",
help="FastAPI backend endpoint (default: localhost:5180)", help=f"FastAPI backend endpoint (default: localhost:{DEFAULT_DEV_PORT})",
) )
args = parser.parse_args() args, extra_args = parser.parse_known_args()
with contextlib.suppress(KeyboardInterrupt): with suppress(KeyboardInterrupt):
asyncio.run(run_devserver(args.frontend, args.backend)) asyncio.run(run_devserver(args.frontend, args.backend, extra_args))
HELP_EPILOG = """ HELP_EPILOG = """
scripts/devserver.py # Default ports on localhost scripts/devserver.py [args to PROJECT_CLI]
scripts/devserver.py 3000 # Vite on localhost:3000
scripts/devserver.py :3000 --backend 8000 # *:3000, localhost:8000
JS_RUNTIME environment variable can be used to select the JS runtime 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).
""" """
+2 -2
View File
@@ -134,7 +134,7 @@ def find_dev_tool() -> list[str]:
Raises RuntimeError if no runtime is found. Raises RuntimeError if no runtime is found.
""" """
dev_args = { dev_args = {
"deno": ("run", "dev", "--"), "deno": ("run", "-A", "npm:vite"),
"npm": ("--silent", "run", "dev", "--"), "npm": ("--silent", "run", "dev", "--"),
"bun": ("run", "dev", "--"), "bun": ("run", "dev", "--"),
} }
@@ -179,7 +179,7 @@ def build(folder: str = "frontend") -> None:
raise SystemExit(1) raise SystemExit(1)
def run(cmd): def run(cmd):
display_cmd = [Path(cmd[0]).name, *cmd[1:]] display_cmd = [Path(cmd[0]).stem, *cmd[1:]]
logger.info("### %s", " ".join(display_cmd)) logger.info("### %s", " ".join(display_cmd))
subprocess.run(cmd, check=True, cwd=folder) subprocess.run(cmd, check=True, cwd=folder)
+75 -61
View File
@@ -2,7 +2,9 @@
import asyncio import asyncio
import subprocess import subprocess
import sys
from collections.abc import Coroutine from collections.abc import Coroutine
from contextlib import suppress
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -10,9 +12,6 @@ import httpx
from buildutil import find_dev_tool, find_install_tool, logger from buildutil import find_dev_tool, find_install_tool, logger
from fastapi_vue.hostutil import parse_endpoint from fastapi_vue.hostutil import parse_endpoint
DEFAULT_VITE_PORT = 5173
DEFAULT_BACKEND_PORT = 5180
class ProcessGroup: class ProcessGroup:
"""Manage async subprocesses with automatic cleanup, like TaskGroup for processes.""" """Manage async subprocesses with automatic cleanup, like TaskGroup for processes."""
@@ -25,7 +24,7 @@ class ProcessGroup:
self, *cmd: str, cwd: str | None = None self, *cmd: str, cwd: str | None = None
) -> asyncio.subprocess.Process: ) -> asyncio.subprocess.Process:
"""Spawn a subprocess and track it.""" """Spawn a subprocess and track it."""
cmd_name = Path(cmd[0]).name cmd_name = Path(cmd[0]).stem
logger.info(">>> %s", " ".join([cmd_name, *cmd[1:]])) logger.info(">>> %s", " ".join([cmd_name, *cmd[1:]]))
proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd) proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd)
self._procs.append(proc) self._procs.append(proc)
@@ -58,14 +57,7 @@ class ProcessGroup:
async def __aexit__(self, exc_type, *_): async def __aexit__(self, exc_type, *_):
"""Wait for one process to exit, terminate others, then wait for all.""" """Wait for one process to exit, terminate others, then wait for all."""
cleanup_task = asyncio.create_task( await self._cleanup(immediate=exc_type is not None)
self._cleanup(immediate=exc_type is not None)
)
try:
await asyncio.shield(cleanup_task)
except asyncio.CancelledError:
# Shield was cancelled but cleanup_task continues - wait for it
await cleanup_task
async def _cleanup(self, immediate: bool = False): async def _cleanup(self, immediate: bool = False):
running = [p for p in self._procs if p.returncode is None] running = [p for p in self._procs if p.returncode is None]
@@ -74,51 +66,49 @@ class ProcessGroup:
if not immediate: if not immediate:
# Wait for any one process to exit # Wait for any one process to exit
await asyncio.wait( with suppress(asyncio.CancelledError):
[asyncio.create_task(p.wait()) for p in running], await asyncio.wait(
return_when=asyncio.FIRST_COMPLETED, [asyncio.create_task(p.wait()) for p in running],
) return_when=asyncio.FIRST_COMPLETED,
)
# Terminate remaining processes # Terminate remaining processes
for p in self._procs: for p in self._procs:
if p.returncode is None: if p.returncode is None:
try: with suppress(ProcessLookupError):
p.terminate() p.terminate()
except ProcessLookupError:
pass
# Wait for all to finish (with overall timeout) # Wait for all to finish (with overall timeout), shielded from cancellation
still_running = [p for p in self._procs if p.returncode is None] still_running = [p for p in self._procs if p.returncode is None]
if still_running: if still_running:
try: with suppress(asyncio.CancelledError):
await asyncio.wait_for( try:
asyncio.gather(*[p.wait() for p in still_running]), await asyncio.shield(
timeout=10, asyncio.wait_for(
) asyncio.gather(*[p.wait() for p in still_running]),
except TimeoutError: timeout=10,
for p in self._procs: )
if p.returncode is None: )
try: except TimeoutError:
p.kill() for p in self._procs:
except ProcessLookupError: if p.returncode is None:
pass with suppress(ProcessLookupError):
await p.wait() p.kill()
await p.wait()
async def check_ports_free(*urls: str) -> None: async def check_ports_free(*urls: str) -> None:
"""Verify URLs are not responding (ports are free). Raise SystemExit if any respond.""" """Verify URLs are not responding (ports are free). Raise SystemExit if any respond."""
async def check(client: httpx.AsyncClient, url: str) -> None:
with suppress(httpx.RequestError):
res = await client.get(url, timeout=0.1)
server = res.headers.get("server", "server")
logger.warning("Conflicting %s already running at %s", server, url)
raise SystemExit(1)
async with httpx.AsyncClient() as client: async with httpx.AsyncClient() as client:
for url in urls: await asyncio.gather(*[check(client, url) for url in urls])
try:
res = await client.get(url, timeout=0.1)
logger.warning(
"Conflicting %s already running at %s",
res.headers.get("server", "server"),
url,
)
raise SystemExit(1)
except httpx.RequestError:
pass # Expected - port is free
async def ready(url: str, path: str = "") -> None: async def ready(url: str, path: str = "") -> None:
@@ -142,13 +132,15 @@ async def ready(url: str, path: str = "") -> None:
await asyncio.sleep(0.1) await asyncio.sleep(0.1)
def setup_vite(endpoint: str) -> tuple[str, list[str], list[str]]: def setup_vite(
endpoint: str, default_port: int = 5173
) -> tuple[str, list[str], list[str]]:
"""Parse frontend endpoint and build commands. """Parse frontend endpoint and build commands.
Returns (url, install_cmd, dev_cmd). Returns (url, install_cmd, dev_cmd).
Raises SystemExit(1) on invalid config. Raises SystemExit(1) on invalid config.
""" """
endpoints = parse_endpoint(endpoint, DEFAULT_VITE_PORT) endpoints = parse_endpoint(endpoint, default_port)
if "uds" in endpoints[0]: if "uds" in endpoints[0]:
logger.warning("Unix sockets not supported with vite devserver") logger.warning("Unix sockets not supported with vite devserver")
@@ -161,18 +153,49 @@ def setup_vite(endpoint: str) -> tuple[str, list[str], list[str]]:
dev_cmd = find_dev_tool() dev_cmd = find_dev_tool()
if host != "localhost": if host != "localhost":
dev_cmd.append("--host" if len(endpoints) > 1 else f"--host={host}") dev_cmd.append("--host" if len(endpoints) > 1 else f"--host={host}")
if port != 5173: dev_cmd.append(f"--port={port}")
dev_cmd.append(f"--port={port}")
return f"http://{host}:{port}", install_cmd, dev_cmd return f"http://{host}:{port}", install_cmd, dev_cmd
def setup_fastapi( def setup_fastapi(
endpoint: str, module: str, default_port: int = DEFAULT_BACKEND_PORT endpoint: str, module: str, default_port: int = 8000
) -> tuple[str, list[str]]: ) -> tuple[str, list[str]]:
"""Parse backend endpoint and build fastapi dev command. """Parse backend endpoint and build uvicorn command.
Returns (url, cmd). 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(".")[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. Raises SystemExit(1) on invalid config.
""" """
endpoints = parse_endpoint(endpoint, default_port) endpoints = parse_endpoint(endpoint, default_port)
@@ -184,14 +207,5 @@ def setup_fastapi(
host = endpoints[0]["host"] host = endpoints[0]["host"]
port = endpoints[0]["port"] port = endpoints[0]["port"]
cmd = [ cmd = [cli, f"--listen={host}:{port}"]
"fastapi",
"dev",
"--entrypoint",
module,
"--host",
host,
"--port",
str(port),
]
return f"http://{host}:{port}", cmd return f"http://{host}:{port}", cmd