diff --git a/template/scripts/devserver.py b/template/scripts/devserver.py index 5362bc4..4ecd0d0 100755 --- a/template/scripts/devserver.py +++ b/template/scripts/devserver.py @@ -5,8 +5,8 @@ import argparse import asyncio import os +import subprocess import sys -from contextlib import suppress from pathlib import Path import tracerite @@ -41,6 +41,7 @@ async def run_devserver( viteurl, npm_install, vite = setup_vite(listen, DEFAULT_VITE_PORT) backurl, MODULE_NAME = setup_cli("PROJECT_CLI", backend, DEFAULT_DEV_PORT) + await check_ports_free(viteurl, backurl) # Tell everyone via environment (vite proxy and backend devmode use these) os.environ["ENVPREFIX_VITE_URL"] = viteurl @@ -49,10 +50,9 @@ async def run_devserver( async with ProcessGroup() as pg: npm_i = await pg.spawn(*npm_install, cwd=front) - await check_ports_free(viteurl, backurl) - await pg.spawn(*MODULE_NAME, *(extra_args or [])) + await pg.spawn(*MODULE_NAME, *(extra_args or []), vital=True) await pg.wait(npm_i, ready(backurl, path=HEALTH)) - await pg.spawn(*vite, cwd=front) + await pg.spawn(*vite, cwd=front, vital=True) def main() -> None: @@ -75,8 +75,12 @@ def main() -> None: help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})", ) args, extra_args = parser.parse_known_args() - with suppress(KeyboardInterrupt): + try: asyncio.run(run_devserver(args.listen, args.backend, extra_args)) + except* KeyboardInterrupt: + pass # user stopped the devserver: exit 0 + except* subprocess.SubprocessError: + raise SystemExit(1) from None # error already logged; exit 1 HELP_EPILOG = """ diff --git a/template/scripts/fastapi-vue/devutil.py b/template/scripts/fastapi-vue/devutil.py index 2c8023b..f331646 100644 --- a/template/scripts/fastapi-vue/devutil.py +++ b/template/scripts/fastapi-vue/devutil.py @@ -2,106 +2,76 @@ """Utilities meant for devserver script, used only in source repository with dev deps.""" import asyncio -import subprocess import sys +from asyncio.subprocess import Process +from collections.abc import Awaitable from contextlib import suppress from pathlib import Path -from typing import TYPE_CHECKING, Any, Self +from subprocess import CalledProcessError +from typing import 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 +class ProcessGroup(asyncio.TaskGroup): + """TaskGroup with structured ownership of async subprocesses.""" -class ProcessGroup: - """Manage async subprocesses with automatic cleanup, like TaskGroup for processes.""" + 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, ...]] = {} - 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, vital: bool = False) -> Process: + """Spawn and own a subprocess. If a vital process exits, the group cancels.""" - async def spawn( - self, - *cmd: str, - cwd: str | None = None, - ) -> asyncio.subprocess.Process: - """Spawn a subprocess and track it.""" - cmd_name = Path(cmd[0]).stem - logger.info(">>> %s", " ".join([cmd_name, *cmd[1:]])) - proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd) - self._procs.append(proc) - self._cmds[proc.pid] = cmd_name - return proc + async def 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( - self, - *waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]", - ) -> None: - """Wait for processes/coroutines to complete, raise SystemExit on failure.""" - - async def wait_proc(proc: asyncio.subprocess.Process) -> None: - returncode = await proc.wait() - if returncode != 0: - cmd_name = self._cmds.get(proc.pid, "unknown") - raise subprocess.CalledProcessError(returncode, cmd_name) - - tasks = [ - wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w for w in waitables - ] - try: - await asyncio.gather(*tasks) - except subprocess.CalledProcessError as e: - logger.warning("%s failed with exit status %d", e.cmd, e.returncode) - raise SystemExit(1) from None - - async def __aenter__(self) -> 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(): + raise CalledProcessError(retcode, self._cmds[w]) + 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