Compare commits

...
17 Commits
Author SHA1 Message Date
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
LeoVasanko f5fcd626e8 Update templates for fastapi_vue 0.5.0 catch-all toggle. 2026-01-25 01:07:56 +00:00
LeoVasanko 1d9ddf7dc5 Update target project python version requirement if needed. 2026-01-24 21:31:07 +00:00
LeoVasanko b189ae66e7 Smarter .gitignore patching. 2026-01-24 21:16:05 +00:00
15 changed files with 1362 additions and 504 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])
+267
View File
@@ -0,0 +1,267 @@
"""FastAPI static file serving with zstd compression and SPA support."""
import logging
import mimetypes
import os
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
# Dev mode: index files but don't load content, return error responses
_DEVMODE = os.getenv("FASTAPI_VUE_FRONTEND_URL")
logger = logging.getLogger("uvicorn.error") # Use FastAPI logging style
class Frontend:
"""Static file server with automatic zstd compression and caching.
Features:
- Automatic zstd compression for compressible files
- ETag-based caching with configurable cache headers
- SPA (Single Page Application) support
- Favicon handling from hashed assets
- 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 should have immutable cache headers
zstdlevel: Zstd compression level (default: 18)
favicon: Path to favicon for automatic /favicon.ico handling
"""
def __init__(
self,
directory: Path | str,
*,
index: str = "index.html",
spa: bool = False,
catch_all: bool | None = None,
cached: str | list[str] | None = None,
zstdlevel: int = 18,
favicon: str | None = None,
) -> 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
if cached is None:
self.cached_paths = []
elif isinstance(cached, str):
self.cached_paths = [cached]
else:
self.cached_paths = cached
self.zstdlevel = zstdlevel
self.favicon = favicon
self.devmode = bool(_DEVMODE)
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:
p = PurePosixPath(self.favicon)
base = str(p.with_suffix(""))
ext = p.suffix
hashed_path = next(
(path for path in www if path.startswith(base) and path.endswith(ext)),
None,
)
if hashed_path:
www["/favicon.ico"] = www[hashed_path]
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, *, log=True):
"""Load or reload static files from disk.
In dev mode (FASTAPI_VUE_FRONTEND_URL set), only indexes paths without loading content.
"""
if self.devmode:
# 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} %)"
)
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)
paths = self._devmode_paths if self.devmode else self.www.keys()
handler = _devmode_respond if self.devmode 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 self.devmode 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)
files = self._devmode_paths if self.devmode 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 self.devmode 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": f"Frontend assets served by Vite in dev mode. Connect via {_DEVMODE} 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 = ".."
+399 -112
View File
@@ -7,12 +7,14 @@ 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 os import os
import re import re
import shutil
import subprocess import subprocess
import sys import sys
from pathlib import Path from pathlib import Path
@@ -23,6 +25,10 @@ import tomlkit
# Template directory # Template directory
TEMPLATE_DIR = Path(__file__).parent / "template" TEMPLATE_DIR = Path(__file__).parent / "template"
# 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"
@@ -51,9 +57,7 @@ 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( frontend = Frontend(Path(__file__).with_name("frontend-build"), cached=["/assets/"])
Path(__file__).with_name("frontend-build"), spa=True, cached=["/assets/"]
)
""" """
# TypeScript health check script for Vue components # TypeScript health check script for Vue components
@@ -118,9 +122,83 @@ 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 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() return (TEMPLATE_DIR / path).read_text("UTF-8")
def find_module_name(project_dir: Path) -> str | None: def find_module_name(project_dir: Path) -> str | None:
@@ -129,7 +207,7 @@ def find_module_name(project_dir: Path) -> str | None:
if not pyproject.exists(): if not pyproject.exists():
return None return None
data = tomlkit.parse(pyproject.read_text()) data = tomlkit.parse(pyproject.read_text("UTF-8"))
if "project" in data and "name" in data["project"]: if "project" in data and "name" in data["project"]:
name = data["project"]["name"] name = data["project"]["name"]
@@ -138,10 +216,18 @@ def find_module_name(project_dir: Path) -> str | None:
return None return None
def find_fastapi_app(module_dir: Path) -> tuple[Path, str] | None: def find_fastapi_app(
module_dir: Path, project_dir: Path | None = None
) -> tuple[Path, str] | None:
"""Find the FastAPI app in a module directory. """Find the FastAPI app in a module directory.
Returns (file_path, app_variable_name) or None if not found. Returns (file_path, app_variable_name) or None if not found.
Search order:
1. Common app files in module_dir (app.py, main.py, etc.)
2. All .py files in module_dir
3. Subpackage indicated by CLI entrypoint in pyproject.toml
4. Follow re-exports in __init__.py files
""" """
# Common app file names to check first # Common app file names to check first
candidates = ["app.py", "main.py", "server.py", "api.py", "__init__.py"] candidates = ["app.py", "main.py", "server.py", "api.py", "__init__.py"]
@@ -154,20 +240,162 @@ def find_fastapi_app(module_dir: Path) -> tuple[Path, str] | None:
if result: if result:
return path, result return path, result
# Then check all .py files # Then check all .py files in module_dir
for path in module_dir.glob("*.py"): for path in module_dir.glob("*.py"):
if path.name not in candidates: if path.name not in candidates:
result = _find_app_in_file(path) result = _find_app_in_file(path)
if result: if result:
return path, result return path, result
# Try to find app via CLI entrypoint in pyproject.toml
if project_dir:
result = _find_app_via_entrypoint(module_dir, project_dir)
if result:
return result
return None
def _find_app_via_entrypoint(
module_dir: Path, project_dir: Path
) -> tuple[Path, str] | None:
"""Find FastAPI app by following the CLI entrypoint in pyproject.toml.
If pyproject.toml has a script like `myapp = "myapp.subpkg.__main__:main"`,
look in myapp/subpkg/ for the app (checking __init__.py exports and common files).
"""
pyproject = project_dir / "pyproject.toml"
if not pyproject.exists():
return None
try:
data = tomlkit.parse(pyproject.read_text("UTF-8"))
except Exception:
return None
scripts = data.get("project", {}).get("scripts", {})
if not scripts:
return None
module_name = module_dir.name
# Find script entries that reference this module
for script_name, entry in scripts.items():
if not isinstance(entry, str):
continue
# Parse entry like "module.subpkg.__main__:main"
if ":" not in entry:
continue
module_path, _ = entry.rsplit(":", 1)
parts = module_path.split(".")
# Check if this entry starts with our module
if not parts or parts[0] != module_name:
continue
# If there's a subpackage (e.g., module.fastapi.__main__), check there
if len(parts) >= 2:
# Build path to subpackage (exclude __main__ or similar)
subpkg_parts = [p for p in parts[1:] if not p.startswith("_")]
if subpkg_parts:
subpkg_dir = module_dir / "/".join(subpkg_parts)
if subpkg_dir.is_dir():
result = _find_app_in_subpackage(subpkg_dir)
if result:
return result
return None
def _find_app_in_subpackage(subpkg_dir: Path) -> tuple[Path, str] | None:
"""Find FastAPI app in a subpackage, following __init__.py exports."""
# First check __init__.py for re-exports like `from .mainapp import app`
init_file = subpkg_dir / "__init__.py"
if init_file.exists():
result = _follow_init_reexport(init_file, subpkg_dir)
if result:
return result
# Check common app file names in subpackage
for name in ["app.py", "main.py", "mainapp.py", "server.py", "api.py"]:
path = subpkg_dir / name
if path.exists():
result = _find_app_in_file(path)
if result:
return path, result
return None
def _find_existing_cli_entrypoint(project_dir: Path, module_name: str) -> str | None:
"""Check if pyproject.toml already has a CLI entrypoint for this module.
Returns the entrypoint string if found, None otherwise.
"""
pyproject = project_dir / "pyproject.toml"
if not pyproject.exists():
return None
try:
data = tomlkit.parse(pyproject.read_text("UTF-8"))
except Exception:
return None
scripts = data.get("project", {}).get("scripts", {})
if not scripts:
return None
# Look for any script that references our module
for script_name, entry in scripts.items():
if isinstance(entry, str) and entry.startswith(f"{module_name}."):
return f'{script_name} = "{entry}"'
return None
def _follow_init_reexport(init_file: Path, subpkg_dir: Path) -> tuple[Path, str] | None:
"""Follow a re-export in __init__.py to find the actual app file.
Looks for patterns like:
- from .mainapp import app
- from module.subpkg.mainapp import app
"""
try:
content = init_file.read_text("UTF-8")
except Exception:
return None
# Look for: from .module import app (or similar variable names)
# Pattern matches: from .mainapp import app, application, etc.
pattern = r"from\s+\.(\w+)\s+import\s+(\w+)"
for match in re.finditer(pattern, content):
module_name, var_name = match.groups()
if var_name.lower() in ("app", "application", "api"):
target_file = subpkg_dir / f"{module_name}.py"
if target_file.exists():
# Verify the app is actually there
app_var = _find_app_in_file(target_file)
if app_var:
return target_file, app_var
# Also check for absolute imports: from pkg.subpkg.module import app
abs_pattern = r"from\s+[\w.]+\.(\w+)\s+import\s+(\w+)"
for match in re.finditer(abs_pattern, content):
module_name, var_name = match.groups()
if var_name.lower() in ("app", "application", "api"):
target_file = subpkg_dir / f"{module_name}.py"
if target_file.exists():
app_var = _find_app_in_file(target_file)
if app_var:
return target_file, app_var
return None return None
def _find_app_in_file(path: Path) -> str | None: def _find_app_in_file(path: Path) -> str | None:
"""Find FastAPI app variable name in a file.""" """Find FastAPI app variable name in a file."""
try: try:
content = path.read_text() content = path.read_text("UTF-8")
except Exception: except Exception:
return None return None
@@ -202,7 +430,7 @@ def patch_app_file(
print(f"❌ Cannot patch {path} - file not found") print(f"❌ Cannot patch {path} - file not found")
return False return False
original_content = path.read_text() original_content = path.read_text("UTF-8")
marker = "from fastapi_vue import Frontend" marker = "from fastapi_vue import Frontend"
if marker in original_content: if marker in original_content:
@@ -242,31 +470,20 @@ def patch_app_file(
# Append route at end # Append route at end
lines.append("") lines.append("")
lines.append("# Serve the Vue frontend (needs to be last if SPA catch-all is used)")
lines.append(route_line) lines.append(route_line)
content = "\n".join(lines) content = "\n".join(lines)
# Try to patch lifespan function # Try to patch lifespan function - insert await frontend.load() before yield
lifespan_patched = False lifespan_patched = False
# Look for async def lifespan pattern and insert after the opening (and docstring if present) # Look for yield inside an async def lifespan function
lifespan_pattern = r"(async\s+def\s+lifespan\s*\([^)]*\)\s*(?:->.*?)?:\s*\n)" # Find the yield statement and insert before it
match = re.search(lifespan_pattern, content) yield_pattern = r"^([ \t]+)(yield\b)"
if match: yield_match = re.search(yield_pattern, content, re.MULTILINE)
insert_pos = match.end() if yield_match:
rest = content[insert_pos:] indent = yield_match.group(1)
insert_pos = yield_match.start()
# Detect indentation from the next line
indent_match = re.match(r"([ \t]*)", rest)
indent = (
indent_match.group(1) if indent_match and indent_match.group(1) else " "
)
# Check if there's a docstring and skip past it
docstring_pattern = r'^([ \t]*)("""[\s\S]*?"""|\'\'\'\'[\s\S]*?\'\'\')\s*\n'
docstring_match = re.match(docstring_pattern, rest)
if docstring_match:
insert_pos += docstring_match.end()
load_code = f"{indent}await frontend.load()\n" load_code = f"{indent}await frontend.load()\n"
content = content[:insert_pos] + load_code + content[insert_pos:] content = content[:insert_pos] + load_code + content[insert_pos:]
lifespan_patched = True lifespan_patched = True
@@ -280,7 +497,7 @@ def patch_app_file(
print(f"[DRY RUN] Would patch {path}") print(f"[DRY RUN] Would patch {path}")
return True return True
path.write_text(content) path.write_text(content, "UTF-8", newline="\n")
print(f"✅ Patched {path}") print(f"✅ Patched {path}")
if not lifespan_patched: if not lifespan_patched:
@@ -315,7 +532,7 @@ def patch_vite_config(
print(f"❌ Cannot patch {path} - file not found") print(f"❌ Cannot patch {path} - file not found")
return False return False
original_content = path.read_text() original_content = path.read_text("UTF-8")
marker = "vite-plugin-fastapi" marker = "vite-plugin-fastapi"
if marker in original_content: if marker in original_content:
@@ -363,14 +580,14 @@ 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"[DRY RUN] Would patch {path}")
return True return True
path.write_text(content) path.write_text(content, "UTF-8", newline="\n")
print(f"✅ Patched {path}") print(f"✅ Patched {path}")
return True return True
@@ -401,10 +618,10 @@ 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() original_content = target_file.read_text("UTF-8")
# Check if already patched # Check if already patched
if "/api/health" in original_content: if "/api/health" in original_content:
@@ -446,7 +663,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>")
@@ -463,14 +680,14 @@ 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"[DRY RUN] Would patch {target_file}")
return True return True
target_file.write_text(content) target_file.write_text(content, "UTF-8", newline="\n")
print(f"✅ Patched {target_file}") print(f"✅ Patched {target_file}")
return True return True
@@ -486,26 +703,29 @@ def write_file(
dry_run: bool = False, dry_run: bool = False,
executable: bool = False, executable: bool = False,
fallback_path: Path | None = None, fallback_path: Path | None = None,
force: bool = False,
) -> bool: ) -> bool:
"""Write content to a file, handling existing files and dry-run. """Write content to a file, handling existing files and dry-run.
If fallback_path is provided and the file exists without the upgrade marker, If fallback_path is provided and the file exists without the upgrade marker,
the content will be written to fallback_path instead of being skipped. the content will be written to fallback_path instead of being skipped.
If force=True, always overwrite without checking for upgrade marker.
""" """
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
if exists: if exists:
existing_content = path.read_text() existing_content = path.read_text("UTF-8")
if existing_content == content: if existing_content == content:
print(f"✔️ {path} (already up to date)") print(f"✔️ {path} (already up to date)")
return False return False
# If overwrite requested but file doesn't have upgrade marker # If overwrite requested but file doesn't have upgrade marker (unless force)
if overwrite and UPGRADE_MARKER not in existing_content: if overwrite and not force and UPGRADE_MARKER not in existing_content:
if fallback_path is not None: if fallback_path is not None:
# Write to fallback path instead # Write to fallback path instead
return _write_fallback_file( return _write_fallback_file(
@@ -520,7 +740,7 @@ def write_file(
return True return True
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content) 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)
action = "Updated" if exists else "Created" action = "Updated" if exists else "Created"
@@ -538,7 +758,7 @@ def _write_fallback_file(
"""Write content to a fallback .new.py file when original can't be overwritten.""" """Write content to a fallback .new.py file when original can't be overwritten."""
# Check if fallback already has same content # Check if fallback already has same content
if fallback_path.exists(): if fallback_path.exists():
if fallback_path.read_text() == content: if fallback_path.read_text("UTF-8") == content:
print(f"✔️ {fallback_path} (already up to date)") print(f"✔️ {fallback_path} (already up to date)")
return False return False
@@ -548,7 +768,7 @@ def _write_fallback_file(
return True return True
fallback_path.parent.mkdir(parents=True, exist_ok=True) fallback_path.parent.mkdir(parents=True, exist_ok=True)
fallback_path.write_text(content) 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)
print(f"✅ Created {fallback_path} (original customized by user)") print(f"✅ Created {fallback_path} (original customized by user)")
@@ -580,6 +800,20 @@ def merge_pyproject(
if "project" not in data: if "project" not in data:
data["project"] = tomlkit.table() data["project"] = tomlkit.table()
# Ensure Python version is at least 3.11 (required by fastapi-vue)
if "requires-python" in data["project"]:
req = data["project"]["requires-python"]
# Parse minimum version from strings like ">=3.10" or ">=3.9,<4"
import re
match = re.search(r">=\s*(\d+)\.(\d+)", req)
if match:
major, minor = int(match.group(1)), int(match.group(2))
if major < 3 or (major == 3 and minor < 11):
data["project"]["requires-python"] = ">=3.11"
else:
data["project"]["requires-python"] = ">=3.11"
# Add hatch build config # Add hatch build config
if "tool" not in data: if "tool" not in data:
data["tool"] = tomlkit.table() data["tool"] = tomlkit.table()
@@ -635,8 +869,6 @@ def find_js_runtime() -> tuple[str, str] | None:
Returns (tool_path, tool_name) where tool_name is "deno", "npm", or "bun". Returns (tool_path, tool_name) where tool_name is "deno", "npm", or "bun".
Returns None if no runtime is found. Returns None if no runtime is found.
""" """
import shutil
options = ["deno", "npm", "bun"] options = ["deno", "npm", "bun"]
# Check for JS_RUNTIME environment variable # Check for JS_RUNTIME environment variable
@@ -743,6 +975,9 @@ def cmd_setup(args: argparse.Namespace) -> int:
- Patching existing projects with integration files - Patching existing projects with integration files
- Updating already-patched projects - Updating already-patched projects
""" """
# Clear any existing virtual environment to avoid uv using the wrong one (warning)
os.environ.pop("VIRTUAL_ENV", None)
project_path = Path(args.project_dir) project_path = Path(args.project_dir)
# Handle both "." and "/path/to/project" # Handle both "." and "/path/to/project"
@@ -787,10 +1022,31 @@ def cmd_setup(args: argparse.Namespace) -> int:
print(f"📦 Module: {module_name}") 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}"
)
# 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),
} }
module_dir = project_dir / module_name module_dir = project_dir / module_name
@@ -798,7 +1054,9 @@ def cmd_setup(args: argparse.Namespace) -> int:
fastapi_vue_scripts = scripts_dir / "fastapi-vue" fastapi_vue_scripts = scripts_dir / "fastapi-vue"
# Find existing FastAPI app # Find existing FastAPI app
app_info = find_fastapi_app(module_dir) if module_dir.exists() else None app_info = (
find_fastapi_app(module_dir, project_dir) if module_dir.exists() else None
)
if app_info: if app_info:
app_file, app_var = app_info app_file, app_var = app_info
@@ -814,30 +1072,37 @@ def cmd_setup(args: argparse.Namespace) -> int:
# Create directories # Create directories
if not dry_run: if not dry_run:
fastapi_vue_scripts.mkdir(parents=True, exist_ok=True)
if not module_dir.exists(): if not module_dir.exists():
module_dir.mkdir(parents=True) module_dir.mkdir(parents=True)
fastapi_vue_scripts.mkdir(parents=True, exist_ok=True)
# === Install scripts (always update our own scripts) === # === Install scripts (always update our own scripts) ===
# util.py and build-frontend.py are internal and always overwritten # Scripts in fastapi-vue folder are internal and always overwritten
# devserver.py can be customized, so use fallback if needed # devserver.py can be customized, so use fallback if needed
internal_script_files = [
(fastapi_vue_scripts / "util.py", "scripts/fastapi-vue/util.py"),
(
fastapi_vue_scripts / "build-frontend.py",
"scripts/fastapi-vue/build-frontend.py",
),
]
for dest_path, template_path in internal_script_files: # Remove obsolete util.py if present
template = load_template(template_path) obsolete_util = fastapi_vue_scripts / "util.py"
content = render_template(template, **tpl_vars) if obsolete_util.exists():
write_file( if dry_run:
dest_path, print(f"[DRY RUN] Would remove obsolete {obsolete_util}")
content, else:
overwrite=True, obsolete_util.unlink()
dry_run=dry_run, print(f"🗑️ Removed obsolete {obsolete_util}")
)
# Copy all files from the template's fastapi-vue folder
template_fastapi_vue_dir = TEMPLATE_DIR / "scripts" / "fastapi-vue"
for template_file in template_fastapi_vue_dir.iterdir():
if template_file.is_file():
dest_path = fastapi_vue_scripts / template_file.name
template = template_file.read_text("UTF-8")
content = render_template(template, **tpl_vars)
write_file(
dest_path,
content,
overwrite=True,
dry_run=dry_run,
force=True, # Internal files, always overwrite
)
# devserver.py - use fallback if customized by user # devserver.py - use fallback if customized by user
devserver_path = scripts_dir / "devserver.py" devserver_path = scripts_dir / "devserver.py"
@@ -878,17 +1143,29 @@ def cmd_setup(args: argparse.Namespace) -> int:
template = load_template("backend/__main__.py") template = load_template("backend/__main__.py")
main_content = render_template(template, **tpl_vars) main_content = render_template(template, **tpl_vars)
# Use write_file with fallback - it handles all cases: # Check if project already has a CLI entrypoint in pyproject.toml
# - File doesn't exist: create it existing_cli = _find_existing_cli_entrypoint(project_dir, module_name)
# - File exists with marker: update it
# - File exists without marker: write to fallback if main_file.exists():
write_file( # File exists: update if it has the auto-upgrade marker, otherwise use fallback
main_file, write_file(
main_content, main_file,
overwrite=True, main_content,
dry_run=dry_run, overwrite=True,
fallback_path=main_fallback, dry_run=dry_run,
) fallback_path=main_fallback,
)
elif not existing_cli:
# No file and no existing entrypoint: create new __main__.py
write_file(
main_file,
main_content,
overwrite=False,
dry_run=dry_run,
)
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"
@@ -917,7 +1194,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
# === Update pyproject.toml === # === Update pyproject.toml ===
pyproject_path = project_dir / "pyproject.toml" pyproject_path = project_dir / "pyproject.toml"
if pyproject_path.exists(): if pyproject_path.exists():
old_content = pyproject_path.read_text() old_content = pyproject_path.read_text("UTF-8")
data = tomlkit.parse(old_content) data = tomlkit.parse(old_content)
updated = merge_pyproject(data, PYPROJECT_ADDITIONS, module_name) updated = merge_pyproject(data, PYPROJECT_ADDITIONS, module_name)
@@ -936,50 +1213,54 @@ def cmd_setup(args: argparse.Namespace) -> int:
elif dry_run: elif dry_run:
print(f"[DRY RUN] Would update {pyproject_path}") print(f"[DRY RUN] Would update {pyproject_path}")
else: else:
pyproject_path.write_text(new_content) pyproject_path.write_text(new_content, "UTF-8", newline="\n")
print(f"✅ Updated {pyproject_path}") print(f"✅ Updated {pyproject_path}")
# === Add dependencies using uv ===
if dry_run:
print("[DRY RUN] Would run: uv add -U 'fastapi[standard]' fastapi-vue")
print("[DRY RUN] Would run: uv add -U --group dev httpx")
else:
print("📦 Adding dependencies...")
result = subprocess.run(
["uv", "add", "-U", "fastapi[standard]", "fastapi-vue"],
cwd=project_dir,
check=False,
)
if result.returncode != 0:
print("⚠️ Failed to add main dependencies")
result = subprocess.run(
["uv", "add", "-U", "--group", "dev", "httpx"],
cwd=project_dir,
check=False,
)
if result.returncode != 0:
print("⚠️ Failed to add dev dependencies")
# === Update .gitignore === # === Update .gitignore ===
gitignore_path = project_dir / ".gitignore" gitignore_path = project_dir / ".gitignore"
gitignore_entry = f"{module_name}/frontend-build/" gitignore_entry = f"/{module_name}/frontend-build"
if gitignore_path.exists(): if gitignore_path.exists():
gitignore_content = gitignore_path.read_text() gitignore_content = gitignore_path.read_bytes()
if gitignore_entry not in gitignore_content: if b"frontend-build" in gitignore_content:
if dry_run: print("✔️ .gitignore (frontend-build already ignored)")
print(f"[DRY RUN] Would add {gitignore_entry} to .gitignore") elif dry_run:
else: print(f"[DRY RUN] Would add {gitignore_entry} to .gitignore")
with open(gitignore_path, "a") as f: else:
if not gitignore_content.endswith("\n"): nl = b"\r\n" if b"\r\n" in gitignore_content else b"\n"
f.write("\n") suffix = b"" if gitignore_content.endswith(nl) else nl
f.write(f"{gitignore_entry}\n") gitignore_path.write_bytes(
print(f"✅ Added {gitignore_entry} to .gitignore") gitignore_content + suffix + gitignore_entry.encode() + nl
)
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"[DRY RUN] Would create .gitignore with {gitignore_entry}")
else: else:
gitignore_path.write_text(f"{gitignore_entry}\n") gitignore_path.write_text(f"{gitignore_entry}\n", "UTF-8", newline="\n")
print("✅ Created .gitignore") print("✅ Created .gitignore")
# === Add dependencies using uv ===
uv_add_main = [
"uv",
"add",
"-q",
"-U",
"--no-sync",
"fastapi[standard]",
"fastapi-vue",
]
uv_add_dev = ["uv", "add", "-q", "-U", "--group", "dev", "httpx"]
if dry_run:
print(f"[DRY RUN] Would run: {' '.join(uv_add_main)}")
print(f"[DRY RUN] Would run: {' '.join(uv_add_dev)}")
else:
print("📦 Adding dependencies...")
result = subprocess.run(uv_add_main, cwd=project_dir, check=False)
if result.returncode != 0:
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)
print("✅ Setup complete!") print("✅ Setup complete!")
@@ -1044,6 +1325,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(
@@ -1053,6 +1335,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"
) )
+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()
+11 -40
View File
@@ -1,54 +1,25 @@
# 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 uvicorn from fastapi_vue import server
from fastapi_vue.hostutil import parse_endpoint
from uvicorn import Config, Server
from .APP_MODULE import APP_VAR DEFAULT_PORT = TEMPLATE_DEFAULT_PORT
DEFAULT_PORT = 5080
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():
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") server.run(
devmode = bool(os.getenv("FASTAPI_VUE_FRONTEND_URL")) "MODULE_NAME.APP_MODULE:APP_VAR",
endpoints = parse_endpoint(args.endpoint, DEFAULT_PORT) listen=args.listen,
run_server(endpoints, proxy=proxy, devmode=devmode) default_port=DEFAULT_PORT,
)
if __name__ == "__main__": if __name__ == "__main__":
+2 -4
View File
@@ -5,9 +5,7 @@ from fastapi import FastAPI
from fastapi_vue import Frontend from fastapi_vue import Frontend
# Vue Frontend static files # Vue Frontend static files
frontend = Frontend( frontend = Frontend(Path(__file__).with_name("frontend-build"), cached=["/assets/"])
Path(__file__).with_name("frontend-build"), spa=True, cached=["/assets/"]
)
@asynccontextmanager @asynccontextmanager
@@ -29,5 +27,5 @@ async def health_check():
return {"status": "ok"} return {"status": "ok"}
# Final catch-all route for frontend files (keep at end of file) # Serve the Vue frontend (needs to be last if SPA catch-all is used)
frontend.route(app, "/") frontend.route(app, "/")
+45 -238
View File
@@ -1,276 +1,83 @@
#!/usr/bin/env -S uv run #!/usr/bin/env -S uv run
# auto-upgrade@fastapi-vue-setup - remove this if you modify this file # auto-upgrade@fastapi-vue-setup - remove this if you modify this file
"""Run Vite development server for frontend and FastAPI backend with auto-reload. """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.
"""
import argparse import argparse
import asyncio import asyncio
import contextlib
import os import os
import sys
from contextlib import suppress
from pathlib import Path from pathlib import Path
from sys import stderr
import httpx # Import util.py from scripts/fastapi-vue (not a package, so we adjust sys.path)
from fastapi_vue.hostutil import parse_endpoint sys.path.insert(0, str(Path(__file__).with_name("fastapi-vue")))
from devutil import ( # type: ignore
ProcessGroup,
check_ports_free,
logger,
ready,
setup_fastapi,
setup_vite,
)
exec((Path(__file__).parent / "fastapi-vue/util.py").read_text("UTF-8")) # noqa: S102 DEFAULT_VITE_PORT = TEMPLATE_VITE_PORT
DEFAULT_DEV_PORT = TEMPLATE_DEV_PORT
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( async def run_devserver(frontend: str, backend: str) -> None:
vite_port: int, all_ifaces: bool reporoot = Path(__file__).parent.parent
) -> tuple[list[str], list[str], str]: front = reporoot / "frontend"
"""Resolve frontend install and dev commands. if not (front / "package.json").exists():
logger.warning("Frontend source not found at %s", front)
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) raise SystemExit(1)
result = find_js_runtime() # noqa # type: ignore viteurl, npm_install, vite = setup_vite(frontend, DEFAULT_VITE_PORT)
if result is None: backurl, uvicorn = setup_fastapi(
if not os.environ.get("JS_RUNTIME"): backend, "MODULE_NAME.APP_MODULE:APP_VAR", DEFAULT_DEV_PORT
stderr.write("┃ ⚠️ deno, npm or bun needed to run the frontend server.\n") )
raise SystemExit(1)
tool, name = result # Tell the everyone where the frontend and backend are (vite proxy, etc)
os.environ["FASTAPI_VUE_FRONTEND_URL"] = viteurl
os.environ["FASTAPI_VUE_BACKEND_URL"] = backurl
install_args = { async with ProcessGroup() as pg:
"deno": ("install", "--quiet", "--allow-scripts=npm:vue-demi"), npm_i = await pg.spawn(*npm_install, cwd=front)
"npm": ("install", "--silent"), await check_ports_free(viteurl, backurl)
"bun": ("install", "--silent"), await pg.spawn(*uvicorn)
} await pg.wait(npm_i, ready(backurl, path="/api/health?from=devserver.py"))
dev_args = { await pg.spawn(*vite, cwd=front)
"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()
async def run_devserver(
vite_port: int,
all_ifaces: bool,
backend_host: str,
backend_port: int,
) -> None:
"""Run the development server with install, backend, and frontend."""
install_cmd, dev_cmd, tool_name = resolve_frontend_tools(vite_port, all_ifaces)
# 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}"
backend_cmd = [
"uvicorn",
"MODULE_NAME.app:app",
"--host",
backend_host,
"--port",
str(backend_port),
"--reload",
]
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")
def main(): def main():
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description="Run Vite and FastAPI development servers", description="Run Vite and FastAPI development servers",
formatter_class=argparse.RawDescriptionHelpFormatter, formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=EPILOG, epilog=HELP_EPILOG,
) )
parser.add_argument( parser.add_argument(
"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 = parser.parse_args()
with suppress(KeyboardInterrupt):
asyncio.run(run_devserver(args.frontend, args.backend))
# 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 HELP_EPILOG = """
if "uds" in vite_endpoints[0]: scripts/devserver.py # Default ports on localhost
stderr.write("┃ ⚠️ Unix sockets not supported for frontend\n") scripts/devserver.py 3000 # Vite on localhost:3000
raise SystemExit(1) scripts/devserver.py :3000 --backend 8000 # *:3000, localhost:8000
if "uds" in backend_endpoints[0]:
stderr.write("┃ ⚠️ Unix sockets not supported for backend\n")
raise SystemExit(1)
vite_port = vite_endpoints[0]["port"] JS_RUNTIME environment variable can be used to select the JS runtime:
all_ifaces = len(vite_endpoints) > 1 npm, deno, bun, or full path to the runtime executable (node maps to npm).
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))
if __name__ == "__main__": if __name__ == "__main__":
+4 -23
View File
@@ -1,34 +1,15 @@
"""Hatch build hook for building Vue frontend during package build.""" """Hatch build hook for building Vue frontend during package build."""
import subprocess import sys
from pathlib import Path from pathlib import Path
from sys import stderr
from hatchling.builders.hooks.plugin.interface import BuildHookInterface # type: ignore from hatchling.builders.hooks.plugin.interface import BuildHookInterface # type: ignore
exec(Path(__file__).with_name("util.py").read_text("UTF-8")) # noqa: S102 sys.path.insert(0, str(Path(__file__).parent))
from buildutil import build
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): class CustomBuildHook(BuildHookInterface):
"""Build hook that compiles Vue frontend before packaging."""
def initialize(self, version, build_data): def initialize(self, version, build_data):
super().initialize(version, build_data) super().initialize(version, build_data)
stderr.write(">>> Building the frontend\n") build("frontend")
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
+191
View File
@@ -0,0 +1,191 @@
"""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
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(
[node_path, "--version"], capture_output=True, text=True, check=True
)
version_str = result.stdout.strip()
# Parse version like "v20.10.0" or "v18.17.1"
match = re.match(r"v(\d+)", version_str)
if match:
major_version = int(match.group(1))
if major_version >= 20:
return
raise RuntimeError(
f"Node.js {version_str} found, but v20+ required (install with nvm)"
)
except (subprocess.CalledProcessError, FileNotFoundError, ValueError):
pass
raise RuntimeError("Could not determine Node.js version")
def find_js_runtime() -> tuple[str, str]:
"""Find a JavaScript runtime from JS_RUNTIME env or auto-detect.
Returns (tool_path, tool_name) where tool_name is "deno", "npm", or "bun".
Raises JSRuntimeError if no suitable runtime is found.
"""
options = ["npm", "deno", "bun"]
node_version_error: RuntimeError | None = None
# Check for JS_RUNTIME environment variable
if js_runtime_env := os.environ.get("JS_RUNTIME"):
js_runtime = js_runtime_env
js_path = Path(js_runtime)
runtime_name = js_path.name
# Map node to npm
if runtime_name == "node":
runtime_name = "npm"
js_runtime = str(js_path.parent / "npm") if js_path.parent.name else "npm"
for option in options:
if option == runtime_name or runtime_name.startswith(option):
tool = shutil.which(js_runtime)
if tool is None:
raise RuntimeError(
f"JS_RUNTIME={js_runtime_env}: {option} not found"
)
# Check Node.js version if using npm
if option == "npm":
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path is None:
raise RuntimeError(
f"JS_RUNTIME={js_runtime_env}: node not found"
)
_check_node_version(node_path) # Raises on failure
return tool, option
raise RuntimeError(f"JS_RUNTIME={js_runtime_env} not recognized")
# Auto-detect
for option in options:
if tool := shutil.which(option):
# Check Node.js version if using npm
if option == "npm":
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path is None:
continue
try:
_check_node_version(node_path)
except RuntimeError as e:
node_version_error = e
continue # Try next runtime
return tool, option
# No runtime found - provide helpful error
if node_version_error:
raise node_version_error
raise RuntimeError("Node.js (v20+), Deno or Bun is required but none was found")
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"),
}
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", "dev", "--"),
"npm": ("--silent", "run", "dev", "--"),
"bun": ("run", "dev", "--"),
}
tool, name = find_js_runtime()
if name == "bun":
logger.warning(
"Bun has a bug in WS proxying (https://github.com/oven-sh/bun/issues/9882). Consider using npm instead."
)
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)
def run(cmd):
display_cmd = [Path(cmd[0]).stem, *cmd[1:]]
logger.info("### %s", " ".join(display_cmd))
subprocess.run(cmd, check=True, cwd=folder)
try:
run(install_cmd)
logger.info("")
run(build_cmd)
except subprocess.CalledProcessError:
raise SystemExit(1)
+190
View File
@@ -0,0 +1,190 @@
"""Utilities meant for devserver script, used only in source repository with dev deps."""
import asyncio
import subprocess
import sys
from collections.abc import Coroutine
from contextlib import suppress
from pathlib import Path
from typing import Any
import httpx
from buildutil import find_dev_tool, find_install_tool, logger
from fastapi_vue.hostutil import parse_endpoint
class ProcessGroup:
"""Manage async subprocesses with automatic cleanup, like TaskGroup for processes."""
def __init__(self):
self._procs: list[asyncio.subprocess.Process] = []
self._cmds: dict[int, str] = {} # pid -> command name
async def spawn(
self, *cmd: str, cwd: str | None = None
) -> asyncio.subprocess.Process:
"""Spawn a subprocess and track it."""
cmd_name = Path(cmd[0]).stem
logger.info(">>> %s", " ".join([cmd_name, *cmd[1:]]))
proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd)
self._procs.append(proc)
self._cmds[proc.pid] = cmd_name
return proc
async def wait(
self, *waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]"
) -> None:
"""Wait for processes/coroutines to complete, raise SystemExit on failure."""
async def wait_proc(proc: asyncio.subprocess.Process) -> None:
returncode = await proc.wait()
if returncode != 0:
cmd_name = self._cmds.get(proc.pid, "unknown")
raise subprocess.CalledProcessError(returncode, cmd_name)
tasks = [
wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w
for w in waitables
]
try:
await asyncio.gather(*tasks)
except subprocess.CalledProcessError as e:
logger.warning("%s failed with exit status %d", e.cmd, e.returncode)
raise SystemExit(1) from None
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, *_):
"""Wait for one process to exit, terminate others, then wait for all."""
await self._cleanup(immediate=exc_type is not None)
async def _cleanup(self, immediate: bool = False):
running = [p for p in self._procs if p.returncode is None]
if not running:
return
if not immediate:
# Wait for any one process to exit
with suppress(asyncio.CancelledError):
await asyncio.wait(
[asyncio.create_task(p.wait()) for p in running],
return_when=asyncio.FIRST_COMPLETED,
)
# Terminate remaining processes
for p in self._procs:
if p.returncode is None:
with suppress(ProcessLookupError):
p.terminate()
# Wait for all to finish (with overall timeout), shielded from cancellation
still_running = [p for p in self._procs if p.returncode is None]
if still_running:
with suppress(asyncio.CancelledError):
try:
await asyncio.shield(
asyncio.wait_for(
asyncio.gather(*[p.wait() for p in still_running]),
timeout=10,
)
)
except TimeoutError:
for p in self._procs:
if p.returncode is None:
with suppress(ProcessLookupError):
p.kill()
await p.wait()
async def check_ports_free(*urls: str) -> None:
"""Verify URLs are not responding (ports are free). Raise SystemExit if any respond."""
async def check(client: httpx.AsyncClient, url: str) -> None:
with suppress(httpx.RequestError):
res = await client.get(url, timeout=0.1)
server = res.headers.get("server", "server")
logger.warning("Conflicting %s already running at %s", server, url)
raise SystemExit(1)
async with httpx.AsyncClient() as client:
await asyncio.gather(*[check(client, url) for url in urls])
async def ready(url: str, path: str = "") -> None:
"""Wait for the server to be ready by polling an endpoint.
Raises SystemExit(1) if server doesn't start in time.
"""
max_attempts = 50
full_url = f"{url}{path}"
async with httpx.AsyncClient() as client:
for attempt in range(max_attempts):
try:
await client.get(full_url, timeout=1.0)
logger.info("✓ Backend ready!")
return
except httpx.RequestError:
if attempt == max_attempts - 1:
logger.warning("Backend didn't start in time")
raise SystemExit(1)
await asyncio.sleep(0.1)
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(".")[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
-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