diff --git a/fastapi-vue/fastapi_vue/__init__.py b/fastapi-vue/fastapi_vue/__init__.py index 5409b89..fbbb0e5 100644 --- a/fastapi-vue/fastapi_vue/__init__.py +++ b/fastapi-vue/fastapi_vue/__init__.py @@ -1,3 +1,5 @@ +"""FastAPI Vue integration - serve Vue frontend from FastAPI.""" + from .staticfiles import Frontend __all__ = ["Frontend"] diff --git a/fastapi-vue/fastapi_vue/hostutil.py b/fastapi-vue/fastapi_vue/hostutil.py index 50bee9b..9364788 100644 --- a/fastapi-vue/fastapi_vue/hostutil.py +++ b/fastapi-vue/fastapi_vue/hostutil.py @@ -1,8 +1,60 @@ +"""Parse endpoint strings for uvicorn server configuration.""" + import contextlib import ipaddress from urllib.parse import urlparse +def _parse_all_interfaces(value: str) -> list[dict] | None: + """Parse ':port' format to bind all interfaces.""" + if not (value.startswith(":") and value != ":"): + return None + port_part = value[1:] + if not port_part.isdigit(): + msg = f"Invalid port in '{value}'" + raise SystemExit(msg) + port = int(port_part) + return [{"host": "0.0.0.0", "port": port}, {"host": "::", "port": port}] # noqa: S104 + + +def _parse_unix_socket(value: str) -> list[dict] | None: + """Parse UNIX domain socket paths.""" + if value.startswith("/"): + return [{"uds": value}] + if value.startswith("unix:"): + uds_path = value[5:] or None + if uds_path is None: + msg = "unix: path must not be empty" + raise SystemExit(msg) + return [{"uds": uds_path}] + return None + + +def _parse_unbracketed_ipv6(value: str, default_port: int) -> list[dict] | None: + """Parse unbracketed IPv6 addresses.""" + if value.count(":") <= 1 or value.startswith("["): + return None + try: + ipaddress.IPv6Address(value) + except ValueError as e: + msg = f"Invalid IPv6 address '{value}': {e}" + raise SystemExit(msg) from e + return [{"host": value, "port": default_port}] + + +def _parse_host_port(value: str, default_port: int) -> list[dict]: + """Parse host[:port] or [ipv6][:port] using urlparse.""" + parsed = urlparse(f"//{value}") # // prefix lets urlparse treat it as netloc + host = parsed.hostname or "localhost" + port = parsed.port or default_port + + # Validate IP literals (optional; hostname passes through) + with contextlib.suppress(ValueError): + ipaddress.ip_address(host) + + return [{"host": host, "port": port}] + + def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]: """Parse an endpoint string into uvicorn bind configurations. @@ -23,6 +75,7 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]: - [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}] @@ -31,51 +84,33 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]: 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 + # Try specialized parsers in order + result = _parse_all_interfaces(value) + if result is not None: + return result - # 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}] + result = _parse_unix_socket(value) + if result is not None: + return result - # 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}] + result = _parse_unbracketed_ipv6(value, default_port) + if result is not None: + return result - # 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}] + # Fallback: host[:port], [ipv6][:port] + return _parse_host_port(value, default_port) def parse_endpoints( - listen: str | list[str] | None = None, default_port: int = 8000 + 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}"] diff --git a/fastapi-vue/fastapi_vue/server.py b/fastapi-vue/fastapi_vue/server.py index dd7fae6..70c8443 100644 --- a/fastapi-vue/fastapi_vue/server.py +++ b/fastapi-vue/fastapi_vue/server.py @@ -1,7 +1,10 @@ +"""Uvicorn server runner with multi-endpoint support.""" + import asyncio import logging import os from contextlib import suppress +from typing import Any import uvicorn from uvicorn import Config, Server @@ -18,8 +21,8 @@ def run( default_port: int = 8000, reload: bool = False, workers: int | None = None, - **uvicorn_config, -): + **uvicorn_config: Any, # noqa: ANN401 +) -> None: """Run uvicorn server(s) for the given app. Args: @@ -29,10 +32,12 @@ def run( 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). + """ endpoints = parse_endpoints(listen, default_port) if not endpoints: - raise ValueError("No endpoints to serve; check listen configuration") + msg = "No endpoints to serve; check listen configuration" + raise ValueError(msg) conf: dict[str, object] = {"app": app, "reload": reload, "workers": workers} proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1") @@ -48,7 +53,7 @@ def run( 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.""" forbidden = {"reload", "workers"} & {k for k, v in kwargs.items() if v} if forbidden: @@ -59,13 +64,10 @@ async def serve(endpoints: list[dict], **kwargs) -> None: 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.""" if len(endpoints) > 1: - eps = [ - ep["uds"] if "uds" in ep else f"{ep['host']}:{ep['port']}" - for ep in endpoints - ] + 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], diff --git a/fastapi-vue/fastapi_vue/staticfiles.py b/fastapi-vue/fastapi_vue/staticfiles.py index 9290881..252f3c9 100644 --- a/fastapi-vue/fastapi_vue/staticfiles.py +++ b/fastapi-vue/fastapi_vue/staticfiles.py @@ -25,7 +25,7 @@ __all__ = ["Frontend"] class Assets: - """Default cached value to /assets/""" + """Default cached value to /assets/.""" @staticmethod def parse(cached: str | list[str] | Assets) -> list[str]: @@ -37,7 +37,8 @@ class Assets: case list(): return cached case _: - raise ValueError(f"Invalid cached value: {cached!r}") + msg = f"Invalid cached value: {cached!r}" + raise ValueError(msg) class Frontend: @@ -55,27 +56,29 @@ class Frontend: index: Name of the index file (default: "index.html") spa: Enable SPA mode - serve index.html for unknown routes (default: False) cached: Path prefixes that are immutable (default: "/assets/") - favicon: May use wildcards of full path. E.g. /assets/logo*.png matches logo.hash.png created by Vite + favicon: Wildcard path to favicon. E.g. /assets/logo*.png matches Vite output zstdlevel: Zstd compression level (default: 18) + """ - def __init__( + def __init__( # noqa: PLR0913 self, directory: Path | str, *, index: str = "index.html", spa: bool = False, catch_all: bool | None = None, - cached: str | list[str] | Assets = Assets(), + cached: str | list[str] | Assets | None = None, favicon: str | None = None, zstdlevel: int = 18, ) -> None: + """Initialize Frontend with given configuration.""" self.www: dict[str, tuple[bytes, bytes | None, dict]] = {} self.base: Path = Path(directory) self.index = index self.spa = spa self._catch_all = spa if catch_all is None else catch_all - self.cached_paths = Assets.parse(cached) + self.cached_paths = Assets.parse(cached if cached is not None else Assets()) self.zstdlevel = zstdlevel self.favicon = favicon self._app: FastAPI | None = None @@ -107,11 +110,12 @@ class Frontend: paths.add("/favicon.ico") return paths - def _load(self): + def _load(self) -> dict[str, tuple[bytes, bytes | None, dict]]: """Load static files from disk with compression.""" www: dict[str, tuple[bytes, bytes | None, dict]] = {} if not self.base.exists(): - 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()] while paths: current = self.base / paths.pop(0) @@ -133,21 +137,18 @@ class Frontend: headers = { "etag": f'"{etag}"', "last-modified": format_date_time(mtime), - "cache-control": ( - "max-age=31536000, immutable" if cached else "no-cache" - ), + "cache-control": ("max-age=31536000, immutable" if cached else "no-cache"), "content-type": mime, } zstd = ZstdCompressor(self.zstdlevel).compress(data) if len(zstd) >= len(data): zstd = None www[name] = data, zstd, headers - if self.favicon: - if m := fnmatch.filter(www, self.favicon): - data, zstd, headers = www[m[0]] - if "immutable" in headers.get("cache-control", ""): - headers = {**headers, "cache-control": "max-age=86400"} - www["/favicon.ico"] = data, zstd, headers + if self.favicon and (m := fnmatch.filter(www, self.favicon)): + data, zstd, headers = www[m[0]] + if "immutable" in headers.get("cache-control", ""): + headers = {**headers, "cache-control": "max-age=86400"} + www["/favicon.ico"] = data, zstd, headers if not www: msg = "Frontend files missing, check your installation.\n" www["/"] = ( @@ -161,7 +162,7 @@ class Frontend: ) 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. 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 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} %)" + "%s: %d files in %.1f ms | zstd %d files %.2f->%.2f MB (%.0f %%)", + self.base.name, + len(self.www), + 1000 * duration, + len(compfiles), + 1e-6 * raw, + 1e-6 * comp, + ratio, ) if self.favicon and "/favicon.ico" not in self.www: logger.warning("Favicon not found: %s", self.favicon) - def route(self, app: FastAPI, mount_path="/"): + def route(self, app: FastAPI, mount_path: str = "/") -> None: """Register frontend routes with a FastAPI app. In SPA/catch-all mode, this must only be called only after all other routes. @@ -204,6 +211,7 @@ class Frontend: Args: app: FastAPI application instance mount_path: Path where the frontend should be mounted (default: "/") + """ self._app = app self._mount_path = mount_path.rstrip("/") @@ -214,7 +222,7 @@ class Frontend: path = self._mount_path + "{path:path}" 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).""" if self._app is None or self._catch_all: return @@ -240,18 +248,19 @@ class Frontend: 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.""" 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"} + content=zstd, + headers={**headers, "content-encoding": "zstd"}, ) 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.""" name = path.removesuffix(self.index) debug = getattr(self._app, "debug", False) @@ -271,11 +280,9 @@ class Frontend: 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 JSONResponse( status_code=409, - content={ - "detail": "[devmode] Not serving frontend files here. Should you connect to Vite instead?" - }, + content={"detail": "[devmode] Use Vite devserver instead."}, ) diff --git a/fastapi_vue_setup.py b/fastapi_vue_setup.py index 1dddcd0..8401d51 100644 --- a/fastapi_vue_setup.py +++ b/fastapi_vue_setup.py @@ -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. @@ -13,8 +13,11 @@ Options: import argparse import ast +import contextlib +import hashlib import importlib.metadata import os +import platform import re import shutil import subprocess @@ -53,46 +56,59 @@ def ruff_format_content( temp_file = target_path.with_suffix(".new.py") try: temp_file.write_text(content, "UTF-8", newline="\n") - # Sort imports first - subprocess.run( - [ - "uv", - "run", - "--with", - "ruff", + if mode == "isort": + # Sort imports only (ignore exit code) + result = subprocess.run( # noqa: S603 + [ # noqa: S607 + "ruff", + "check", + "--select", + "I", + "--fix", + "--output-format=concise", + str(temp_file), + ], + cwd=target_path.parent, + capture_output=True, + check=False, + ) + if result.returncode != 0: + print(result.stdout.decode()) + 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", - "--select", - "I", "--fix", + "--output-format=concise", str(temp_file), ], cwd=target_path.parent, capture_output=True, + check=False, ) - if mode == "isort": - return temp_file.read_text("UTF-8") - # Then format - result = subprocess.run( - ["uv", "run", "--with", "ruff", "ruff", "format", str(temp_file)], + 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: - return temp_file.read_text("UTF-8") - except Exception: + if result.returncode != 0: + print(result.stdout.decode()) + return temp_file.read_text("UTF-8") + except OSError: pass finally: - try: + with contextlib.suppress(Exception): temp_file.unlink(missing_ok=True) - except Exception: - pass return content -def uv_add_packages( - packages: list[str], *, cwd: Path, group: str | None = None -) -> None: +def uv_add_packages(packages: list[str], *, cwd: Path, group: str | None = None) -> None: """Add packages using uv.""" cmd = ["uv", "add", "-q", "-U"] if group: @@ -100,7 +116,7 @@ def uv_add_packages( else: cmd.append("--no-sync") 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: label = f" ({group})" if group else "" print(f"⚠️ Failed to add{label} dependencies") @@ -126,9 +142,7 @@ PYPROJECT_ADDITIONS = { "artifacts": ["MODULE_NAME/frontend-build"], "targets": { "sdist": { - "hooks": { - "custom": {"path": "scripts/fastapi-vue/build-frontend.py"} - }, + "hooks": {"custom": {"path": "scripts/fastapi-vue/build-frontend.py"}}, } }, "only-packages": True, @@ -236,7 +250,8 @@ def parse_ports(ports_str: str | None) -> tuple[int, int, int]: vite = int(parts[1]) dev = int(parts[2]) 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 if dev == vite: @@ -263,9 +278,7 @@ def find_import_insertion_line(source: str) -> int: return 2 if source.startswith("#!") else 1 -def extract_existing_ports( - project_dir: Path, main: 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. Returns (default, vite, dev) or None if not found. @@ -313,6 +326,7 @@ def extract_existing_health(project_dir: Path) -> str | object: 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(): @@ -346,9 +360,7 @@ def find_module_name(project_dir: Path) -> str | None: return None -def find_fastapi_app( - module_dir: Path, project_dir: Path | None = None -) -> 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. Returns (file_path, app_variable_name) or None if not found. @@ -386,9 +398,7 @@ def find_fastapi_app( return None -def _find_app_via_entrypoint( - module_dir: Path, project_dir: Path -) -> tuple[Path, str] | 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"`, @@ -400,7 +410,7 @@ def _find_app_via_entrypoint( try: data = tomlkit.parse(pyproject.read_text("UTF-8")) - except Exception: + except (OSError, ValueError): return None scripts = data.get("project", {}).get("scripts", {}) @@ -410,7 +420,7 @@ def _find_app_via_entrypoint( module_name = module_dir.name # Find script entries that reference this module - for script_name, entry in scripts.items(): + for entry in scripts.values(): if not isinstance(entry, str): continue # Parse entry like "module.subpkg.__main__:main" @@ -468,7 +478,7 @@ def _add_devmode_to_main(content: str) -> str: insert_idx = 0 for i, line in enumerate(lines): stripped = line.strip() - if stripped.startswith("import ") or stripped.startswith("from "): + if stripped.startswith(("import ", "from ")): insert_idx = i + 1 elif stripped and not stripped.startswith("#"): break @@ -501,7 +511,7 @@ def _find_existing_cli_module_path(project_dir: Path, module_name: str) -> str | try: data = tomlkit.parse(pyproject.read_text("UTF-8")) - except Exception: + except (OSError, ValueError): return None scripts = data.get("project", {}).get("scripts", {}) @@ -509,12 +519,11 @@ def _find_existing_cli_module_path(project_dir: Path, module_name: str) -> str | 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}."): + for entry in scripts.values(): + if isinstance(entry, str) and entry.startswith(f"{module_name}.") and ":" in entry: # Extract module path from "module.subpkg.__main__:main" - if ":" in entry: - module_path, _ = entry.rsplit(":", 1) - return module_path + module_path, _ = entry.rsplit(":", 1) + return module_path return None @@ -528,7 +537,7 @@ def _follow_init_reexport(init_file: Path, subpkg_dir: Path) -> tuple[Path, str] """ try: content = init_file.read_text("UTF-8") - except Exception: + except OSError: return None # Look for: from .module import app (or similar variable names) @@ -562,7 +571,7 @@ def _find_app_in_file(path: Path) -> str | None: """Find FastAPI app variable name in a file.""" try: content = path.read_text("UTF-8") - except Exception: + except OSError: return None # Look for FastAPI() instantiation patterns @@ -574,17 +583,15 @@ def _find_app_in_file(path: Path) -> str | None: return None -def render_template(template: str, **kwargs) -> str: - """Simple template rendering replacing KEY with value.""" +def render_template(template: str, **kwargs: str) -> str: + """Render a template, replacing KEY with value.""" result = template for key, value in kwargs.items(): result = result.replace(key, value) return result -def patch_app_file( - path: Path, main_module_path: str, app_var: str, dry: bool = False -) -> bool: +def patch_app_file(path: Path, main_module_path: str, app_var: str, *, dry: bool = False) -> bool: """Patch an existing app.py with frontend integration. Inserts imports at top (ruff will sort them), route at bottom, @@ -628,9 +635,7 @@ def patch_app_file( content = content.rstrip("\n") + "\n" + import_text else: # Insert at the found position - content = ( - "".join(lines[:insert_idx]) + import_text + "".join(lines[insert_idx:]) - ) + content = "".join(lines[:insert_idx]) + import_text + "".join(lines[insert_idx:]) # Insert FRONTEND_BLOCK after last import (only if Frontend wasn't already there) if not has_frontend: @@ -638,7 +643,7 @@ def patch_app_file( last_import_idx = 0 for i, line in enumerate(lines): stripped = line.strip() - if stripped.startswith("import ") or stripped.startswith("from "): + if stripped.startswith(("import ", "from ")): last_import_idx = i elif stripped and not stripped.startswith("#") and last_import_idx > 0: break @@ -649,9 +654,7 @@ def patch_app_file( if route_line not in content: lines = content.split("\n") lines.append("") - lines.append( - "# Serve the Vue frontend (needs to be last if SPA catch-all is used)" - ) + lines.append("# Serve the Vue frontend (needs to be last if SPA catch-all is used)") lines.append(route_line) content = "\n".join(lines) @@ -662,10 +665,7 @@ def patch_app_file( args = match.group(2) if "debug" not in args: # Add debug=DEVMODE as last argument - if args.strip(): - new_args = f"{args}, debug=DEVMODE" - else: - new_args = "debug=DEVMODE" + new_args = f"{args}, debug=DEVMODE" if args.strip() else "debug=DEVMODE" content = ( content[: match.start()] + match.group(1) @@ -701,11 +701,7 @@ def patch_app_file( if insert_idx >= len(lines): content = content.rstrip("\n") + "\n" + import_text else: - content = ( - "".join(lines[:insert_idx]) - + import_text - + "".join(lines[insert_idx:]) - ) + content = "".join(lines[:insert_idx]) + import_text + "".join(lines[insert_idx:]) # Insert lifespan block before the FastAPI() call fastapi_line_pattern = r"^(\w+\s*=\s*FastAPI\s*\()" @@ -723,10 +719,7 @@ def patch_app_file( fastapi_match = re.search(fastapi_pattern, content, re.DOTALL) if fastapi_match and "lifespan" not in fastapi_match.group(2): args = fastapi_match.group(2) - if args.strip(): - new_args = f"{args}, lifespan=lifespan" - else: - new_args = "lifespan=lifespan" + new_args = f"{args}, lifespan=lifespan" if args.strip() else "lifespan=lifespan" content = ( content[: fastapi_match.start()] + fastapi_match.group(1) @@ -770,7 +763,7 @@ def patch_app_file( def patch_vite_config( path: Path, - module_name: str, + *, dry: bool = False, ) -> bool: """Patch an existing vite.config.js/ts by adding fastapi-vue plugin. @@ -801,15 +794,13 @@ def patch_vite_config( # Insert after the last import line before non-import content if not import_inserted: stripped = line.strip() - if stripped.startswith("import ") or stripped.startswith("from "): - # Check if next line is not an import - if i + 1 < len(lines): - next_stripped = lines[i + 1].strip() - if not next_stripped.startswith( - "import " - ) and not next_stripped.startswith("from "): - new_lines.append(import_line) - import_inserted = True + if stripped.startswith(("import ", "from ")) and i + 1 < len(lines): + next_stripped = lines[i + 1].strip() + if not next_stripped.startswith("import ") and not next_stripped.startswith( + "from " + ): + new_lines.append(import_line) + import_inserted = True if not import_inserted: # No imports found, add at top @@ -842,7 +833,7 @@ def patch_vite_config( 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. Tries HelloWorld.vue first (full demo), then falls back to App.vue (minimal). @@ -884,9 +875,7 @@ def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool: is_typescript = 'lang="ts"' in content # Build the script content based on JS/TS - script_addition = ( - TS_HEALTH_CHECK_SCRIPT if is_typescript else JS_HEALTH_CHECK_SCRIPT - ) + script_addition = TS_HEALTH_CHECK_SCRIPT if is_typescript else JS_HEALTH_CHECK_SCRIPT # Insert script addition before script_end_match = re.search(r"", content) @@ -904,9 +893,7 @@ def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool: # Insert before closing h3_close = content.find(" ") if h3_close == -1: - print( - f"⚠️ Skipping {target_file} (no tag found for status insertion)" - ) + print(f"⚠️ Skipping {target_file} (no tag found for status insertion)") return False before, after = content[:h3_close], content[h3_close:] content = f"{before}{indent(STATUS_SPAN_TEMPLATE, ' ')}{after}" @@ -944,12 +931,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) # with module name replaced by MODULE_NAME in the outDir path -_OLD_VITE_PLUGIN_SHA256 = ( - "93713e879c15a25c750a70ce1de684adeaf11b0c723c38da56e5e7ba207f6632" -) +_OLD_VITE_PLUGIN_SHA256 = "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. Old versions didn't have the upgrade marker, so write_file skips them as @@ -961,7 +946,6 @@ def _upgrade_old_vite_plugin(path: Path, module_name: str, dry: bool = False) -> content = path.read_text("UTF-8") if UPGRADE_MARKER in content: return # Already new format, write_file handles it - import hashlib normalized = content.replace( f"../{module_name}/frontend-build", "../MODULE_NAME/frontend-build" @@ -983,6 +967,7 @@ _new_files_written: list[tuple[Path, Path]] = [] def write_file( path: Path, content: str, + *, overwrite: bool = True, dry: bool = False, executable: bool = False, @@ -1020,7 +1005,7 @@ def write_file( if fallback_path is not None: # Write to fallback path instead 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)") return False @@ -1043,6 +1028,7 @@ def _write_fallback_file( original_path: Path, fallback_path: Path, content: str, + *, dry: bool, executable: bool, ) -> bool: @@ -1100,8 +1086,6 @@ def merge_pyproject( 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)) @@ -1147,9 +1131,9 @@ def merge_pyproject( if "custom" not in hatch_build["targets"]["sdist"]["hooks"]: hatch_build["targets"]["sdist"]["hooks"]["custom"] = tomlkit.table() if "path" not in hatch_build["targets"]["sdist"]["hooks"]["custom"]: - hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] = hatch_additions[ - "targets" - ]["sdist"]["hooks"]["custom"]["path"] + hatch_build["targets"]["sdist"]["hooks"]["custom"]["path"] = hatch_additions["targets"][ + "sdist" + ]["hooks"]["custom"]["path"] return data @@ -1193,7 +1177,7 @@ def find_js_runtime() -> tuple[str, str] | 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.""" pyproject = project_dir / "pyproject.toml" if pyproject.exists(): @@ -1205,7 +1189,7 @@ def ensure_python_project(project_dir: Path, dry: bool = False) -> bool: print("📦 No pyproject.toml found, initializing Python project...") 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: print("❌ uv init failed") return False @@ -1219,7 +1203,7 @@ def ensure_python_project(project_dir: Path, dry: bool = False) -> bool: 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.""" frontend_dir = project_dir / "frontend" package_json = frontend_dir / "package.json" @@ -1251,11 +1235,7 @@ def ensure_frontend(project_dir: Path, dry: bool = False) -> bool: print(f">>> {' '.join(create_cmd)}") print("(Follow the prompts to configure your Vue app)") print() - result = subprocess.run( - create_cmd, - cwd=project_dir, - check=False, - ) + result = subprocess.run(create_cmd, cwd=project_dir, check=False) # noqa: S603 if result.returncode != 0: print("❌ create-vue failed") return False @@ -1277,10 +1257,7 @@ def cmd_setup(args: argparse.Namespace) -> int: project_path = Path(args.project_dir) # Handle both "." and "/path/to/project" - if project_path.is_absolute(): - project_dir = project_path - else: - project_dir = Path.cwd() / project_path + project_dir = project_path if project_path.is_absolute() else Path.cwd() / project_path project_dir = project_dir.resolve() dry = args.dry @@ -1300,11 +1277,11 @@ def cmd_setup(args: argparse.Namespace) -> int: print(f"🔧 Setting up project: {project_dir}") # 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 # 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 # Detect module name @@ -1345,9 +1322,7 @@ def cmd_setup(args: argparse.Namespace) -> int: 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}" - ) + print(f"📡 Ports: default={default_port}, vite={vite_port}, dev={dev_port} {ports_note}") # Determine health path configuration # Priority: --health argument > existing project value > default @@ -1374,9 +1349,7 @@ def cmd_setup(args: argparse.Namespace) -> int: project_title = module_name.replace("_", " ").title() # Find existing FastAPI app - app_info = ( - find_fastapi_app(module_dir, project_dir) if module_dir.exists() else None - ) + app_info = find_fastapi_app(module_dir, project_dir) if module_dir.exists() else None # Template variables tpl_vars = { @@ -1531,7 +1504,7 @@ def cmd_setup(args: argparse.Namespace) -> int: # Install the vite plugin file (always update) plugin_file = frontend_dir / "vite-plugin-fastapi.js" # Upgrade old plugin versions that lack the auto-upgrade marker - _upgrade_old_vite_plugin(plugin_file, module_name, dry) + _upgrade_old_vite_plugin(plugin_file, module_name, dry=dry) template = load_template("frontend/vite-plugin-fastapi.js") content = render_template(template, **tpl_vars) write_file(plugin_file, content, overwrite=True, dry=dry) @@ -1541,15 +1514,15 @@ def cmd_setup(args: argparse.Namespace) -> int: vite_config_js = frontend_dir / "vite.config.js" 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(): - patch_vite_config(vite_config_js, module_name, dry) + patch_vite_config(vite_config_js, dry=dry) else: print("⚠️ No vite.config.ts or vite.config.js found in frontend/") print(" Run create-vue first to generate a Vite config to patch.") # 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 === pyproject_path = project_dir / "pyproject.toml" @@ -1588,9 +1561,7 @@ def cmd_setup(args: argparse.Namespace) -> int: else: nl = b"\r\n" if b"\r\n" in gitignore_content else b"\n" suffix = b"" if gitignore_content.endswith(nl) else nl - gitignore_path.write_bytes( - gitignore_content + suffix + gitignore_entry.encode() + nl - ) + gitignore_path.write_bytes(gitignore_content + suffix + gitignore_entry.encode() + nl) print(f"✅ Added {gitignore_entry} to .gitignore") elif dry: print(f"✅ Would create .gitignore with {gitignore_entry}") @@ -1610,12 +1581,12 @@ def cmd_setup(args: argparse.Namespace) -> int: print_boxed("Setup complete!") # 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("_", "-") - message = SETUP_COMPLETE_MESSAGE.replace("CD_CMD", cd_cmd).replace( - "SCRIPT_NAME", script_name - ) + message = SETUP_COMPLETE_MESSAGE.replace("CD_CMD", cd_cmd).replace("SCRIPT_NAME", script_name) + if platform.system() == "Windows": + message = message.replace(" && ", "; ") print(message) # Show merge note if any .new.py files were written @@ -1640,25 +1611,19 @@ def cmd_setup(args: argparse.Namespace) -> int: def is_uninitialized_folder(path: Path) -> bool: """Check if a folder appears to be completely uninitialized.""" - return ( - not (path / "pyproject.toml").exists() and not (path / "package.json").exists() - ) + return not (path / "pyproject.toml").exists() and not (path / "package.json").exists() def is_already_patched(path: Path) -> bool: """Check if a folder has already been patched by fastapi-vue-setup.""" - # Check for our scripts directory - if (path / "scripts" / "fastapi-vue").exists(): - return True - - # Check for vite plugin in frontend - if (path / "frontend" / "vite-plugin-fastapi.js").exists(): - return True - - return False + # Check for our scripts directory for vite plugin in frontend + scriptdir = path / "scripts" / "fastapi-vue" + viteplugin = path / "frontend" / "vite-plugin-fastapi.js" + return scriptdir.exists() or viteplugin.exists() def main() -> int: + """CLI entry point.""" parser = argparse.ArgumentParser( description=f"fastapi-vue-setup {version} - FastAPI + Vue project setup tool", formatter_class=argparse.RawDescriptionHelpFormatter, @@ -1686,11 +1651,9 @@ Examples: parser.add_argument( "--health", metavar="PATH", - help="Health check path for devserver (default: /api/health?from=devserver.py, '' to disable)", - ) - parser.add_argument( - "--dry", "--dry-run", action="store_true", help="Show what would be done" + 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() diff --git a/pyproject.toml b/pyproject.toml index 5902b81..a167c43 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,3 +31,20 @@ dev = ["ruff", "fastapi-vue"] [tool.uv.sources] fastapi-vue = { path = "fastapi-vue", editable = true } + +[tool.uv.workspace] +members = [ + "f", +] + +[tool.ruff] +line-length = 100 + +[tool.ruff.lint] +select = ["ALL"] +ignore = ["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"] diff --git a/scripts/release.py b/scripts/release.py index ecbbbb5..78b5f14 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -10,25 +10,18 @@ DIST = ROOT / "dist" FASTAPI_VUE = ROOT / "fastapi-vue" -def main(): +def main() -> None: + """Build both packages to dist directory.""" # Clear the dist directory if DIST.exists(): shutil.rmtree(DIST) DIST.mkdir() # Build fastapi-vue (subdirectory) to root dist - subprocess.run( - ["uv", "build", "--out-dir", str(DIST)], - cwd=FASTAPI_VUE, - check=True, - ) + subprocess.run(["uv", "build", "--out-dir", str(DIST)], cwd=FASTAPI_VUE, check=True) # noqa: S603, S607 # Build fastapi-vue-setup (root) - subprocess.run( - ["uv", "build", "--out-dir", str(DIST)], - cwd=ROOT, - check=True, - ) + subprocess.run(["uv", "build", "--out-dir", str(DIST)], cwd=ROOT, check=True) # noqa: S603, S607 if __name__ == "__main__": diff --git a/template/backend/__init__.py b/template/backend/__init__.py index e69de29..288b2af 100644 --- a/template/backend/__init__.py +++ b/template/backend/__init__.py @@ -0,0 +1 @@ +"""Backend package with FastAPI application and Vue frontend integration.""" diff --git a/template/backend/__main__.py b/template/backend/__main__.py index 34e26eb..c507a35 100644 --- a/template/backend/__main__.py +++ b/template/backend/__main__.py @@ -1,4 +1,6 @@ # auto-upgrade@fastapi-vue-setup - remove this if you modify this file +"""Command-line entry point for running the backend server.""" + import argparse import os @@ -8,7 +10,8 @@ DEFAULT_PORT = TEMPLATE_DEFAULT_PORT 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.add_argument( "-l", diff --git a/template/backend/app.py b/template/backend/app.py index 9f8fafb..d5e3e46 100644 --- a/template/backend/app.py +++ b/template/backend/app.py @@ -1,3 +1,6 @@ +"""FastAPI application module with Vue frontend integration.""" + +from collections.abc import AsyncIterator from contextlib import asynccontextmanager from pathlib import Path @@ -10,7 +13,7 @@ frontend = Frontend(Path(__file__).with_name("frontend-build")) @asynccontextmanager -async def lifespan(app: FastAPI): +async def lifespan(_app: FastAPI) -> AsyncIterator[None]: """Manage app startup and shutdown resources.""" await frontend.load() 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 @app.get("/api/health") -async def health_check(): +async def health_check() -> dict[str, str]: + """Return backend status for health monitoring.""" return {"status": "ok"} diff --git a/template/scripts/devserver.py b/template/scripts/devserver.py index 51769f4..b518eb1 100755 --- a/template/scripts/devserver.py +++ b/template/scripts/devserver.py @@ -11,7 +11,7 @@ from pathlib import 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"))) -from devutil import ( # type: ignore +from devutil import ( ProcessGroup, check_ports_free, logger, @@ -26,8 +26,11 @@ HEALTH = TEMPLATE_HEALTH 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: + """Start Vite and FastAPI dev servers with hot reload.""" reporoot = Path(__file__).parent.parent front = reporoot / "frontend" if not (front / "package.json").exists(): @@ -50,7 +53,8 @@ async def run_devserver( await pg.spawn(*vite, cwd=front) -def main(): +def main() -> None: + """Parse CLI arguments and run the devserver.""" parser = argparse.ArgumentParser( description="Run Vite and FastAPI development servers", formatter_class=argparse.RawDescriptionHelpFormatter, diff --git a/template/scripts/fastapi-vue/build-frontend.py b/template/scripts/fastapi-vue/build-frontend.py index 152c48c..407e4bf 100644 --- a/template/scripts/fastapi-vue/build-frontend.py +++ b/template/scripts/fastapi-vue/build-frontend.py @@ -1,15 +1,19 @@ +# ruff: noqa: INP001 """Hatch build hook for building Vue frontend during package build.""" import sys from pathlib import Path -from hatchling.builders.hooks.plugin.interface import BuildHookInterface # type: ignore +from hatchling.builders.hooks.plugin.interface import BuildHookInterface sys.path.insert(0, str(Path(__file__).parent)) from buildutil import build -class CustomBuildHook(BuildHookInterface): - def initialize(self, version, build_data): +class CustomBuildHook(BuildHookInterface): # type: ignore[misc] + """Hatch build hook that builds Vue frontend during package build.""" + + def initialize(self, version: str, build_data: dict) -> None: # type: ignore[override] + """Build frontend before package is built.""" super().initialize(version, build_data) build("frontend") diff --git a/template/scripts/fastapi-vue/buildutil.py b/template/scripts/fastapi-vue/buildutil.py index 533c46e..a6fad30 100644 --- a/template/scripts/fastapi-vue/buildutil.py +++ b/template/scripts/fastapi-vue/buildutil.py @@ -1,3 +1,4 @@ +# ruff: noqa: INP001 """Utilities used at build time and in devserver script. No dependencies.""" import logging @@ -7,6 +8,8 @@ import shutil import subprocess from pathlib import Path +MIN_NODE_VERSION = 20 + class _PrefixFormatter(logging.Formatter): """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. """ try: - result = subprocess.run( - [node_path, "--version"], capture_output=True, text=True, check=True + result = subprocess.run( # noqa: S603 + [node_path, "--version"], + capture_output=True, + text=True, + check=True, ) version_str = result.stdout.strip() # Parse version like "v20.10.0" or "v18.17.1" match = re.match(r"v(\d+)", version_str) if match: major_version = int(match.group(1)) - if major_version >= 20: + if major_version >= MIN_NODE_VERSION: return - raise RuntimeError( - f"Node.js {version_str} found, but v20+ required (install with nvm)" - ) + msg = f"Node.js {version_str} found, but v{MIN_NODE_VERSION}+ required" + raise RuntimeError(msg) except (subprocess.CalledProcessError, FileNotFoundError, ValueError): 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]: """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. + Raises RuntimeError 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") + if result := _find_runtime_from_env(options): + return result # 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") + return _auto_detect_runtime(options) -def find_build_tool(): +def find_build_tool() -> tuple[list[str], list[str]]: """Find JavaScript runtime and construct install/build commands. Returns (install_cmd, build_cmd) tuples of command lists. @@ -143,7 +183,7 @@ def find_dev_tool() -> list[str]: if name == "bun": 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]] @@ -176,16 +216,16 @@ def build(folder: str = "frontend") -> None: install_cmd, build_cmd = find_build_tool() except RuntimeError as 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:]] logger.info("### %s", " ".join(display_cmd)) - subprocess.run(cmd, check=True, cwd=folder) + subprocess.run(cmd, check=True, cwd=folder) # noqa: S603 try: run(install_cmd) logger.info("") run(build_cmd) except subprocess.CalledProcessError: - raise SystemExit(1) + raise SystemExit(1) from None diff --git a/template/scripts/fastapi-vue/devutil.py b/template/scripts/fastapi-vue/devutil.py index f4cf749..d691801 100644 --- a/template/scripts/fastapi-vue/devutil.py +++ b/template/scripts/fastapi-vue/devutil.py @@ -1,27 +1,33 @@ +# ruff: noqa: INP001 """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 +from typing import TYPE_CHECKING, Any, Self import httpx from buildutil import find_dev_tool, find_install_tool, logger from fastapi_vue.hostutil import parse_endpoint +if TYPE_CHECKING: + from collections.abc import Coroutine + class ProcessGroup: """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._cmds: dict[int, str] = {} # pid -> command name async def spawn( - self, *cmd: str, cwd: str | None = None + self, + *cmd: str, + cwd: str | None = None, ) -> asyncio.subprocess.Process: """Spawn a subprocess and track it.""" cmd_name = Path(cmd[0]).stem @@ -32,7 +38,8 @@ class ProcessGroup: return proc async def wait( - self, *waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]" + self, + *waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]", ) -> None: """Wait for processes/coroutines to complete, raise SystemExit on failure.""" @@ -43,8 +50,7 @@ class ProcessGroup: raise subprocess.CalledProcessError(returncode, cmd_name) tasks = [ - wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w - for w in waitables + wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w for w in waitables ] try: await asyncio.gather(*tasks) @@ -52,14 +58,15 @@ class ProcessGroup: logger.warning("%s failed with exit status %d", e.cmd, e.returncode) raise SystemExit(1) from None - async def __aenter__(self): + async def __aenter__(self) -> Self: + """Enter the async context manager.""" 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.""" 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] if not running: return @@ -87,7 +94,7 @@ class ProcessGroup: asyncio.wait_for( asyncio.gather(*[p.wait() for p in still_running]), timeout=10, - ) + ), ) except TimeoutError: for p in self._procs: @@ -111,7 +118,7 @@ async def check_ports_free(*urls: str) -> None: await asyncio.gather(*[check(client, url) for url in urls]) -async def ready(url: str, path: str = "", max_attempts=50) -> None: +async def ready(url: str, path: str = "", max_attempts: int = 50) -> None: """Wait for the server to be ready by polling an endpoint. Use empty path to disable the check and make this return immediately. @@ -124,17 +131,19 @@ async def ready(url: str, path: str = "", max_attempts=50) -> None: for attempt in range(max_attempts): try: await client.get(f"{url}{path}", 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) + raise SystemExit(1) from None await asyncio.sleep(0.1) + else: + logger.info("✓ Backend ready!") + return def setup_vite( - endpoint: str, default_port: int = 5173 + endpoint: str, + default_port: int = 5173, ) -> tuple[str, list[str], list[str]]: """Parse frontend endpoint and build commands. @@ -160,7 +169,9 @@ def setup_vite( def setup_fastapi( - endpoint: str, module: str, default_port: int = 8000 + endpoint: str, + module: str, + default_port: int = 8000, ) -> tuple[str, list[str]]: """Parse backend endpoint and build uvicorn command. @@ -175,7 +186,7 @@ def setup_fastapi( host = endpoints[0]["host"] 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 = [ sys.executable, @@ -192,7 +203,9 @@ def setup_fastapi( def setup_cli( - cli: str, endpoint: str, default_port: int = 8000 + cli: str, + endpoint: str, + default_port: int = 8000, ) -> tuple[str, list[str]]: """Parse backend endpoint and build CLI command.