Compare commits

..
10 Commits
15 changed files with 556 additions and 368 deletions
+1 -1
View File
@@ -84,7 +84,7 @@ my-app/
└── scripts/ └── scripts/
├── devserver.py # Run Vite and FastAPI together in dev mode ├── devserver.py # Run Vite and FastAPI together in dev mode
└── fastapi-vue/ # Dev utilities (only on the source tree) └── fastapi-vue/ # Dev utilities (only on the source tree)
├── build-frontend.py ├── buildhook.py
├── buildutil.py ├── buildutil.py
└── devutil.py └── devutil.py
``` ```
+2
View File
@@ -1,3 +1,5 @@
"""FastAPI Vue integration - serve Vue frontend from FastAPI."""
from .staticfiles import Frontend from .staticfiles import Frontend
__all__ = ["Frontend"] __all__ = ["Frontend"]
+82 -30
View File
@@ -1,8 +1,60 @@
"""Parse endpoint strings for uvicorn server configuration."""
import contextlib import contextlib
import ipaddress import ipaddress
from urllib.parse import urlparse from urllib.parse import urlparse
def _parse_all_interfaces(value: str) -> list[dict] | None:
"""Parse ':port' format to bind all interfaces."""
if not (value.startswith(":") and value != ":"):
return None
port_part = value[1:]
if not port_part.isdigit():
msg = f"Invalid port in '{value}'"
raise SystemExit(msg)
port = int(port_part)
return [{"host": "0.0.0.0", "port": port}, {"host": "::", "port": port}] # noqa: S104
def _parse_unix_socket(value: str) -> list[dict] | None:
"""Parse UNIX domain socket paths."""
if value.startswith("/"):
return [{"uds": value}]
if value.startswith("unix:"):
uds_path = value[5:] or None
if uds_path is None:
msg = "unix: path must not be empty"
raise SystemExit(msg)
return [{"uds": uds_path}]
return None
def _parse_unbracketed_ipv6(value: str, default_port: int) -> list[dict] | None:
"""Parse unbracketed IPv6 addresses."""
if value.count(":") <= 1 or value.startswith("["):
return None
try:
ipaddress.IPv6Address(value)
except ValueError as e:
msg = f"Invalid IPv6 address '{value}': {e}"
raise SystemExit(msg) from e
return [{"host": value, "port": default_port}]
def _parse_host_port(value: str, default_port: int) -> list[dict]:
"""Parse host[:port] or [ipv6][:port] using urlparse."""
parsed = urlparse(f"//{value}") # // prefix lets urlparse treat it as netloc
host = parsed.hostname or "localhost"
port = parsed.port or default_port
# Validate IP literals (optional; hostname passes through)
with contextlib.suppress(ValueError):
ipaddress.ip_address(host)
return [{"host": host, "port": port}]
def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]: def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
"""Parse an endpoint string into uvicorn bind configurations. """Parse an endpoint string into uvicorn bind configurations.
@@ -12,6 +64,7 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
Returns: Returns:
List of dicts with uvicorn bind kwargs (host/port or uds). List of dicts with uvicorn bind kwargs (host/port or uds).
Two entries may be returned for IPv4 and IPv6 (all interaces).
Supported forms: Supported forms:
- None or empty -> [{host: "localhost", port: default_port}] - None or empty -> [{host: "localhost", port: default_port}]
@@ -22,6 +75,7 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
- [ipv6]:port -> [{host: ipv6, port}] - [ipv6]:port -> [{host: ipv6, port}]
- ipv6 (unbracketed) -> [{host: ipv6, port: default_port}] - ipv6 (unbracketed) -> [{host: ipv6, port: default_port}]
- /path or unix:/path -> [{uds: path}] - /path or unix:/path -> [{uds: path}]
""" """
if not value: if not value:
return [{"host": "localhost", "port": default_port}] return [{"host": "localhost", "port": default_port}]
@@ -30,38 +84,36 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
if value.isdigit(): if value.isdigit():
return [{"host": "localhost", "port": int(value)}] return [{"host": "localhost", "port": int(value)}]
# Leading colon :port -> bind all interfaces (0.0.0.0 + ::) # Try specialized parsers in order
if value.startswith(":") and value != ":": result = _parse_all_interfaces(value)
port_part = value[1:] if result is not None:
if not port_part.isdigit(): return result
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) result = _parse_unix_socket(value)
if value.startswith("/"): if result is not None:
return [{"uds": value}] return result
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 result = _parse_unbracketed_ipv6(value, default_port)
if value.count(":") > 1 and not value.startswith("["): if result is not None:
try: return result
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]) # Fallback: host[:port], [ipv6][:port]
parsed = urlparse(f"//{value}") # // prefix lets urlparse treat it as netloc return _parse_host_port(value, default_port)
host = parsed.hostname or "localhost"
port = parsed.port or default_port
# Validate IP literals (optional; hostname passes through)
with contextlib.suppress(ValueError):
ipaddress.ip_address(host)
return [{"host": host, "port": port}] def parse_endpoints(
listen: str | list[str] | None = None,
default_port: int = 8000,
) -> list[dict]:
"""Parse listen strings into a list of endpoint dicts.
Args:
listen: Endpoint string(s) (see parse_endpoint for formats).
default_port: Port to use when not specified in listen args.
"""
if listen is None:
listen = [f"localhost:{default_port}"]
elif isinstance(listen, str):
listen = [listen]
return [ep for s in listen for ep in parse_endpoint(s, default_port)]
+15 -17
View File
@@ -1,12 +1,15 @@
"""Uvicorn server runner with multi-endpoint support."""
import asyncio import asyncio
import logging import logging
import os import os
from contextlib import suppress from contextlib import suppress
from typing import Any
import uvicorn import uvicorn
from uvicorn import Config, Server from uvicorn import Config, Server
from .hostutil import parse_endpoint from .hostutil import parse_endpoints
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -18,8 +21,8 @@ def run(
default_port: int = 8000, default_port: int = 8000,
reload: bool = False, reload: bool = False,
workers: int | None = None, workers: int | None = None,
**uvicorn_config, **uvicorn_config: Any, # noqa: ANN401
): ) -> None:
"""Run uvicorn server(s) for the given app. """Run uvicorn server(s) for the given app.
Args: Args:
@@ -29,14 +32,12 @@ def run(
reload: Enable auto-reload (requires uvicorn.run, single endpoint only). reload: Enable auto-reload (requires uvicorn.run, single endpoint only).
workers: Number of worker processes (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). **uvicorn_config: Additional uvicorn config options (overrides all other settings).
""" """
if listen is None: endpoints = parse_endpoints(listen, default_port)
listen = [f"localhost:{default_port}"] if not endpoints:
elif isinstance(listen, str): msg = "No endpoints to serve; check listen configuration"
listen = [listen] raise ValueError(msg)
endpoints: list[dict] = []
for ep in listen:
endpoints.extend(parse_endpoint(ep, default_port))
conf: dict[str, object] = {"app": app, "reload": reload, "workers": workers} conf: dict[str, object] = {"app": app, "reload": reload, "workers": workers}
proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1") proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1")
@@ -45,14 +46,14 @@ def run(
conf["forwarded_allow_ips"] = proxy conf["forwarded_allow_ips"] = proxy
conf.update(uvicorn_config) conf.update(uvicorn_config)
with suppress(KeyboardInterrupt): with suppress(KeyboardInterrupt, asyncio.CancelledError):
if reload or workers: if reload or workers:
serve_multiprocess(endpoints, **conf) serve_multiprocess(endpoints, **conf)
else: else:
asyncio.run(serve(endpoints, **conf)) asyncio.run(serve(endpoints, **conf))
async def serve(endpoints: list[dict], **kwargs) -> None: async def serve(endpoints: list[dict], **kwargs: Any) -> None: # noqa: ANN401
"""Serve the given endpoints in current process/loop. Does not spawn extra processes.""" """Serve the given endpoints in current process/loop. Does not spawn extra processes."""
forbidden = {"reload", "workers"} & {k for k, v in kwargs.items() if v} forbidden = {"reload", "workers"} & {k for k, v in kwargs.items() if v}
if forbidden: if forbidden:
@@ -63,13 +64,10 @@ async def serve(endpoints: list[dict], **kwargs) -> None:
await asyncio.gather(*(Server(Config(**kwargs, **ep)).serve() for ep in endpoints)) await asyncio.gather(*(Server(Config(**kwargs, **ep)).serve() for ep in endpoints))
def serve_multiprocess(endpoints: list[dict], **kwargs) -> None: def serve_multiprocess(endpoints: list[dict], **kwargs: Any) -> None: # noqa: ANN401
"""Serve using uvicorn.run() for reload/workers support. Only first endpoint is used.""" """Serve using uvicorn.run() for reload/workers support. Only first endpoint is used."""
if len(endpoints) > 1: if len(endpoints) > 1:
eps = [ eps = [ep["uds"] if "uds" in ep else f"{ep['host']}:{ep['port']}" for ep in endpoints]
ep["uds"] if "uds" in ep else f"{ep['host']}:{ep['port']}"
for ep in endpoints
]
logger.warning( logger.warning(
"Current mode supports only one endpoint. Listening: %s, skipped: %s", "Current mode supports only one endpoint. Listening: %s, skipped: %s",
eps[0], eps[0],
+32 -25
View File
@@ -25,7 +25,7 @@ __all__ = ["Frontend"]
class Assets: class Assets:
"""Default cached value to /assets/""" """Default cached value to /assets/."""
@staticmethod @staticmethod
def parse(cached: str | list[str] | Assets) -> list[str]: def parse(cached: str | list[str] | Assets) -> list[str]:
@@ -37,7 +37,8 @@ class Assets:
case list(): case list():
return cached return cached
case _: case _:
raise ValueError(f"Invalid cached value: {cached!r}") msg = f"Invalid cached value: {cached!r}"
raise ValueError(msg)
class Frontend: class Frontend:
@@ -55,27 +56,29 @@ class Frontend:
index: Name of the index file (default: "index.html") index: Name of the index file (default: "index.html")
spa: Enable SPA mode - serve index.html for unknown routes (default: False) spa: Enable SPA mode - serve index.html for unknown routes (default: False)
cached: Path prefixes that are immutable (default: "/assets/") 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 favicon: Wildcard path to favicon. E.g. /assets/logo*.png matches Vite output
zstdlevel: Zstd compression level (default: 18) zstdlevel: Zstd compression level (default: 18)
""" """
def __init__( def __init__( # noqa: PLR0913
self, self,
directory: Path | str, directory: Path | str,
*, *,
index: str = "index.html", index: str = "index.html",
spa: bool = False, spa: bool = False,
catch_all: bool | None = None, catch_all: bool | None = None,
cached: str | list[str] | Assets = Assets(), cached: str | list[str] | Assets | None = None,
favicon: str | None = None, favicon: str | None = None,
zstdlevel: int = 18, zstdlevel: int = 18,
) -> None: ) -> None:
"""Initialize Frontend with given configuration."""
self.www: dict[str, tuple[bytes, bytes | None, dict]] = {} self.www: dict[str, tuple[bytes, bytes | None, dict]] = {}
self.base: Path = Path(directory) self.base: Path = Path(directory)
self.index = index self.index = index
self.spa = spa self.spa = spa
self._catch_all = spa if catch_all is None else catch_all self._catch_all = spa if catch_all is None else catch_all
self.cached_paths = Assets.parse(cached) self.cached_paths = Assets.parse(cached if cached is not None else Assets())
self.zstdlevel = zstdlevel self.zstdlevel = zstdlevel
self.favicon = favicon self.favicon = favicon
self._app: FastAPI | None = None self._app: FastAPI | None = None
@@ -107,11 +110,12 @@ class Frontend:
paths.add("/favicon.ico") paths.add("/favicon.ico")
return paths return paths
def _load(self): def _load(self) -> dict[str, tuple[bytes, bytes | None, dict]]:
"""Load static files from disk with compression.""" """Load static files from disk with compression."""
www: dict[str, tuple[bytes, bytes | None, dict]] = {} www: dict[str, tuple[bytes, bytes | None, dict]] = {}
if not self.base.exists(): if not self.base.exists():
raise ValueError(f"Frontend folder {self.base} not found (try uv build)") msg = f"Frontend folder {self.base} not found (try uv build)"
raise ValueError(msg)
paths = [PurePath()] paths = [PurePath()]
while paths: while paths:
current = self.base / paths.pop(0) current = self.base / paths.pop(0)
@@ -133,17 +137,14 @@ class Frontend:
headers = { headers = {
"etag": f'"{etag}"', "etag": f'"{etag}"',
"last-modified": format_date_time(mtime), "last-modified": format_date_time(mtime),
"cache-control": ( "cache-control": ("max-age=31536000, immutable" if cached else "no-cache"),
"max-age=31536000, immutable" if cached else "no-cache"
),
"content-type": mime, "content-type": mime,
} }
zstd = ZstdCompressor(self.zstdlevel).compress(data) zstd = ZstdCompressor(self.zstdlevel).compress(data)
if len(zstd) >= len(data): if len(zstd) >= len(data):
zstd = None zstd = None
www[name] = data, zstd, headers www[name] = data, zstd, headers
if self.favicon: if self.favicon and (m := fnmatch.filter(www, self.favicon)):
if m := fnmatch.filter(www, self.favicon):
data, zstd, headers = www[m[0]] data, zstd, headers = www[m[0]]
if "immutable" in headers.get("cache-control", ""): if "immutable" in headers.get("cache-control", ""):
headers = {**headers, "cache-control": "max-age=86400"} headers = {**headers, "cache-control": "max-age=86400"}
@@ -161,7 +162,7 @@ class Frontend:
) )
return www return www
async def load(self, *, debug: bool | None = None, log: bool = True): async def load(self, *, debug: bool | None = None, log: bool = True) -> None:
"""Load or reload static files from disk. """Load or reload static files from disk.
In debug mode, returns 409 instead of files (avoid accidental use of stale builds) In debug mode, returns 409 instead of files (avoid accidental use of stale builds)
@@ -187,13 +188,19 @@ class Frontend:
ratio = comp / raw * 100 if raw else 100.0 ratio = comp / raw * 100 if raw else 100.0
if log and self.www: if log and self.www:
logger.info( logger.info(
f"{self.base.name}: {len(self.www)} files in {1000 * duration:.1f} ms | " "%s: %d files in %.1f ms | zstd %d files %.2f->%.2f MB (%.0f %%)",
f"zstd {len(compfiles)} files {1e-6 * raw:.2f}->{1e-6 * comp:.2f} MB ({ratio:.0f} %)" self.base.name,
len(self.www),
1000 * duration,
len(compfiles),
1e-6 * raw,
1e-6 * comp,
ratio,
) )
if self.favicon and "/favicon.ico" not in self.www: if self.favicon and "/favicon.ico" not in self.www:
logger.warning("Favicon not found: %s", self.favicon) logger.warning("Favicon not found: %s", self.favicon)
def route(self, app: FastAPI, mount_path="/"): def route(self, app: FastAPI, mount_path: str = "/") -> None:
"""Register frontend routes with a FastAPI app. """Register frontend routes with a FastAPI app.
In SPA/catch-all mode, this must only be called only after all other routes. In SPA/catch-all mode, this must only be called only after all other routes.
@@ -204,6 +211,7 @@ class Frontend:
Args: Args:
app: FastAPI application instance app: FastAPI application instance
mount_path: Path where the frontend should be mounted (default: "/") mount_path: Path where the frontend should be mounted (default: "/")
""" """
self._app = app self._app = app
self._mount_path = mount_path.rstrip("/") self._mount_path = mount_path.rstrip("/")
@@ -214,7 +222,7 @@ class Frontend:
path = self._mount_path + "{path:path}" path = self._mount_path + "{path:path}"
app.api_route(path, methods=["GET", "HEAD"], name="frontend")(self.handle) app.api_route(path, methods=["GET", "HEAD"], name="frontend")(self.handle)
def _register_routes(self): def _register_routes(self) -> None:
"""Register individual routes for each loaded file (non-catch_all mode).""" """Register individual routes for each loaded file (non-catch_all mode)."""
if self._app is None or self._catch_all: if self._app is None or self._catch_all:
return return
@@ -240,18 +248,19 @@ class Frontend:
for p in paths for p in paths
] ]
def _respond(self, request: Request, name: str): def _respond(self, request: Request, name: str) -> Response:
"""Serve a static file with ETag and compression support.""" """Serve a static file with ETag and compression support."""
data, zstd, headers = self.www[name] data, zstd, headers = self.www[name]
if request.headers.get("if-none-match") == headers["etag"]: if request.headers.get("if-none-match") == headers["etag"]:
return Response(status_code=304, headers=headers) return Response(status_code=304, headers=headers)
if zstd and "zstd" in request.headers.get("accept-encoding", ""): if zstd and "zstd" in request.headers.get("accept-encoding", ""):
return Response( return Response(
content=zstd, headers={**headers, "content-encoding": "zstd"} content=zstd,
headers={**headers, "content-encoding": "zstd"},
) )
return Response(content=data, headers=headers) return Response(content=data, headers=headers)
def handle(self, request: Request, path: str): def handle(self, request: Request, path: str) -> Response | RedirectResponse:
"""SPA catch-all handler with directory redirects and fallback to index.""" """SPA catch-all handler with directory redirects and fallback to index."""
name = path.removesuffix(self.index) name = path.removesuffix(self.index)
debug = getattr(self._app, "debug", False) debug = getattr(self._app, "debug", False)
@@ -271,11 +280,9 @@ class Frontend:
return (_devmode_respond if debug else self._respond)(request, name) return (_devmode_respond if debug else self._respond)(request, name)
def _devmode_respond(request: Request, name=""): def _devmode_respond(_request: Request, _name: str = "") -> JSONResponse:
"""Return error response directing to Vite server.""" """Return error response directing to Vite server."""
return JSONResponse( return JSONResponse(
status_code=409, status_code=409,
content={ content={"detail": "[devmode] Use Vite devserver instead."},
"detail": "[devmode] Not serving frontend files here. Should you connect to Vite instead?"
},
) )
+224 -176
View File
@@ -1,4 +1,4 @@
"""FastAPI-Vue Integration Tool """FastAPI-Vue Integration Tool.
Create new FastAPI+Vue projects or patch existing ones with integrated build/dev systems. Create new FastAPI+Vue projects or patch existing ones with integrated build/dev systems.
@@ -13,14 +13,18 @@ Options:
import argparse import argparse
import ast import ast
import contextlib
import hashlib
import importlib.metadata import importlib.metadata
import os import os
import platform
import re import re
import shutil import shutil
import subprocess import subprocess
import sys import sys
from pathlib import Path from pathlib import Path
from textwrap import indent from textwrap import indent
from typing import Literal
import tomlkit import tomlkit
@@ -38,54 +42,73 @@ def print_boxed(text: str) -> None:
print(f"{'' * width}") print(f"{'' * width}")
def ruff_format_content(content: str, target_path: Path) -> str: def ruff_format_content(
content: str, target_path: Path, *, mode: Literal["isort", "full"] = "full"
) -> str:
"""Format Python content using ruff with project settings. """Format Python content using ruff with project settings.
Writes to a temp file (.new.py) next to target, runs ruff check (import sorting) Writes to a temp file next to target, runs ruff check (import sorting)
and ruff format on it, reads back the result, and cleans up. and optionally ruff format on it, reads back the result, and cleans up.
Returns the formatted content, or original if ruff fails. Returns the formatted content, or original if ruff fails.
mode='isort' only sorts imports; mode='full' also formats.
""" """
temp_file = target_path.with_suffix(".new.py") temp_file = target_path.with_suffix(".new.py")
try: try:
temp_file.write_text(content, "UTF-8", newline="\n") temp_file.write_text(content, "UTF-8", newline="\n")
# Sort imports first if mode == "isort":
subprocess.run( # Sort imports only (ignore exit code)
[ result = subprocess.run( # noqa: S603
"uv", [ # noqa: S607
"run",
"--with",
"ruff",
"ruff", "ruff",
"check", "check",
"--select", "--select",
"I", "I",
"--fix", "--fix",
"--output-format=concise",
str(temp_file), str(temp_file),
], ],
cwd=target_path.parent, cwd=target_path.parent,
capture_output=True, capture_output=True,
check=False,
) )
# Then format if result.returncode != 0:
result = subprocess.run( print(result.stdout.decode())
["uv", "run", "--with", "ruff", "ruff", "format", str(temp_file)], return temp_file.read_text("UTF-8")
# Full mode: fix all auto-fixable lint violations (ignore exit code)
result = subprocess.run( # noqa: S603
[ # noqa: S607
"ruff",
"check",
"--fix",
"--output-format=concise",
str(temp_file),
],
cwd=target_path.parent, cwd=target_path.parent,
capture_output=True, capture_output=True,
check=False,
) )
if result.returncode == 0: if result.returncode != 0:
print(result.stdout.decode())
# Then format (ignore exit code)
result = subprocess.run( # noqa: S603
["ruff", "format", str(temp_file)], # noqa: S607
cwd=target_path.parent,
capture_output=True,
check=False,
)
if result.returncode != 0:
print(result.stdout.decode())
return temp_file.read_text("UTF-8") return temp_file.read_text("UTF-8")
except Exception: except OSError:
pass pass
finally: finally:
try: with contextlib.suppress(Exception):
temp_file.unlink(missing_ok=True) temp_file.unlink(missing_ok=True)
except Exception:
pass
return content return content
def uv_add_packages( def uv_add_packages(packages: list[str], *, cwd: Path, group: str | None = None) -> None:
packages: list[str], *, cwd: Path, group: str | None = None
) -> None:
"""Add packages using uv.""" """Add packages using uv."""
cmd = ["uv", "add", "-q", "-U"] cmd = ["uv", "add", "-q", "-U"]
if group: if group:
@@ -93,7 +116,7 @@ def uv_add_packages(
else: else:
cmd.append("--no-sync") cmd.append("--no-sync")
cmd.extend(packages) cmd.extend(packages)
result = subprocess.run(cmd, cwd=cwd, check=False) result = subprocess.run(cmd, cwd=cwd, check=False) # noqa: S603
if result.returncode != 0: if result.returncode != 0:
label = f" ({group})" if group else "" label = f" ({group})" if group else ""
print(f"⚠️ Failed to add{label} dependencies") print(f"⚠️ Failed to add{label} dependencies")
@@ -103,6 +126,9 @@ def uv_add_packages(
# If vite == dev, dev is incremented by 100 # If vite == dev, dev is incremented by 100
DEFAULT_PORTS = (3100, 3100, 3200) DEFAULT_PORTS = (3100, 3100, 3200)
# Default health check path for devserver backend readiness check
DEFAULT_HEALTH = "/api/health?from=devserver.py"
# 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"
@@ -116,9 +142,7 @@ PYPROJECT_ADDITIONS = {
"artifacts": ["MODULE_NAME/frontend-build"], "artifacts": ["MODULE_NAME/frontend-build"],
"targets": { "targets": {
"sdist": { "sdist": {
"hooks": { "hooks": {"custom": {"path": "scripts/fastapi-vue/buildhook.py"}},
"custom": {"path": "scripts/fastapi-vue/build-frontend.py"}
},
} }
}, },
"only-packages": True, "only-packages": True,
@@ -127,6 +151,10 @@ PYPROJECT_ADDITIONS = {
}, },
} }
# Old build hook path that should be migrated to the new name
OLD_BUILD_HOOK_PATH = "scripts/fastapi-vue/build-frontend.py"
NEW_BUILD_HOOK_PATH = "scripts/fastapi-vue/buildhook.py"
# Frontend instantiation block for patching existing apps # Frontend instantiation block for patching existing apps
FRONTEND_BLOCK = """ FRONTEND_BLOCK = """
@@ -226,7 +254,8 @@ def parse_ports(ports_str: str | None) -> tuple[int, int, int]:
vite = int(parts[1]) vite = int(parts[1])
dev = int(parts[2]) dev = int(parts[2])
else: else:
raise ValueError(f"Invalid ports format: {ports_str}") msg = f"Invalid ports format: {ports_str}"
raise ValueError(msg)
# Auto-adjust dev if it conflicts with vite # Auto-adjust dev if it conflicts with vite
if dev == vite: if dev == vite:
@@ -253,23 +282,18 @@ def find_import_insertion_line(source: str) -> int:
return 2 if source.startswith("#!") else 1 return 2 if source.startswith("#!") else 1
def extract_existing_ports(project_dir: Path) -> tuple[int, int, int] | None: def extract_existing_ports(project_dir: Path, main: Path) -> tuple[int, int, int] | None:
"""Extract existing port configuration from project files. """Extract existing port configuration from project files.
Returns (default, vite, dev) or None if not found. 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 default_port = None
vite_port = None vite_port = None
dev_port = None dev_port = None
# Try to extract DEFAULT_PORT from __main__.py # Try to extract DEFAULT_PORT from the CLI main module
main_file = project_dir / module_name / "__main__.py" if main.exists():
if main_file.exists(): content = main.read_text("UTF-8")
content = main_file.read_text("UTF-8")
match = re.search(r"DEFAULT_PORT\s*=\s*(\d+)", content) match = re.search(r"DEFAULT_PORT\s*=\s*(\d+)", content)
if match: if match:
default_port = int(match.group(1)) default_port = int(match.group(1))
@@ -296,6 +320,30 @@ def extract_existing_ports(project_dir: Path) -> tuple[int, int, int] | None:
return None return None
# Sentinel for "not found" in extract_existing_health
_HEALTH_NOT_FOUND = object()
def extract_existing_health(project_dir: Path) -> str | object:
"""Extract existing health path configuration from devserver.py.
Returns:
- The path string (may be empty to disable)
- _HEALTH_NOT_FOUND sentinel if not found or file doesn't exist
"""
devserver_file = project_dir / "scripts" / "devserver.py"
if not devserver_file.exists():
return _HEALTH_NOT_FOUND
content = devserver_file.read_text("UTF-8")
# Match HEALTH = "/path" or HEALTH = ""
match = re.search(r'^HEALTH\s*=\s*"([^"]*)"', content, re.MULTILINE)
if match:
return match.group(1)
return _HEALTH_NOT_FOUND
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")
@@ -316,9 +364,7 @@ def find_module_name(project_dir: Path) -> str | None:
return None return None
def find_fastapi_app( def find_fastapi_app(module_dir: Path, project_dir: Path | None = None) -> tuple[Path, str] | None:
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.
@@ -356,9 +402,7 @@ def find_fastapi_app(
return None return None
def _find_app_via_entrypoint( def _find_app_via_entrypoint(module_dir: Path, project_dir: Path) -> tuple[Path, str] | None:
module_dir: Path, project_dir: Path
) -> tuple[Path, str] | None:
"""Find FastAPI app by following the CLI entrypoint in pyproject.toml. """Find FastAPI app by following the CLI entrypoint in pyproject.toml.
If pyproject.toml has a script like `myapp = "myapp.subpkg.__main__:main"`, If pyproject.toml has a script like `myapp = "myapp.subpkg.__main__:main"`,
@@ -370,7 +414,7 @@ def _find_app_via_entrypoint(
try: try:
data = tomlkit.parse(pyproject.read_text("UTF-8")) data = tomlkit.parse(pyproject.read_text("UTF-8"))
except Exception: except (OSError, ValueError):
return None return None
scripts = data.get("project", {}).get("scripts", {}) scripts = data.get("project", {}).get("scripts", {})
@@ -380,7 +424,7 @@ def _find_app_via_entrypoint(
module_name = module_dir.name module_name = module_dir.name
# Find script entries that reference this module # Find script entries that reference this module
for script_name, entry in scripts.items(): for entry in scripts.values():
if not isinstance(entry, str): if not isinstance(entry, str):
continue continue
# Parse entry like "module.subpkg.__main__:main" # Parse entry like "module.subpkg.__main__:main"
@@ -438,7 +482,7 @@ def _add_devmode_to_main(content: str) -> str:
insert_idx = 0 insert_idx = 0
for i, line in enumerate(lines): for i, line in enumerate(lines):
stripped = line.strip() stripped = line.strip()
if stripped.startswith("import ") or stripped.startswith("from "): if stripped.startswith(("import ", "from ")):
insert_idx = i + 1 insert_idx = i + 1
elif stripped and not stripped.startswith("#"): elif stripped and not stripped.startswith("#"):
break break
@@ -471,7 +515,7 @@ def _find_existing_cli_module_path(project_dir: Path, module_name: str) -> str |
try: try:
data = tomlkit.parse(pyproject.read_text("UTF-8")) data = tomlkit.parse(pyproject.read_text("UTF-8"))
except Exception: except (OSError, ValueError):
return None return None
scripts = data.get("project", {}).get("scripts", {}) scripts = data.get("project", {}).get("scripts", {})
@@ -479,10 +523,9 @@ def _find_existing_cli_module_path(project_dir: Path, module_name: str) -> str |
return None return None
# Look for any script that references our module # Look for any script that references our module
for script_name, entry in scripts.items(): for entry in scripts.values():
if isinstance(entry, str) and entry.startswith(f"{module_name}."): if isinstance(entry, str) and entry.startswith(f"{module_name}.") and ":" in entry:
# Extract module path from "module.subpkg.__main__:main" # Extract module path from "module.subpkg.__main__:main"
if ":" in entry:
module_path, _ = entry.rsplit(":", 1) module_path, _ = entry.rsplit(":", 1)
return module_path return module_path
@@ -498,7 +541,7 @@ def _follow_init_reexport(init_file: Path, subpkg_dir: Path) -> tuple[Path, str]
""" """
try: try:
content = init_file.read_text("UTF-8") content = init_file.read_text("UTF-8")
except Exception: except OSError:
return None return None
# Look for: from .module import app (or similar variable names) # Look for: from .module import app (or similar variable names)
@@ -532,7 +575,7 @@ 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("UTF-8") content = path.read_text("UTF-8")
except Exception: except OSError:
return None return None
# Look for FastAPI() instantiation patterns # Look for FastAPI() instantiation patterns
@@ -544,17 +587,15 @@ def _find_app_in_file(path: Path) -> str | None:
return None return None
def render_template(template: str, **kwargs) -> str: def render_template(template: str, **kwargs: str) -> str:
"""Simple template rendering replacing KEY with value.""" """Render a template, replacing KEY with value."""
result = template result = template
for key, value in kwargs.items(): for key, value in kwargs.items():
result = result.replace(key, value) result = result.replace(key, value)
return result return result
def patch_app_file( def patch_app_file(path: Path, main_module_path: str, app_var: str, *, dry: bool = False) -> bool:
path: Path, main_module_path: str, app_var: str, dry: bool = False
) -> bool:
"""Patch an existing app.py with frontend integration. """Patch an existing app.py with frontend integration.
Inserts imports at top (ruff will sort them), route at bottom, Inserts imports at top (ruff will sort them), route at bottom,
@@ -598,9 +639,7 @@ def patch_app_file(
content = content.rstrip("\n") + "\n" + import_text content = content.rstrip("\n") + "\n" + import_text
else: else:
# Insert at the found position # Insert at the found position
content = ( content = "".join(lines[:insert_idx]) + import_text + "".join(lines[insert_idx:])
"".join(lines[:insert_idx]) + import_text + "".join(lines[insert_idx:])
)
# Insert FRONTEND_BLOCK after last import (only if Frontend wasn't already there) # Insert FRONTEND_BLOCK after last import (only if Frontend wasn't already there)
if not has_frontend: if not has_frontend:
@@ -608,7 +647,7 @@ def patch_app_file(
last_import_idx = 0 last_import_idx = 0
for i, line in enumerate(lines): for i, line in enumerate(lines):
stripped = line.strip() stripped = line.strip()
if stripped.startswith("import ") or stripped.startswith("from "): if stripped.startswith(("import ", "from ")):
last_import_idx = i last_import_idx = i
elif stripped and not stripped.startswith("#") and last_import_idx > 0: elif stripped and not stripped.startswith("#") and last_import_idx > 0:
break break
@@ -619,9 +658,7 @@ def patch_app_file(
if route_line not in content: if route_line not in content:
lines = content.split("\n") lines = content.split("\n")
lines.append("") lines.append("")
lines.append( lines.append("# Serve the Vue frontend (needs to be last if SPA catch-all is used)")
"# 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)
@@ -632,10 +669,7 @@ def patch_app_file(
args = match.group(2) args = match.group(2)
if "debug" not in args: if "debug" not in args:
# Add debug=DEVMODE as last argument # Add debug=DEVMODE as last argument
if args.strip(): new_args = f"{args}, debug=DEVMODE" if args.strip() else "debug=DEVMODE"
new_args = f"{args}, debug=DEVMODE"
else:
new_args = "debug=DEVMODE"
content = ( content = (
content[: match.start()] content[: match.start()]
+ match.group(1) + match.group(1)
@@ -671,11 +705,7 @@ def patch_app_file(
if insert_idx >= len(lines): if insert_idx >= len(lines):
content = content.rstrip("\n") + "\n" + import_text content = content.rstrip("\n") + "\n" + import_text
else: else:
content = ( content = "".join(lines[:insert_idx]) + import_text + "".join(lines[insert_idx:])
"".join(lines[:insert_idx])
+ import_text
+ "".join(lines[insert_idx:])
)
# Insert lifespan block before the FastAPI() call # Insert lifespan block before the FastAPI() call
fastapi_line_pattern = r"^(\w+\s*=\s*FastAPI\s*\()" fastapi_line_pattern = r"^(\w+\s*=\s*FastAPI\s*\()"
@@ -693,10 +723,7 @@ def patch_app_file(
fastapi_match = re.search(fastapi_pattern, content, re.DOTALL) fastapi_match = re.search(fastapi_pattern, content, re.DOTALL)
if fastapi_match and "lifespan" not in fastapi_match.group(2): if fastapi_match and "lifespan" not in fastapi_match.group(2):
args = fastapi_match.group(2) args = fastapi_match.group(2)
if args.strip(): new_args = f"{args}, lifespan=lifespan" if args.strip() else "lifespan=lifespan"
new_args = f"{args}, lifespan=lifespan"
else:
new_args = "lifespan=lifespan"
content = ( content = (
content[: fastapi_match.start()] content[: fastapi_match.start()]
+ fastapi_match.group(1) + fastapi_match.group(1)
@@ -715,6 +742,8 @@ def patch_app_file(
print(f"✅ Would patch {path}") print(f"✅ Would patch {path}")
return True return True
# Sort imports only (avoid full formatting of user code)
content = ruff_format_content(content, path, mode="isort")
path.write_text(content, "UTF-8", newline="\n") path.write_text(content, "UTF-8", newline="\n")
print(f"✅ Patched {path}") print(f"✅ Patched {path}")
@@ -738,7 +767,7 @@ def patch_app_file(
def patch_vite_config( def patch_vite_config(
path: Path, path: Path,
module_name: str, *,
dry: bool = False, dry: bool = False,
) -> bool: ) -> bool:
"""Patch an existing vite.config.js/ts by adding fastapi-vue plugin. """Patch an existing vite.config.js/ts by adding fastapi-vue plugin.
@@ -769,13 +798,11 @@ def patch_vite_config(
# Insert after the last import line before non-import content # Insert after the last import line before non-import content
if not import_inserted: if not import_inserted:
stripped = line.strip() stripped = line.strip()
if stripped.startswith("import ") or stripped.startswith("from "): if stripped.startswith(("import ", "from ")) and i + 1 < len(lines):
# Check if next line is not an import
if i + 1 < len(lines):
next_stripped = lines[i + 1].strip() next_stripped = lines[i + 1].strip()
if not next_stripped.startswith( if not next_stripped.startswith("import ") and not next_stripped.startswith(
"import " "from "
) and not next_stripped.startswith("from "): ):
new_lines.append(import_line) new_lines.append(import_line)
import_inserted = True import_inserted = True
@@ -810,7 +837,7 @@ def patch_vite_config(
return True return True
def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool: def patch_frontend_health_check(frontend_dir: Path, *, dry: bool = False) -> bool:
"""Patch Vue app to include FastAPI backend health check. """Patch Vue app to include FastAPI backend health check.
Tries HelloWorld.vue first (full demo), then falls back to App.vue (minimal). Tries HelloWorld.vue first (full demo), then falls back to App.vue (minimal).
@@ -836,7 +863,7 @@ def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool:
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 default App.vue found to patch, not adding /api/health check")
return False return False
original_content = target_file.read_text("UTF-8") original_content = target_file.read_text("UTF-8")
@@ -852,9 +879,7 @@ def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool:
is_typescript = 'lang="ts"' in content is_typescript = 'lang="ts"' in content
# Build the script content based on JS/TS # Build the script content based on JS/TS
script_addition = ( script_addition = TS_HEALTH_CHECK_SCRIPT if is_typescript else JS_HEALTH_CHECK_SCRIPT
TS_HEALTH_CHECK_SCRIPT if is_typescript else JS_HEALTH_CHECK_SCRIPT
)
# Insert script addition before </script> # Insert script addition before </script>
script_end_match = re.search(r"</script>", content) script_end_match = re.search(r"</script>", content)
@@ -872,16 +897,14 @@ def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool:
# Insert before closing </h3> # Insert before closing </h3>
h3_close = content.find(" </h3>") h3_close = content.find(" </h3>")
if h3_close == -1: if h3_close == -1:
print( print(f"⚠️ Skipping {target_file} (no </h3> tag found for status insertion)")
f"⚠️ Skipping {target_file} (no </h3> tag found for status insertion)"
)
return False return False
before, after = content[:h3_close], content[h3_close:] before, after = content[:h3_close], content[h3_close:]
content = f"{before}{indent(STATUS_SPAN_TEMPLATE, ' ')}{after}" content = f"{before}{indent(STATUS_SPAN_TEMPLATE, ' ')}{after}"
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 Vue demo app)")
return False return False
# Insert before the </p> tag # Insert before the </p> tag
template_end = content.find("</template>") template_end = content.find("</template>")
@@ -912,12 +935,10 @@ def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool:
# SHA-256 of old vite-plugin-fastapi.js (before auto-upgrade marker was added) # 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 # with module name replaced by MODULE_NAME in the outDir path
_OLD_VITE_PLUGIN_SHA256 = ( _OLD_VITE_PLUGIN_SHA256 = "93713e879c15a25c750a70ce1de684adeaf11b0c723c38da56e5e7ba207f6632"
"93713e879c15a25c750a70ce1de684adeaf11b0c723c38da56e5e7ba207f6632"
)
def _upgrade_old_vite_plugin(path: Path, module_name: str, dry: bool = False) -> None: def _upgrade_old_vite_plugin(path: Path, module_name: str, *, dry: bool = False) -> None:
"""Remove old vite-plugin-fastapi.js that lacks auto-upgrade marker. """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 Old versions didn't have the upgrade marker, so write_file skips them as
@@ -929,7 +950,6 @@ def _upgrade_old_vite_plugin(path: Path, module_name: str, dry: bool = False) ->
content = path.read_text("UTF-8") content = path.read_text("UTF-8")
if UPGRADE_MARKER in content: if UPGRADE_MARKER in content:
return # Already new format, write_file handles it return # Already new format, write_file handles it
import hashlib
normalized = content.replace( normalized = content.replace(
f"../{module_name}/frontend-build", "../MODULE_NAME/frontend-build" f"../{module_name}/frontend-build", "../MODULE_NAME/frontend-build"
@@ -951,6 +971,7 @@ _new_files_written: list[tuple[Path, Path]] = []
def write_file( def write_file(
path: Path, path: Path,
content: str, content: str,
*,
overwrite: bool = True, overwrite: bool = True,
dry: bool = False, dry: bool = False,
executable: bool = False, executable: bool = False,
@@ -988,7 +1009,7 @@ def write_file(
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(
path, fallback_path, content, dry, executable path, fallback_path, content, dry=dry, executable=executable
) )
print(f"️ Skipping {path} (customized by user)") print(f"️ Skipping {path} (customized by user)")
return False return False
@@ -1011,6 +1032,7 @@ def _write_fallback_file(
original_path: Path, original_path: Path,
fallback_path: Path, fallback_path: Path,
content: str, content: str,
*,
dry: bool, dry: bool,
executable: bool, executable: bool,
) -> bool: ) -> bool:
@@ -1068,8 +1090,6 @@ def merge_pyproject(
if "requires-python" in data["project"]: if "requires-python" in data["project"]:
req = data["project"]["requires-python"] req = data["project"]["requires-python"]
# Parse minimum version from strings like ">=3.10" or ">=3.9,<4" # Parse minimum version from strings like ">=3.10" or ">=3.9,<4"
import re
match = re.search(r">=\s*(\d+)\.(\d+)", req) match = re.search(r">=\s*(\d+)\.(\d+)", req)
if match: if match:
major, minor = int(match.group(1)), int(match.group(2)) major, minor = int(match.group(1)), int(match.group(2))
@@ -1115,9 +1135,12 @@ def merge_pyproject(
if "custom" not in hatch_build["targets"]["sdist"]["hooks"]: if "custom" not in hatch_build["targets"]["sdist"]["hooks"]:
hatch_build["targets"]["sdist"]["hooks"]["custom"] = tomlkit.table() hatch_build["targets"]["sdist"]["hooks"]["custom"] = tomlkit.table()
if "path" not in hatch_build["targets"]["sdist"]["hooks"]["custom"]: if "path" not in hatch_build["targets"]["sdist"]["hooks"]["custom"]:
hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] = hatch_additions[ hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] = hatch_additions["targets"][
"targets" "sdist"
]["sdist"]["hooks"]["custom"]["path"] ]["hooks"]["custom"]["path"]
elif hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] == OLD_BUILD_HOOK_PATH:
# Migrate old build hook path to new name
hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] = NEW_BUILD_HOOK_PATH
return data return data
@@ -1161,7 +1184,7 @@ def find_js_runtime() -> tuple[str, str] | None:
return None return None
def ensure_python_project(project_dir: Path, dry: bool = False) -> bool: def ensure_python_project(project_dir: Path, *, dry: bool = False) -> bool:
"""Ensure pyproject.toml exists, run uv init if needed.""" """Ensure pyproject.toml exists, run uv init if needed."""
pyproject = project_dir / "pyproject.toml" pyproject = project_dir / "pyproject.toml"
if pyproject.exists(): if pyproject.exists():
@@ -1173,7 +1196,7 @@ def ensure_python_project(project_dir: Path, dry: bool = False) -> bool:
print("📦 No pyproject.toml found, initializing Python project...") print("📦 No pyproject.toml found, initializing Python project...")
print(">>> uv init") print(">>> uv init")
result = subprocess.run(["uv", "init", str(project_dir)], check=False) result = subprocess.run(["uv", "init", str(project_dir)], check=False) # noqa: S603, S607
if result.returncode != 0: if result.returncode != 0:
print("❌ uv init failed") print("❌ uv init failed")
return False return False
@@ -1187,7 +1210,7 @@ def ensure_python_project(project_dir: Path, dry: bool = False) -> bool:
return True return True
def ensure_frontend(project_dir: Path, dry: bool = False) -> bool: def ensure_frontend(project_dir: Path, *, dry: bool = False) -> bool:
"""Ensure frontend directory exists with a Vue project, run create-vue if needed.""" """Ensure frontend directory exists with a Vue project, run create-vue if needed."""
frontend_dir = project_dir / "frontend" frontend_dir = project_dir / "frontend"
package_json = frontend_dir / "package.json" package_json = frontend_dir / "package.json"
@@ -1219,11 +1242,7 @@ def ensure_frontend(project_dir: Path, dry: bool = False) -> bool:
print(f">>> {' '.join(create_cmd)}") print(f">>> {' '.join(create_cmd)}")
print("(Follow the prompts to configure your Vue app)") print("(Follow the prompts to configure your Vue app)")
print() print()
result = subprocess.run( result = subprocess.run(create_cmd, cwd=project_dir, check=False) # noqa: S603
create_cmd,
cwd=project_dir,
check=False,
)
if result.returncode != 0: if result.returncode != 0:
print("❌ create-vue failed") print("❌ create-vue failed")
return False return False
@@ -1245,10 +1264,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
project_path = Path(args.project_dir) project_path = Path(args.project_dir)
# Handle both "." and "/path/to/project" # Handle both "." and "/path/to/project"
if project_path.is_absolute(): project_dir = project_path if project_path.is_absolute() else Path.cwd() / project_path
project_dir = project_path
else:
project_dir = Path.cwd() / project_path
project_dir = project_dir.resolve() project_dir = project_dir.resolve()
dry = args.dry dry = args.dry
@@ -1268,11 +1284,11 @@ def cmd_setup(args: argparse.Namespace) -> int:
print(f"🔧 Setting up project: {project_dir}") print(f"🔧 Setting up project: {project_dir}")
# Step 1: Ensure frontend exists (do this first so cancellation doesn't leave partial setup) # Step 1: Ensure frontend exists (do this first so cancellation doesn't leave partial setup)
if not ensure_frontend(project_dir, dry): if not ensure_frontend(project_dir, dry=dry):
return 1 return 1
# Step 2: Ensure Python project exists # Step 2: Ensure Python project exists
if not ensure_python_project(project_dir, dry): if not ensure_python_project(project_dir, dry=dry):
return 1 return 1
# Detect module name # Detect module name
@@ -1284,13 +1300,28 @@ def cmd_setup(args: argparse.Namespace) -> int:
else: else:
print(f"📦 Module: {module_name}") print(f"📦 Module: {module_name}")
module_dir = project_dir / module_name
scripts_dir = project_dir / "scripts"
fastapi_vue_scripts = scripts_dir / "fastapi-vue"
# Check if project already has a CLI entrypoint in pyproject.toml
existing_cli_module = _find_existing_cli_module_path(project_dir, module_name)
# Determine main module path for DEVMODE import
main_module_path = existing_cli_module or f"{module_name}.__main__"
if existing_cli_module:
print(f"️ Using existing CLI: {existing_cli_module}")
# Resolve the CLI main module to a file path
main = project_dir / Path(*main_module_path.split(".")).with_suffix(".py")
# Determine port configuration # Determine port configuration
# Priority: --ports argument > existing project values > defaults # Priority: --ports argument > existing project values > defaults
if args.ports: if args.ports:
default_port, vite_port, dev_port = parse_ports(args.ports) default_port, vite_port, dev_port = parse_ports(args.ports)
ports_note = "(--ports)" ports_note = "(--ports)"
else: else:
existing_ports = extract_existing_ports(project_dir) existing_ports = extract_existing_ports(project_dir, main)
if existing_ports: if existing_ports:
default_port, vite_port, dev_port = existing_ports default_port, vite_port, dev_port = existing_ports
ports_note = "(kept for upgrade)" ports_note = "(kept for upgrade)"
@@ -1298,27 +1329,34 @@ def cmd_setup(args: argparse.Namespace) -> int:
default_port, vite_port, dev_port = DEFAULT_PORTS default_port, vite_port, dev_port = DEFAULT_PORTS
ports_note = "(--ports to override)" ports_note = "(--ports to override)"
print( print(f"📡 Ports: default={default_port}, vite={vite_port}, dev={dev_port} {ports_note}")
f"📡 Ports: default={default_port}, vite={vite_port}, dev={dev_port} {ports_note}"
) # Determine health path configuration
# Priority: --health argument > existing project value > default
if args.health is not None:
health = args.health
health_note = "(--health)" if health else "(disabled via --health)"
else:
existing_health = extract_existing_health(project_dir)
if existing_health is not _HEALTH_NOT_FOUND:
# Explicitly configured (path string, possibly empty to disable)
health = existing_health
health_note = "(kept for upgrade)"
else:
# Not found - use default
health = DEFAULT_HEALTH
health_note = "(--health to override)"
if health:
print(f"🏥 Health check: {health} {health_note}")
else:
print(f"🏥 Health check: disabled {health_note}")
# Title for templates # Title for templates
project_title = module_name.replace("_", " ").title() project_title = module_name.replace("_", " ").title()
module_dir = project_dir / module_name
scripts_dir = project_dir / "scripts"
fastapi_vue_scripts = scripts_dir / "fastapi-vue"
# Find existing FastAPI app # Find existing FastAPI app
app_info = ( app_info = find_fastapi_app(module_dir, project_dir) if module_dir.exists() else None
find_fastapi_app(module_dir, project_dir) if module_dir.exists() else None
)
# Check if project already has a CLI entrypoint in pyproject.toml
existing_cli_module = _find_existing_cli_module_path(project_dir, module_name)
# Determine main module path for DEVMODE import
main_module_path = existing_cli_module or f"{module_name}.__main__"
# Template variables # Template variables
tpl_vars = { tpl_vars = {
@@ -1327,6 +1365,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
"TEMPLATE_DEFAULT_PORT": str(default_port), "TEMPLATE_DEFAULT_PORT": str(default_port),
"TEMPLATE_VITE_PORT": str(vite_port), "TEMPLATE_VITE_PORT": str(vite_port),
"TEMPLATE_DEV_PORT": str(dev_port), "TEMPLATE_DEV_PORT": str(dev_port),
"TEMPLATE_HEALTH": f'"{health}"',
"ENVPREFIX": module_name.upper(), "ENVPREFIX": module_name.upper(),
"PROJECT_CLI": module_name, "PROJECT_CLI": module_name,
"MAIN_MODULE": main_module_path, "MAIN_MODULE": main_module_path,
@@ -1336,13 +1375,15 @@ def cmd_setup(args: argparse.Namespace) -> int:
app_file, app_var = app_info app_file, app_var = app_info
print(f"📍 Found FastAPI app: {app_var} in {app_file.name}") print(f"📍 Found FastAPI app: {app_var} in {app_file.name}")
tpl_vars["APP_VAR"] = app_var tpl_vars["APP_VAR"] = app_var
tpl_vars["APP_MODULE"] = app_file.stem # Dotted module path relative to project dir (e.g. "paskia.fastapi.mainapp")
app_module = ".".join(app_file.relative_to(project_dir).with_suffix("").parts)
tpl_vars["APP_MODULE"] = app_module
else: else:
print("📍 No existing FastAPI app found, will create new one") print("📍 No existing FastAPI app found, will create new one")
app_file = None app_file = None
app_var = "app" app_var = "app"
tpl_vars["APP_VAR"] = app_var tpl_vars["APP_VAR"] = app_var
tpl_vars["APP_MODULE"] = "app" tpl_vars["APP_MODULE"] = f"{module_name}.app"
# Create directories # Create directories
if not dry: if not dry:
@@ -1363,9 +1404,27 @@ def cmd_setup(args: argparse.Namespace) -> int:
obsolete_util.unlink() obsolete_util.unlink()
print(f"🗑️ Removed obsolete {obsolete_util}") print(f"🗑️ Removed obsolete {obsolete_util}")
# Remove obsolete build-frontend.py if present (renamed to buildhook.py)
obsolete_build_hook = fastapi_vue_scripts / "build-frontend.py"
if obsolete_build_hook.exists():
if dry:
print(f"🗑️ Would remove obsolete {obsolete_build_hook}")
else:
obsolete_build_hook.unlink()
print(f"🗑️ Removed obsolete {obsolete_build_hook}")
# Remove obsolete __init__.py if present (folder is no longer a module)
obsolete_init = fastapi_vue_scripts / "__init__.py"
if obsolete_init.exists():
if dry:
print(f"🗑️ Would remove obsolete {obsolete_init}")
else:
obsolete_init.unlink()
print(f"🗑️ Removed obsolete {obsolete_init}")
# Copy all files from the template's fastapi-vue folder # Copy all files from the template's fastapi-vue folder
template_fastapi_vue_dir = TEMPLATE_DIR / "scripts" / "fastapi-vue" template_fastapi_vue_dir = TEMPLATE_DIR / "scripts" / "fastapi-vue"
for template_file in template_fastapi_vue_dir.iterdir(): for template_file in sorted(template_fastapi_vue_dir.iterdir()):
if template_file.is_file(): if template_file.is_file():
dest_path = fastapi_vue_scripts / template_file.name dest_path = fastapi_vue_scripts / template_file.name
template = template_file.read_text("UTF-8") template = template_file.read_text("UTF-8")
@@ -1414,11 +1473,8 @@ def cmd_setup(args: argparse.Namespace) -> int:
# === Handle __main__.py === # === Handle __main__.py ===
if existing_cli_module: if existing_cli_module:
# Existing CLI entrypoint - write .new.py beside the existing main module # Existing CLI entrypoint - write .new.py beside the existing main module
main_module_parts = existing_cli_module.split(".") main_fallback = main.with_suffix(".new.py")
main_module_file = project_dir / ("/".join(main_module_parts) + ".py")
main_fallback = main_module_file.with_suffix(".new.py")
else: else:
main_module_file = None
main_fallback = module_dir / "__main__.new.py" main_fallback = module_dir / "__main__.new.py"
main_file = module_dir / "__main__.py" main_file = module_dir / "__main__.py"
@@ -1445,28 +1501,27 @@ def cmd_setup(args: argparse.Namespace) -> int:
else: else:
# Existing CLI entrypoint: write our template as .new.py beside the existing module # Existing CLI entrypoint: write our template as .new.py beside the existing module
# and also patch the existing module with DEVMODE if needed # and also patch the existing module with DEVMODE if needed
print(f"️ Using existing CLI: {existing_cli_module}")
_write_fallback_file( _write_fallback_file(
main_module_file or main_file, main,
main_fallback, main_fallback,
main_content, main_content,
dry=dry, dry=dry,
executable=False, executable=False,
) )
if main_module_file and main_module_file.exists(): if main.exists():
content = main_module_file.read_text("UTF-8") content = main.read_text("UTF-8")
if "DEVMODE" not in content: if "DEVMODE" not in content:
new_content = _add_devmode_to_main(content) new_content = _add_devmode_to_main(content)
new_file = main_module_file.with_suffix(".new.py") new_file = main.with_suffix(".new.py")
_write_fallback_file( _write_fallback_file(
main_module_file, main,
new_file, new_file,
new_content, new_content,
dry=dry, dry=dry,
executable=False, executable=False,
) )
elif main_module_file: else:
print(f"⚠️ Existing CLI main module {main_module_file} not found") print(f"⚠️ Existing CLI main module {main} not found")
# === Update vite.config.js/ts === # === Update vite.config.js/ts ===
frontend_dir = project_dir / "frontend" frontend_dir = project_dir / "frontend"
@@ -1474,7 +1529,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
# 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 plugin versions that lack the auto-upgrade marker
_upgrade_old_vite_plugin(plugin_file, module_name, dry) _upgrade_old_vite_plugin(plugin_file, module_name, dry=dry)
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=dry) write_file(plugin_file, content, overwrite=True, dry=dry)
@@ -1484,15 +1539,15 @@ def cmd_setup(args: argparse.Namespace) -> int:
vite_config_js = frontend_dir / "vite.config.js" vite_config_js = frontend_dir / "vite.config.js"
if vite_config_ts.exists(): if vite_config_ts.exists():
patch_vite_config(vite_config_ts, module_name, dry) patch_vite_config(vite_config_ts, dry=dry)
elif vite_config_js.exists(): elif vite_config_js.exists():
patch_vite_config(vite_config_js, module_name, dry) patch_vite_config(vite_config_js, dry=dry)
else: else:
print("⚠️ No vite.config.ts or vite.config.js found in frontend/") print("⚠️ No vite.config.ts or vite.config.js found in frontend/")
print(" Run create-vue first to generate a Vite config to patch.") print(" Run create-vue first to generate a Vite config to patch.")
# Patch Vue app with backend health check # Patch Vue app with backend health check
patch_frontend_health_check(frontend_dir, dry) patch_frontend_health_check(frontend_dir, dry=dry)
# === Update pyproject.toml === # === Update pyproject.toml ===
pyproject_path = project_dir / "pyproject.toml" pyproject_path = project_dir / "pyproject.toml"
@@ -1531,9 +1586,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
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
gitignore_path.write_bytes( gitignore_path.write_bytes(gitignore_content + suffix + gitignore_entry.encode() + nl)
gitignore_content + suffix + gitignore_entry.encode() + nl
)
print(f"✅ Added {gitignore_entry} to .gitignore") print(f"✅ Added {gitignore_entry} to .gitignore")
elif dry: elif dry:
print(f"✅ Would create .gitignore with {gitignore_entry}") print(f"✅ Would create .gitignore with {gitignore_entry}")
@@ -1553,20 +1606,18 @@ def cmd_setup(args: argparse.Namespace) -> int:
print_boxed("Setup complete!") print_boxed("Setup complete!")
# Show cd command only if project is not in current directory # Show cd command only if project is not in current directory
cd_cmd = "" if project_dir == Path.cwd() else f"cd {project_dir}; " cd_cmd = "" if project_dir == Path.cwd() else f"cd {project_dir} && "
script_name = module_name.replace("_", "-") script_name = module_name.replace("_", "-")
message = SETUP_COMPLETE_MESSAGE.replace("CD_CMD", cd_cmd).replace( message = SETUP_COMPLETE_MESSAGE.replace("CD_CMD", cd_cmd).replace("SCRIPT_NAME", script_name)
"SCRIPT_NAME", script_name if platform.system() == "Windows":
) message = message.replace(" && ", "; ")
print(message) print(message)
# Show merge note if any .new.py files were written # Show merge note if any .new.py files were written
if _new_files_written: if _new_files_written:
print() print()
print( print("⚠️ Some files could not be auto-upgraded because you customized them.")
"⚠️ Note: Some files could not be auto-upgraded because you customized them."
)
print(" Please manually merge the following files:") print(" Please manually merge the following files:")
for new_file, original_file in _new_files_written: for new_file, original_file in _new_files_written:
try: try:
@@ -1585,25 +1636,19 @@ def cmd_setup(args: argparse.Namespace) -> int:
def is_uninitialized_folder(path: Path) -> bool: def is_uninitialized_folder(path: Path) -> bool:
"""Check if a folder appears to be completely uninitialized.""" """Check if a folder appears to be completely uninitialized."""
return ( return not (path / "pyproject.toml").exists() and not (path / "package.json").exists()
not (path / "pyproject.toml").exists() and not (path / "package.json").exists()
)
def is_already_patched(path: Path) -> bool: def is_already_patched(path: Path) -> bool:
"""Check if a folder has already been patched by fastapi-vue-setup.""" """Check if a folder has already been patched by fastapi-vue-setup."""
# Check for our scripts directory # Check for our scripts directory for vite plugin in frontend
if (path / "scripts" / "fastapi-vue").exists(): scriptdir = path / "scripts" / "fastapi-vue"
return True viteplugin = path / "frontend" / "vite-plugin-fastapi.js"
return scriptdir.exists() or viteplugin.exists()
# Check for vite plugin in frontend
if (path / "frontend" / "vite-plugin-fastapi.js").exists():
return True
return False
def main() -> int: def main() -> int:
"""CLI entry point."""
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description=f"fastapi-vue-setup {version} - FastAPI + Vue project setup tool", description=f"fastapi-vue-setup {version} - FastAPI + Vue project setup tool",
formatter_class=argparse.RawDescriptionHelpFormatter, formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -1629,8 +1674,11 @@ Examples:
help="Port configuration as comma-separated values (default: 3100,3100,3200)", help="Port configuration as comma-separated values (default: 3100,3100,3200)",
) )
parser.add_argument( parser.add_argument(
"--dry", "--dry-run", action="store_true", help="Show what would be done" "--health",
metavar="PATH",
help='Health check endpoint (disable waiting for backend startup by setting "")',
) )
parser.add_argument("--dry", "--dry-run", action="store_true", help="Show what would be done")
args = parser.parse_args() args = parser.parse_args()
+17
View File
@@ -31,3 +31,20 @@ dev = ["ruff", "fastapi-vue"]
[tool.uv.sources] [tool.uv.sources]
fastapi-vue = { path = "fastapi-vue", editable = true } fastapi-vue = { path = "fastapi-vue", editable = true }
[tool.uv.workspace]
members = [
"f",
]
[tool.ruff]
line-length = 100
[tool.ruff.lint]
select = ["ALL"]
ignore = ["D203", "D213", "COM812"] # Conflicting with D211, D212 and formatting
[tool.ruff.lint.per-file-ignores]
"template/**" = ["F821"] # Undefined names are template placeholders
"template/scripts/devserver.py" = ["N806"] # MODULE_NAME is a template variable
"fastapi_vue_setup.py" = ["PLR", "C901", "T201", "RUF001"]
+4 -11
View File
@@ -10,25 +10,18 @@ DIST = ROOT / "dist"
FASTAPI_VUE = ROOT / "fastapi-vue" FASTAPI_VUE = ROOT / "fastapi-vue"
def main(): def main() -> None:
"""Build both packages to dist directory."""
# Clear the dist directory # Clear the dist directory
if DIST.exists(): if DIST.exists():
shutil.rmtree(DIST) shutil.rmtree(DIST)
DIST.mkdir() DIST.mkdir()
# Build fastapi-vue (subdirectory) to root dist # Build fastapi-vue (subdirectory) to root dist
subprocess.run( subprocess.run(["uv", "build", "--out-dir", str(DIST)], cwd=FASTAPI_VUE, check=True) # noqa: S603, S607
["uv", "build", "--out-dir", str(DIST)],
cwd=FASTAPI_VUE,
check=True,
)
# Build fastapi-vue-setup (root) # Build fastapi-vue-setup (root)
subprocess.run( subprocess.run(["uv", "build", "--out-dir", str(DIST)], cwd=ROOT, check=True) # noqa: S603, S607
["uv", "build", "--out-dir", str(DIST)],
cwd=ROOT,
check=True,
)
if __name__ == "__main__": if __name__ == "__main__":
+1
View File
@@ -0,0 +1 @@
"""Backend package with FastAPI application and Vue frontend integration."""
+7 -4
View File
@@ -1,4 +1,6 @@
# auto-upgrade@fastapi-vue-setup - remove this if you modify this file # auto-upgrade@fastapi-vue-setup - remove this if you modify this file
"""Command-line entry point for running the backend server."""
import argparse import argparse
import os import os
@@ -8,7 +10,8 @@ DEFAULT_PORT = TEMPLATE_DEFAULT_PORT
DEVMODE = os.getenv("ENVPREFIX_DEV") == "1" DEVMODE = os.getenv("ENVPREFIX_DEV") == "1"
def main(): def main() -> None:
"""Run the backend server with optional arguments."""
parser = argparse.ArgumentParser(description="Run the MODULE_NAME server.") parser = argparse.ArgumentParser(description="Run the MODULE_NAME server.")
parser.add_argument( parser.add_argument(
"-l", "-l",
@@ -17,12 +20,12 @@ def main():
help=(f"Endpoint (default: localhost:{DEFAULT_PORT})."), help=(f"Endpoint (default: localhost:{DEFAULT_PORT})."),
) )
args = parser.parse_args() args = parser.parse_args()
dev = {"reload": True, "reload_dirs": ["MODULE_NAME"]} dev = {"reload": True, "reload_dirs": ["paskia"]} if DEVMODE else {}
server.run( server.run(
"MODULE_NAME.APP_MODULE:APP_VAR", "APP_MODULE:APP_VAR",
listen=args.listen, listen=args.listen,
default_port=DEFAULT_PORT, default_port=DEFAULT_PORT,
**(dev if DEVMODE else {}), **dev,
) )
+6 -2
View File
@@ -1,3 +1,6 @@
"""FastAPI application module with Vue frontend integration."""
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from pathlib import Path from pathlib import Path
@@ -10,7 +13,7 @@ frontend = Frontend(Path(__file__).with_name("frontend-build"))
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI): async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
"""Manage app startup and shutdown resources.""" """Manage app startup and shutdown resources."""
await frontend.load() await frontend.load()
yield yield
@@ -24,7 +27,8 @@ app = FastAPI(title="PROJECT_TITLE", debug=DEVMODE, lifespan=lifespan)
# Health check endpoint for the Vue demo app to verify the backend is running # Health check endpoint for the Vue demo app to verify the backend is running
@app.get("/api/health") @app.get("/api/health")
async def health_check(): async def health_check() -> dict[str, str]:
"""Return backend status for health monitoring."""
return {"status": "ok"} return {"status": "ok"}
+12 -7
View File
@@ -11,7 +11,7 @@ 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)
sys.path.insert(0, str(Path(__file__).with_name("fastapi-vue"))) sys.path.insert(0, str(Path(__file__).with_name("fastapi-vue")))
from devutil import ( # type: ignore from devutil import (
ProcessGroup, ProcessGroup,
check_ports_free, check_ports_free,
logger, logger,
@@ -22,11 +22,15 @@ from devutil import ( # type: ignore
DEFAULT_VITE_PORT = TEMPLATE_VITE_PORT DEFAULT_VITE_PORT = TEMPLATE_VITE_PORT
DEFAULT_DEV_PORT = TEMPLATE_DEV_PORT DEFAULT_DEV_PORT = TEMPLATE_DEV_PORT
HEALTH = TEMPLATE_HEALTH
async def run_devserver( async def run_devserver(
listen: str, backend: str, extra_args: list[str] | None = None listen: str,
backend: str,
extra_args: list[str] | None = None,
) -> None: ) -> None:
"""Start Vite and FastAPI dev servers with hot reload."""
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():
@@ -45,11 +49,12 @@ async def run_devserver(
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(*MODULE_NAME, *(extra_args or [])) await pg.spawn(*MODULE_NAME, *(extra_args or []))
await pg.wait(npm_i, ready(backurl, path="/api/health?from=devserver.py")) await pg.wait(npm_i, ready(backurl, path=HEALTH))
await pg.spawn(*vite, cwd=front) await pg.spawn(*vite, cwd=front)
def main(): def main() -> None:
"""Parse CLI arguments and run the devserver."""
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,
@@ -58,12 +63,12 @@ def main():
parser.add_argument( parser.add_argument(
"-l", "-l",
"--listen", "--listen",
metavar="host:port", metavar="addr",
help=f"Vite (default: localhost:{DEFAULT_VITE_PORT})", help=f"Vite (default: localhost:{DEFAULT_VITE_PORT})",
) )
parser.add_argument( parser.add_argument(
"--backend", "--backend",
metavar="host:port", metavar="addr",
help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})", help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})",
) )
args, extra_args = parser.parse_known_args() args, extra_args = parser.parse_known_args()
@@ -72,7 +77,7 @@ def main():
HELP_EPILOG = """ HELP_EPILOG = """
scripts/devserver.py [args to PROJECT_CLI] Other options are forwarded to PROJECT_CLI [args]
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). npm, deno, bun, or full path to the runtime executable (node maps to npm).
@@ -1,15 +1,19 @@
# ruff: noqa: INP001
"""Hatch build hook for building Vue frontend during package build.""" """Hatch build hook for building Vue frontend during package build."""
import sys import sys
from pathlib import Path from pathlib import Path
from hatchling.builders.hooks.plugin.interface import BuildHookInterface # type: ignore from hatchling.builders.hooks.plugin.interface import BuildHookInterface
sys.path.insert(0, str(Path(__file__).parent)) sys.path.insert(0, str(Path(__file__).parent))
from buildutil import build from buildutil import build
class CustomBuildHook(BuildHookInterface): class CustomBuildHook(BuildHookInterface): # type: ignore[misc]
def initialize(self, version, build_data): """Hatch build hook that builds Vue frontend during package build."""
def initialize(self, version: str, build_data: dict) -> None: # type: ignore[override]
"""Build frontend before package is built."""
super().initialize(version, build_data) super().initialize(version, build_data)
build("frontend") build("frontend")
+98 -58
View File
@@ -1,3 +1,4 @@
# ruff: noqa: INP001
"""Utilities used at build time and in devserver script. No dependencies.""" """Utilities used at build time and in devserver script. No dependencies."""
import logging import logging
@@ -7,6 +8,8 @@ import shutil
import subprocess import subprocess
from pathlib import Path from pathlib import Path
MIN_NODE_VERSION = 20
class _PrefixFormatter(logging.Formatter): class _PrefixFormatter(logging.Formatter):
"""Formatter that adds prefix based on log level.""" """Formatter that adds prefix based on log level."""
@@ -30,82 +33,119 @@ def _check_node_version(node_path: str) -> None:
Raises RuntimeError if version is too old or cannot be determined. Raises RuntimeError if version is too old or cannot be determined.
""" """
try: try:
result = subprocess.run( result = subprocess.run( # noqa: S603
[node_path, "--version"], capture_output=True, text=True, check=True [node_path, "--version"],
capture_output=True,
text=True,
check=True,
) )
version_str = result.stdout.strip() version_str = result.stdout.strip()
# Parse version like "v20.10.0" or "v18.17.1" # Parse version like "v20.10.0" or "v18.17.1"
match = re.match(r"v(\d+)", version_str) match = re.match(r"v(\d+)", version_str)
if match: if match:
major_version = int(match.group(1)) major_version = int(match.group(1))
if major_version >= 20: if major_version >= MIN_NODE_VERSION:
return return
raise RuntimeError( msg = f"Node.js {version_str} found, but v{MIN_NODE_VERSION}+ required"
f"Node.js {version_str} found, but v20+ required (install with nvm)" raise RuntimeError(msg)
)
except (subprocess.CalledProcessError, FileNotFoundError, ValueError): except (subprocess.CalledProcessError, FileNotFoundError, ValueError):
pass pass
raise RuntimeError("Could not determine Node.js version") msg = "Could not determine Node.js version"
raise RuntimeError(msg)
def _validate_npm_runtime(tool: str) -> bool:
"""Validate npm runtime by checking Node.js version. Returns True if valid."""
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path is None:
return False
try:
_check_node_version(node_path)
except RuntimeError:
return False
return True
def _find_runtime_from_env(options: list[str]) -> tuple[str, str] | None:
"""Find runtime specified by JS_RUNTIME environment variable."""
js_runtime_env = os.environ.get("JS_RUNTIME")
if not js_runtime_env:
return None
js_runtime = js_runtime_env
js_path = Path(js_runtime)
runtime_name = js_path.name
# Map node to npm
if runtime_name == "node":
runtime_name = "npm"
js_runtime = str(js_path.parent / "npm") if js_path.parent.name else "npm"
for option in options:
if option != runtime_name and not runtime_name.startswith(option):
continue
tool = shutil.which(js_runtime)
if tool is None:
msg = f"JS_RUNTIME={js_runtime_env}: {option} not found"
raise RuntimeError(msg)
if option == "npm":
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path is None:
msg = f"JS_RUNTIME={js_runtime_env}: node not found"
raise RuntimeError(msg)
_check_node_version(node_path)
return tool, option
msg = f"JS_RUNTIME={js_runtime_env} not recognized"
raise RuntimeError(msg)
def _auto_detect_runtime(options: list[str]) -> tuple[str, str]:
"""Auto-detect JavaScript runtime from available options."""
node_version_error: RuntimeError | None = None
for option in options:
tool = shutil.which(option)
if not tool:
continue
if option == "npm" and not _validate_npm_runtime(tool):
try:
node_path = shutil.which("node", path=str(Path(tool).parent))
if node_path:
_check_node_version(node_path)
except RuntimeError as e:
node_version_error = e
continue
return tool, option
if node_version_error:
raise node_version_error
msg = "Node.js (v20+), Deno or Bun is required but none was found"
raise RuntimeError(msg)
def find_js_runtime() -> tuple[str, str]: def find_js_runtime() -> tuple[str, str]:
"""Find a JavaScript runtime from JS_RUNTIME env or auto-detect. """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 (tool_path, tool_name) where tool_name is "deno", "npm", or "bun".
Raises JSRuntimeError if no suitable runtime is found. Raises RuntimeError if no suitable runtime is found.
""" """
options = ["npm", "deno", "bun"] options = ["npm", "deno", "bun"]
node_version_error: RuntimeError | None = None
# Check for JS_RUNTIME environment variable # Check for JS_RUNTIME environment variable
if js_runtime_env := os.environ.get("JS_RUNTIME"): if result := _find_runtime_from_env(options):
js_runtime = js_runtime_env return result
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 # Auto-detect
for option in options: return _auto_detect_runtime(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(): def find_build_tool() -> tuple[list[str], list[str]]:
"""Find JavaScript runtime and construct install/build commands. """Find JavaScript runtime and construct install/build commands.
Returns (install_cmd, build_cmd) tuples of command lists. Returns (install_cmd, build_cmd) tuples of command lists.
@@ -143,7 +183,7 @@ def find_dev_tool() -> list[str]:
if name == "bun": if name == "bun":
logger.warning( logger.warning(
"Bun has a bug in WS proxying (https://github.com/oven-sh/bun/issues/9882). Consider using npm instead." "Bun has a WS proxy bug (github.com/oven-sh/bun/issues/9882). Consider npm.",
) )
return [tool, *dev_args[name]] return [tool, *dev_args[name]]
@@ -176,16 +216,16 @@ def build(folder: str = "frontend") -> None:
install_cmd, build_cmd = find_build_tool() install_cmd, build_cmd = find_build_tool()
except RuntimeError as e: except RuntimeError as e:
logger.warning(e) logger.warning(e)
raise SystemExit(1) raise SystemExit(1) from None
def run(cmd): def run(cmd: list[str]) -> None:
display_cmd = [Path(cmd[0]).stem, *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) # noqa: S603
try: try:
run(install_cmd) run(install_cmd)
logger.info("") logger.info("")
run(build_cmd) run(build_cmd)
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
raise SystemExit(1) raise SystemExit(1) from None
+36 -22
View File
@@ -1,27 +1,33 @@
# ruff: noqa: INP001
"""Utilities meant for devserver script, used only in source repository with dev deps.""" """Utilities meant for devserver script, used only in source repository with dev deps."""
import asyncio import asyncio
import subprocess import subprocess
import sys import sys
from collections.abc import Coroutine
from contextlib import suppress from contextlib import suppress
from pathlib import Path from pathlib import Path
from typing import Any from typing import TYPE_CHECKING, Any, Self
import httpx 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
if TYPE_CHECKING:
from collections.abc import Coroutine
class ProcessGroup: class ProcessGroup:
"""Manage async subprocesses with automatic cleanup, like TaskGroup for processes.""" """Manage async subprocesses with automatic cleanup, like TaskGroup for processes."""
def __init__(self): def __init__(self) -> None:
"""Initialize empty process tracking."""
self._procs: list[asyncio.subprocess.Process] = [] self._procs: list[asyncio.subprocess.Process] = []
self._cmds: dict[int, str] = {} # pid -> command name self._cmds: dict[int, str] = {} # pid -> command name
async def spawn( async def spawn(
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]).stem cmd_name = Path(cmd[0]).stem
@@ -32,7 +38,8 @@ class ProcessGroup:
return proc return proc
async def wait( async def wait(
self, *waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]" self,
*waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]",
) -> None: ) -> None:
"""Wait for processes/coroutines to complete, raise SystemExit on failure.""" """Wait for processes/coroutines to complete, raise SystemExit on failure."""
@@ -43,8 +50,7 @@ class ProcessGroup:
raise subprocess.CalledProcessError(returncode, cmd_name) raise subprocess.CalledProcessError(returncode, cmd_name)
tasks = [ tasks = [
wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w for w in waitables
for w in waitables
] ]
try: try:
await asyncio.gather(*tasks) await asyncio.gather(*tasks)
@@ -52,14 +58,15 @@ class ProcessGroup:
logger.warning("%s failed with exit status %d", e.cmd, e.returncode) logger.warning("%s failed with exit status %d", e.cmd, e.returncode)
raise SystemExit(1) from None raise SystemExit(1) from None
async def __aenter__(self): async def __aenter__(self) -> Self:
"""Enter the async context manager."""
return self return self
async def __aexit__(self, exc_type, *_): async def __aexit__(self, exc_type: type[BaseException] | None, *_: object) -> None:
"""Wait for one process to exit, terminate others, then wait for all.""" """Wait for one process to exit, terminate others, then wait for all."""
await self._cleanup(immediate=exc_type is not None) await self._cleanup(immediate=exc_type is not None)
async def _cleanup(self, immediate: bool = False): async def _cleanup(self, *, immediate: bool = False) -> None:
running = [p for p in self._procs if p.returncode is None] running = [p for p in self._procs if p.returncode is None]
if not running: if not running:
return return
@@ -87,7 +94,7 @@ class ProcessGroup:
asyncio.wait_for( asyncio.wait_for(
asyncio.gather(*[p.wait() for p in still_running]), asyncio.gather(*[p.wait() for p in still_running]),
timeout=10, timeout=10,
) ),
) )
except TimeoutError: except TimeoutError:
for p in self._procs: for p in self._procs:
@@ -111,29 +118,32 @@ async def check_ports_free(*urls: str) -> None:
await asyncio.gather(*[check(client, url) for url in urls]) await asyncio.gather(*[check(client, url) for url in urls])
async def ready(url: str, path: str = "") -> None: async def ready(url: str, path: str = "", max_attempts: int = 50) -> None:
"""Wait for the server to be ready by polling an endpoint. """Wait for the server to be ready by polling an endpoint.
Use empty path to disable the check and make this return immediately.
Raises SystemExit(1) if server doesn't start in time. Raises SystemExit(1) if server doesn't start in time.
""" """
max_attempts = 50 if not path:
full_url = f"{url}{path}" return
async with httpx.AsyncClient() as client: async with httpx.AsyncClient() as client:
for attempt in range(max_attempts): for attempt in range(max_attempts):
try: try:
await client.get(full_url, timeout=1.0) await client.get(f"{url}{path}", timeout=1.0)
logger.info("✓ Backend ready!")
return
except httpx.RequestError: except httpx.RequestError:
if attempt == max_attempts - 1: if attempt == max_attempts - 1:
logger.warning("Backend didn't start in time") logger.warning("Backend didn't start in time")
raise SystemExit(1) raise SystemExit(1) from None
await asyncio.sleep(0.1) await asyncio.sleep(0.1)
else:
logger.info("✓ Backend ready!")
return
def setup_vite( def setup_vite(
endpoint: str, default_port: int = 5173 endpoint: str,
default_port: int = 5173,
) -> tuple[str, list[str], list[str]]: ) -> tuple[str, list[str], list[str]]:
"""Parse frontend endpoint and build commands. """Parse frontend endpoint and build commands.
@@ -159,7 +169,9 @@ def setup_vite(
def setup_fastapi( def setup_fastapi(
endpoint: str, module: str, default_port: int = 8000 endpoint: str,
module: str,
default_port: int = 8000,
) -> tuple[str, list[str]]: ) -> tuple[str, list[str]]:
"""Parse backend endpoint and build uvicorn command. """Parse backend endpoint and build uvicorn command.
@@ -174,7 +186,7 @@ def setup_fastapi(
host = endpoints[0]["host"] host = endpoints[0]["host"]
port = endpoints[0]["port"] port = endpoints[0]["port"]
reload_dir = module.split(".")[0] # Don't reload on frontend changes reload_dir = module.split(".", maxsplit=1)[0] # Don't reload on frontend changes
cmd = [ cmd = [
sys.executable, sys.executable,
@@ -191,7 +203,9 @@ def setup_fastapi(
def setup_cli( def setup_cli(
cli: str, endpoint: str, default_port: int = 8000 cli: str,
endpoint: str,
default_port: int = 8000,
) -> tuple[str, list[str]]: ) -> tuple[str, list[str]]:
"""Parse backend endpoint and build CLI command. """Parse backend endpoint and build CLI command.