diff --git a/frontend/vite-plugin-fastapi.js b/frontend/vite-plugin-fastapi.js index e7e5945..3e850ae 100644 --- a/frontend/vite-plugin-fastapi.js +++ b/frontend/vite-plugin-fastapi.js @@ -8,11 +8,11 @@ * - Disables Vite's screen clearing on startup * * Options: - * paths - Array of paths to proxy (default: ["/api"]) + * paths - Array of paths to proxy (default: ['/api']) */ -export default function fastapiVue({ paths = ["/api"] } = {}) { - const backendUrl = process.env.PASKIA_BACKEND_URL || "http://localhost:4402" +export default function fastapiVue({ paths = ['/api'] } = {}) { + const backendUrl = process.env.PASKIA_BACKEND_URL || 'http://localhost:4402' // Build proxy configuration for each path const proxy = {} @@ -25,12 +25,12 @@ export default function fastapiVue({ paths = ["/api"] } = {}) { } return { - name: "vite-plugin-fastapi-paskia", + name: 'vite-plugin-fastapi-paskia', config: () => ({ clearScreen: false, server: { proxy }, build: { - outDir: "../paskia/frontend-build", + outDir: '../paskia/frontend-build', emptyOutDir: true, }, }), diff --git a/paskia/__main__.py b/paskia/__main__.py index 733ba9c..78a9ed9 100644 --- a/paskia/__main__.py +++ b/paskia/__main__.py @@ -5,8 +5,8 @@ import os import sys from pathlib import Path -import msgspec -from fastapi_vue import server +from fastapi_vue import env, server, teleport +from fastapi_vue.logging import setup_logging from kanta import Kanta from paskia.db import legacy @@ -17,8 +17,12 @@ from paskia.domains import build as build_registry from paskia.domains import configure as configure_domains from paskia.domains import validate_config from paskia.util import hostutil, startupbox -from paskia.util.constants import DEFAULT_PORT, DEVMODE -from paskia.util.runtime import ServeConfig +from paskia.util.runtime import serve_config + +# Keep the literal value here: fastapi-vue-setup reads DEFAULT_PORT from +# this module on upgrades. The app-side shared copy is paskia.util.constants. +DEFAULT_PORT = 4401 +os.environ["FASTAPI_VUE"] = "PASKIA" EPILOG = """\ Examples: @@ -226,11 +230,10 @@ def cmd_serve(args: argparse.Namespace) -> None: # is the admin's job via the admin interface) are logged by build(). # Pass process-global serve parameters to the server process(es) - os.environ["PASKIA_CONFIG"] = msgspec.json.encode( - ServeConfig(listen=listen) - ).decode() + serve_config().listen = listen + teleport() # Serialize bound config before spawning workers - startupbox.print_startup_config(registry, listen=listen) + startupbox.print_startup_config(registry, listen=listen, default_port=DEFAULT_PORT) # Run the server (spawns processes in dev mode) # tracerite, access logging and log config are handled by fastapi_vue.server; @@ -241,13 +244,13 @@ def cmd_serve(args: argparse.Namespace) -> None: default_port=DEFAULT_PORT, server_header=False, startup_box=None, - reload=Path(__file__).parent if DEVMODE else False, + reload=Path(__file__).parent if env.dev else False, ) def main(): - # Configure logging to remove the "ERROR:root:" prefix - logging.basicConfig(level=logging.INFO, format="%(message)s", force=True) + # Full logging setup (tracerite, formatting) before any CLI output + setup_logging() parser = argparse.ArgumentParser( prog="paskia", diff --git a/paskia/fastapi/mainapp.py b/paskia/fastapi/mainapp.py index 8ed3367..c2e4a2b 100644 --- a/paskia/fastapi/mainapp.py +++ b/paskia/fastapi/mainapp.py @@ -5,7 +5,7 @@ from pathlib import Path from fastapi import FastAPI, HTTPException, Request, Response from fastapi.responses import FileResponse, RedirectResponse -from kanta.logging import configure_logging as configure_kanta_logging +from fastapi_vue import env from paskia import authcode, db, domains, remoteauth from paskia.bootstrap import bootstrap_if_needed @@ -19,12 +19,8 @@ from paskia.fastapi.dispatch import DispatchMiddleware from paskia.fastapi.front import frontend from paskia.fastapi.session import AUTH_COOKIE from paskia.util import passphrase, vitedev -from paskia.util.constants import DEVMODE from paskia.util.runtime import serve_config -# Configure custom logging -configure_kanta_logging() - # Path to examples/index.html when running from source tree _EXAMPLES_DIR = Path(__file__).parent.parent.parent / "examples" @@ -39,7 +35,7 @@ async def lifespan(app: FastAPI): # pragma: no cover - startup path Domain configuration is read from the database. """ cfg = serve_config() - domains.configure(listen=cfg.listen if cfg else None) + domains.configure(listen=cfg.listen) await asyncio.to_thread( Path(kanta.filename).parent.mkdir, parents=True, exist_ok=True @@ -68,7 +64,7 @@ app = FastAPI( docs_url=None, redoc_url=None, openapi_url=None, - debug=DEVMODE, + debug=env.dev, ) # WebSocket and HTTP access logging is handled by fastapi_vue's ASGI middleware; diff --git a/paskia/util/constants.py b/paskia/util/constants.py index d7c9f9d..b322553 100644 --- a/paskia/util/constants.py +++ b/paskia/util/constants.py @@ -1,6 +1,5 @@ """Small, dependency-free constants shared by CLI and server modules.""" -import os - +# App-side default; paskia.__main__ keeps its own literal copy because +# fastapi-vue-setup reads DEFAULT_PORT from the CLI module on upgrades. DEFAULT_PORT = 4401 -DEVMODE = os.getenv("PASKIA_DEV") == "1" diff --git a/paskia/util/hostutil.py b/paskia/util/hostutil.py index 2aa1122..3893def 100644 --- a/paskia/util/hostutil.py +++ b/paskia/util/hostutil.py @@ -98,15 +98,3 @@ def normalize_host(raw_host: str | None) -> str | None: # Strip port from host:port netloc = netloc.rsplit(":", 1)[0] return netloc.lower().rstrip(".") or None - - -def format_endpoint(ep: dict) -> str: - """Format an endpoint dict to a listen string (e.g. 'unix:/path' or 'host:port').""" - if uds := ep.get("uds"): - return f"unix:{uds}" - host = ep["host"] - port = ep["port"] - # Bracket IPv6 addresses - if ":" in host: - host = f"[{host}]" - return f"{host}:{port}" diff --git a/paskia/util/runtime.py b/paskia/util/runtime.py index 6880a90..fbe7e45 100644 --- a/paskia/util/runtime.py +++ b/paskia/util/runtime.py @@ -2,14 +2,13 @@ Domain configuration lives in the database (``Config.domains``); the ``PASKIA_CONFIG`` environment variable only carries the effective listen -endpoints so that child processes (uvicorn reload / workers) can derive -site URLs the same way the parent did. +endpoints so that child processes (uvicorn reload / workers) derive site +URLs the same way the parent did. The CLI entry point mutates the bound +object before ``server.run()`` calls ``teleport()`` to pass it on. """ -import os -from functools import lru_cache - import msgspec +from fastapi_vue import env class ServeConfig(msgspec.Struct): @@ -18,19 +17,11 @@ class ServeConfig(msgspec.Struct): listen: list[str] | None = None -@lru_cache(maxsize=1) -def _load() -> ServeConfig | None: - raw = os.getenv("PASKIA_CONFIG") - if not raw: - return None - return msgspec.json.decode(raw.encode(), type=ServeConfig) - - -def serve_config() -> ServeConfig | None: - """Return cached serve configuration loaded from PASKIA_CONFIG.""" - return _load() +def serve_config() -> ServeConfig: + """Return the serve configuration bound to PASKIA_CONFIG.""" + return env(ServeConfig, name="CONFIG") def clear_cache() -> None: - """Clear cached serve configuration; next serve_config() reloads.""" - _load.cache_clear() + """Drop the bound configuration; next serve_config() re-decodes.""" + env._bindings.pop("CONFIG", None) # noqa: SLF001 diff --git a/paskia/util/startupbox.py b/paskia/util/startupbox.py index 87abab3..6e4a2c7 100644 --- a/paskia/util/startupbox.py +++ b/paskia/util/startupbox.py @@ -2,24 +2,21 @@ from __future__ import annotations -import os import re -from sys import stderr from typing import TYPE_CHECKING from urllib.parse import urlparse +from fastapi_vue import env from fastapi_vue.hostutil import parse_endpoints +from fastapi_vue.server import print_startup_box -from paskia._version import __version__ from paskia.domains import auth_host_url, origin_url, partition_origins from paskia.util import hostutil -from paskia.util.constants import DEFAULT_PORT, DEVMODE -from paskia.util.hostutil import format_endpoint, wildcard_base +from paskia.util.hostutil import wildcard_base if TYPE_CHECKING: from paskia.domains import DomainRegistry -BOX_WIDTH = 80 # Maximum inner width (excluding box chars) URL_COL = 22 # Column where header URLs start (past the logo graphic) # ANSI color codes @@ -28,46 +25,12 @@ YELLOW = "\033[38;5;184m" # Bright yellow (6x6x6 cube, r=4 g=4) BRIGHT_YELLOW = "\033[38;5;226m" # Brightest yellow (6x6x6 cube) BRIGHT_WHITE = "\033[1;37m" # Bold bright white -_TOKENS = re.compile(r"\033\[[0-9;]*m|.") - def _visible_len(text: str) -> int: """Calculate visible length of text, ignoring ANSI escape codes.""" return len(re.sub(r"\033\[[0-9;]*m", "", text)) -def _truncate(text: str, width: int) -> str: - """Cut text to at most `width` visible chars, keeping ANSI codes intact.""" - if _visible_len(text) <= width: - return text - out = [] - visible = 0 - for tok in _TOKENS.findall(text): - if tok.startswith("\033"): - out.append(tok) - elif visible < width - 1: - out.append(tok) - visible += 1 - else: - break - return "".join(out) + "…" + RESET - - -def line(text: str = "", width: int = BOX_WIDTH) -> str: - """Format a line inside the box with proper padding, truncating if needed.""" - text = _truncate(text, width) - padding = width - _visible_len(text) - return f"┃ {text}{' ' * padding} ┃\n" - - -def top(width: int = BOX_WIDTH) -> str: - return "┏" + "━" * (width + 2) + "┓\n" - - -def bottom(width: int = BOX_WIDTH) -> str: - return "┗" + "━" * (width + 2) + "┛\n" - - def _compact_url(url: str) -> str: """Bare host for https URLs; scheme and port kept for plain http.""" stripped = url.removeprefix("https://") @@ -135,7 +98,10 @@ def _signin_summary(in_domain: list[str], rp_id: str) -> str: def print_startup_config( - registry: DomainRegistry, listen: list[str] | None = None + registry: DomainRegistry, + listen: list[str] | None = None, + *, + default_port: int, ) -> None: """Print server configuration on startup (one section per domain).""" # Key graphic with yellow shading (bright for highlights, dark for body) @@ -146,18 +112,15 @@ def print_startup_config( domains = sorted(registry.domains, key=lambda d: d.rp_id) - # Format listen endpoints (dev mode only uses the first endpoint) - endpoints = list(parse_endpoints(listen, DEFAULT_PORT)) - if DEVMODE: - endpoints = endpoints[:1] # server.run reload=True uses only one - parts = [format_endpoint(ep) for ep in endpoints] + # Endpoints as bound, passed to fastapi_vue for the {listen} field + endpoints = list(parse_endpoints(listen, default_port)) # Header URLs: when a vite dev server is configured, its URL (marked # "vite dev"); otherwise one per configured auth host (a full origin URL, # clickable in terminals). If none are configured, guess one domain # (prefer the shortest https rp_id) and link its /auth/ site path. # Entries are pre-styled: bold for the URL, plain for any marker. - vite_url = os.environ.get("PASKIA_VITE_URL") if DEVMODE else None + vite_url = env.vite_url if env.dev else None if vite_url: header_urls = [f"{w}{vite_url}{r} (vite dev)"] else: @@ -178,9 +141,10 @@ def print_startup_config( rows = [] # Logo lines 4-5 carry the first two header URLs; further URLs go on # blank-gutter lines beneath the graphic, all at the same column. + # @VERSION@/@LISTEN@ are filled in by fastapi_vue's print_startup_box. logo = [ f" {b}▄▄▄▄▄{r}", - f"{b}█{y} {b}█{r} Paskia {__version__} @ {' '.join(parts)}", + f"{b}█{y} {b}█{r} Paskia @VERSION@ @ @LISTEN@", f"{b}█{y} {b}█{y}▄▄▄▄▄▄▄▄▄▄▄▄{r}", f"{b}█{y} {b}█{y}▀▀▀▀{b}█{y}▀▀{b}█{y}▀▀{b}█{r}", f" {y}▀▀▀▀▀{r}", @@ -196,7 +160,7 @@ def print_startup_config( rows.append(f"{' ' * URL_COL}{url}") for domain in domains: - # One compact line per domain; overlong lines are capped at render. + # One compact line per domain. rp_name = domain.rp_name suffix = f" ({rp_name})" if rp_name and rp_name != domain.rp_id else "" head = f"{w}{domain.rp_id}{r}{suffix}" @@ -204,21 +168,22 @@ def print_startup_config( rows.append(f"{head} — no sign-in sites") continue in_domain, related = partition_origins(domain.rp_id, domain.config.origins) - parts = [] + phrases = [] if in_domain: - parts.append(_signin_summary(in_domain, domain.rp_id)) - parts.extend(_compact_url(origin_url(k)) for k in sorted(related)) + phrases.append(_signin_summary(in_domain, domain.rp_id)) + phrases.extend(_compact_url(origin_url(k)) for k in sorted(related)) # "with" implies the rp_id itself may sign in (exact key or a full # wildcard); otherwise the origins are a mere list, after a colon. covers_self = any( k == domain.rp_id or k == f"**.{domain.rp_id}" for k in in_domain ) sep = " with " if covers_self else ": " - rows.append(f"{head}{sep}{' and '.join(parts)}") + rows.append(f"{head}{sep}{' and '.join(phrases)}") - # Size the box to the widest row, capped at BOX_WIDTH. - width = min(BOX_WIDTH, max(_visible_len(t) for t in rows)) - out = [top(width)] - out.extend(line(text, width) for text in rows) - out.append(bottom(width)) - stderr.write("".join(out)) + # fastapi_vue prints the box: version from package metadata, listen + # addresses as bound (localhost expanded to both loopbacks). Braces in + # our content (e.g. an rp-name) are escaped before template formatting. + text = "\n".join(rows) + text = text.replace("{", "{{").replace("}", "}}") + text = text.replace("@VERSION@", "{version}").replace("@LISTEN@", "{listen}") + print_startup_box(text, "paskia.fastapi.mainapp:app", endpoints) diff --git a/pyproject.toml b/pyproject.toml index aab438f..36acd44 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,8 +21,8 @@ dependencies = [ "pyjwt[crypto]>=2.11.0", "jsondiff>=2.2.1", "msgspec>=0.20.0", - "fastapi-vue~=1.4.2", - "kanta>=0.7.0", + "fastapi-vue~=1.7.1", + "kanta>=0.9.2", "uarite>=0.2.1", ] [dependency-groups] diff --git a/scripts/devserver.py b/scripts/devserver.py index 71a8adb..256f50a 100755 --- a/scripts/devserver.py +++ b/scripts/devserver.py @@ -10,6 +10,7 @@ import subprocess import sys from contextlib import suppress from pathlib import Path +from subprocess import CalledProcessError from urllib.parse import urlparse import tracerite @@ -68,10 +69,13 @@ def build_caddyfile(origins: list[str], viteurl: str, backurl: str) -> str: return "\n".join(caddyfile_parts) -async def run_caddy( - origins: list[str], viteurl: str, backurl: str -) -> asyncio.subprocess.Process: - """Start Caddy as HTTPS reverse proxy, wait for ready signal.""" +async def run_caddy(origins: list[str], viteurl: str, backurl: str) -> None: + """Run Caddy as HTTPS reverse proxy for the group's lifetime. + + Waits for the ready signal, then drains stderr until Caddy exits or the + task is cancelled (ProcessGroup shutdown), terminating Caddy on exit. + Raises CalledProcessError if Caddy dies, cancelling the group. + """ caddy_path = shutil.which("caddy") if not caddy_path: logger.warning("Caddy not found. Install it to use --caddy option.") @@ -86,57 +90,56 @@ async def run_caddy( stdin=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) - proc.stdin.write(caddyfile.encode()) - await proc.stdin.drain() - proc.stdin.close() + try: + proc.stdin.write(caddyfile.encode()) + await proc.stdin.drain() + proc.stdin.close() - # Wait for ready signal or failure - while True: - if proc.returncode is not None: - remaining = await proc.stderr.read() - for line in remaining.decode().splitlines(): - if line: - logger.info("caddy: %s", line) - logger.warning("Caddy startup failed (exit code %d)", proc.returncode) - raise SystemExit(1) - - line = await proc.stderr.readline() - if not line: - continue - - decoded = line.decode().rstrip() - if "serving initial configuration" in decoded: - break - - # Parse and show errors during startup - if decoded: - try: - log = json.loads(decoded) - level = log.get("level", "") - if level in ("error", "fatal", "warn"): - logger.warning("caddy: %s", log.get("msg", decoded)) - except json.JSONDecodeError: - if "error" in decoded.lower() or "fatal" in decoded.lower(): - logger.warning("caddy: %s", decoded) - - # Start background task to drain stderr - async def drain_caddy_stderr(): + # Wait for ready signal or failure while True: + if proc.returncode is not None: + await log_caddy_stderr(proc.stderr, starting=True) + logger.warning("Caddy startup failed (exit code %d)", proc.returncode) + raise CalledProcessError(proc.returncode, cmd) + line = await proc.stderr.readline() if not line: - break - decoded = line.decode().rstrip() - if decoded: - try: - log = json.loads(decoded) - level = log.get("level", "") - if level in ("error", "fatal", "warn"): - logger.warning("caddy: %s", log.get("msg", decoded)) - except json.JSONDecodeError: - pass # Ignore non-JSON output after startup + continue - asyncio.create_task(drain_caddy_stderr()) - return proc + decoded = line.decode().rstrip() + if "serving initial configuration" in decoded: + break + + log_caddy_line(decoded, starting=True) + + # Drain stderr until Caddy exits + await proc.wait() + await log_caddy_stderr(proc.stderr) + raise CalledProcessError(proc.returncode, cmd) + finally: + with suppress(ProcessLookupError): + proc.terminate() + await proc.wait() + + +def log_caddy_line(decoded: str, *, starting: bool = False) -> None: + """Log one Caddy stderr line (JSON during/after startup).""" + if not decoded: + return + try: + log = json.loads(decoded) + level = log.get("level", "") + if level in ("error", "fatal", "warn"): + logger.warning("caddy: %s", log.get("msg", decoded)) + except json.JSONDecodeError: + if starting and ("error" in decoded.lower() or "fatal" in decoded.lower()): + logger.warning("caddy: %s", decoded) + + +async def log_caddy_stderr(stream: asyncio.StreamReader, *, starting: bool = False) -> None: + """Drain and log remaining Caddy stderr.""" + while line := await stream.readline(): + log_caddy_line(line.decode().rstrip(), starting=starting) def _split_multi(values: list[str] | None) -> list[str]: @@ -204,22 +207,25 @@ async def run_devserver(args: argparse.Namespace, remaining: list[str]) -> None: caddy_origins.append(f"https://{rp_id}") seen: set = set() caddy_origins = [x for x in caddy_origins if not (x in seen or seen.add(x))] - caddy_proc = await run_caddy(caddy_origins, viteurl, backurl) - pg._procs.append(caddy_proc) - pg._cmds[caddy_proc.pid] = "caddy" + pg.create_task(run_caddy(caddy_origins, viteurl, backurl)) + pg.create_task(check_ports_free(viteurl, backurl)) npm_proc = await pg.spawn(*npm_install, cwd=frontend_path) - await check_ports_free(viteurl, backurl) - await pg.spawn(*paskia) + await pg.spawn(*paskia, vital=True) await pg.wait( npm_proc, ready(backurl, path="/auth/api/settings?from=devserver.py") ) - await pg.spawn(*vite, cwd=frontend_path) + await pg.spawn(*vite, cwd=frontend_path, vital=True) def main(): tracerite.load() - parser = argparse.ArgumentParser(add_help=False) + parser = argparse.ArgumentParser( + add_help=False, + description="Run Vite and FastAPI development servers", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=HELP_EPILOG, + ) parser.add_argument( "-l", "--listen", @@ -243,8 +249,12 @@ def main(): ) args, remaining = parser.parse_known_args() - with suppress(KeyboardInterrupt): + try: asyncio.run(run_devserver(args, remaining)) + except* KeyboardInterrupt: + pass # user stopped the devserver: normal exit + except* subprocess.SubprocessError, RuntimeError: + raise SystemExit(1) from None # logged in devutil already; exit 1 HELP_EPILOG = """ diff --git a/scripts/fastapi-vue/buildutil.py b/scripts/fastapi-vue/buildutil.py index 3150423..5241b00 100644 --- a/scripts/fastapi-vue/buildutil.py +++ b/scripts/fastapi-vue/buildutil.py @@ -10,21 +10,31 @@ from pathlib import Path MIN_NODE_VERSION = 20 +# Duplicated from fastapi_vue.logging because build environment is isolated +_LEVEL_EMOJI = { + logging.DEBUG: "🐛", + logging.INFO: "🔷", + logging.WARNING: "❗", + logging.ERROR: "🛑", + logging.CRITICAL: "🚨", +} -class _PrefixFormatter(logging.Formatter): - """Formatter that adds prefix based on log level.""" + +class _Formatter(logging.Formatter): + """Emoji level prefix formatter, mirroring fastapi_vue.logging.Formatter.""" def format(self, record: logging.LogRecord) -> str: - if record.levelno >= logging.WARNING: - return f"⚠️ {record.getMessage()}" - return record.getMessage() + emoji = _LEVEL_EMOJI.get(record.levelno) + prefix = f"{emoji} " if emoji else f"{record.levelname}: " + return prefix + record.getMessage() _handler = logging.StreamHandler() -_handler.setFormatter(_PrefixFormatter()) +_handler.setFormatter(_Formatter()) logger = logging.getLogger("fastapi-vue") logger.addHandler(_handler) logger.setLevel(logging.INFO) +logger.propagate = False # own handler; do not double-print via a configured root def _check_node_version(node_path: str) -> None: diff --git a/scripts/fastapi-vue/devutil.py b/scripts/fastapi-vue/devutil.py index 72b34b6..eeb2587 100644 --- a/scripts/fastapi-vue/devutil.py +++ b/scripts/fastapi-vue/devutil.py @@ -1,108 +1,87 @@ # ruff: noqa: INP001 """Utilities meant for devserver script, used only in source repository with dev deps.""" +from __future__ import annotations + import asyncio -import subprocess import sys +from asyncio.subprocess import Process from contextlib import suppress from pathlib import Path -from typing import TYPE_CHECKING, Any, Self +from subprocess import CalledProcessError +from typing import TYPE_CHECKING, Any from urllib.parse import urlsplit from buildutil import find_dev_tool, find_install_tool, logger from fastapi_vue.hostutil import parse_endpoint if TYPE_CHECKING: - from collections.abc import Coroutine + from collections.abc import Awaitable -class ProcessGroup: - """Manage async subprocesses with automatic cleanup, like TaskGroup for processes.""" +class ProcessGroup(asyncio.TaskGroup): + """TaskGroup with structured ownership of async subprocesses.""" - def __init__(self) -> None: - """Initialize empty process tracking.""" - self._procs: list[asyncio.subprocess.Process] = [] - self._cmds: dict[int, str] = {} # pid -> command name + def __init__(self, *, terminate_timeout: float = 10) -> None: + """Set the grace period before terminate() escalates to kill().""" + super().__init__() + self._terminate_timeout = terminate_timeout + self._cmds: dict[Process, tuple[str, ...]] = {} async def spawn( - self, - *cmd: str, - cwd: str | None = None, - ) -> asyncio.subprocess.Process: - """Spawn a subprocess and track it.""" - cmd_name = Path(cmd[0]).stem - logger.info(">>> %s", " ".join([cmd_name, *cmd[1:]])) - proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd) - self._procs.append(proc) - self._cmds[proc.pid] = cmd_name - return proc + self, *cmd: str, cwd: str | None = None, vital: bool = False + ) -> Process: + """Spawn and own a subprocess. If a vital process exits, the group cancels.""" - async def wait( - self, - *waitables: asyncio.subprocess.Process | Coroutine[Any, Any, Any], - ) -> None: - """Wait for processes/coroutines to complete, raise SystemExit on failure.""" + async def run() -> None: + name = Path(cmd[0]).stem + logger.info(">>> %s", " ".join([name, *cmd[1:]])) + try: + proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd) + self._cmds[proc] = cmd + started.set_result(proc) + except Exception as e: # noqa: BLE001 + started.set_exception(e) + return - async def wait_proc(proc: asyncio.subprocess.Process) -> None: - returncode = await proc.wait() - if returncode != 0: - cmd_name = self._cmds.get(proc.pid, "unknown") - raise subprocess.CalledProcessError(returncode, cmd_name) - - tasks = [ - wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w - for w in waitables - ] - try: - await asyncio.gather(*tasks) - except subprocess.CalledProcessError as e: - logger.warning("%s failed with exit status %d", e.cmd, e.returncode) - raise SystemExit(1) from None - - async def __aenter__(self) -> Self: - """Enter the async context manager.""" - return self - - 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) -> None: - running = [p for p in self._procs if p.returncode is None] - if not running: - return - - if not immediate: - # Wait for any one process to exit - with suppress(asyncio.CancelledError): - await asyncio.wait( - [asyncio.create_task(p.wait()) for p in running], - return_when=asyncio.FIRST_COMPLETED, - ) - - # Terminate remaining processes - for p in self._procs: - if p.returncode is None: + try: + returncode = await proc.wait() + finally: with suppress(ProcessLookupError): - p.terminate() - - # Wait for all to finish (with overall timeout), shielded from cancellation - still_running = [p for p in self._procs if p.returncode is None] - if still_running: - with suppress(asyncio.CancelledError): + proc.terminate() try: - await asyncio.shield( - asyncio.wait_for( - asyncio.gather(*[p.wait() for p in still_running]), - timeout=10, - ), - ) + await asyncio.wait_for(proc.wait(), self._terminate_timeout) except TimeoutError: - for p in self._procs: - if p.returncode is None: - with suppress(ProcessLookupError): - p.kill() - await p.wait() + with suppress(ProcessLookupError): + proc.kill() + await proc.wait() + + if vital: + logger.warning("Vital process %s exited", name) + raise CalledProcessError(returncode, cmd) + + started = asyncio.get_running_loop().create_future() + self.create_task(run()) + return await asyncio.shield(started) + + async def wait(self, *waitables: Process | Awaitable) -> tuple[Any, ...]: + """Wait concurrently and return results in argument order.""" + + async def task(w: Process | Awaitable) -> Any: # noqa: ANN401 + if not isinstance(w, Process): + return await w + if retcode := await w.wait(): + cmd = self._cmds[w] + logger.warning( + "Process %s exited with status %d", Path(cmd[0]).stem, retcode + ) + raise CalledProcessError(retcode, cmd) + return retcode + + async with asyncio.TaskGroup() as group: + tasks = [group.create_task(task(w)) for w in waitables] + + return tuple(task.result() for task in tasks) async def http_get_server(url: str, timeout: float) -> str | None: # noqa: ASYNC109 @@ -128,31 +107,32 @@ async def http_get_server(url: str, timeout: float) -> str | None: # noqa: ASYN writer.close() except OSError, EOFError, ValueError, TimeoutError: return None - for line in data.decode("latin-1").split("\r\n"): + for line in data.decode(errors="replace").split("\r\n"): if line.lower().startswith("server:"): - return line.split(":", 1)[1].strip() + return line[7:].strip() return "" async def check_ports_free(*urls: str) -> None: - """Verify URLs are not responding (ports are free). Raise SystemExit if any respond.""" + """Verify URLs are not responding (ports are free). - async def check(url: str) -> None: - server = await http_get_server(url, timeout=0.1) + Meant to run as a task inside a TaskGroup. Logs the conflict and raises + RuntimeError (handled like a failed process) if any URL responds. + """ + servers = await asyncio.gather(*(http_get_server(url, timeout=0.1) for url in urls)) + for url, server in zip(urls, servers, strict=True): if server is not None: - logger.warning( + logger.error( "Conflicting %s already running at %s", server or "server", url ) - raise SystemExit(1) - - await asyncio.gather(*[check(url) for url in urls]) + raise RuntimeError(url) async def ready(url: str, path: str = "", max_attempts: int = 50) -> None: """Wait for the server to be ready by polling an endpoint. Use empty path to disable the check and make this return immediately. - Raises SystemExit(1) if server doesn't start in time. + Logs, then raises RuntimeError if the server doesn't start in time. """ if not path: return @@ -162,8 +142,8 @@ async def ready(url: str, path: str = "", max_attempts: int = 50) -> None: logger.info("✓ Backend ready!") return if attempt == max_attempts - 1: - logger.warning("Backend didn't start in time") - raise SystemExit(1) + logger.error("Backend at %s didn't start in time", url) + raise RuntimeError(url) await asyncio.sleep(0.1)