Proper error messages on npm and port check failures, avoid leaking asyncio tasks.

- Terminates instantly when vital process dies (e.g. backend doesn't start)
- Derived of TaskGroup, augmenting its functionality with similar semantics

fastapi-vue/fastapi_vue/logging.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import io
 import logging
+import os
 import re
 import sys
 from contextlib import suppress
@@ -43,6 +44,21 @@ def strip_ansi(text: str) -> str:
     return ANSI_ESCAPE_RE.sub("", text)
 
 
+def use_color(stream: io.TextIOBase = sys.stderr) -> bool:
+    """Test if the stream supports color codes."""
+    if os.environ.get("NO_COLOR"):  # Non empty means no (no-color.org)
+        return False
+    if os.environ.get("FORCE_COLOR", "") not in {"", "0"}:  # force-color.org, node
+        return True
+    if hasattr(stream, "isatty") and stream.isatty():
+        return True
+    with suppress(KeyError, ValueError, OSError):  # Journald does color (-ocat)
+        dev, ino = map(int, os.environ["JOURNAL_STREAM"].split(":", 1))
+        st = os.fstat(stream.fileno())
+        return st.st_dev == dev and st.st_ino == ino
+    return False
+
+
 _LEVEL_EMOJI = {
     logging.DEBUG: "🐛",
     logging.INFO: "🔷",
@@ -94,7 +110,7 @@ class Formatter(logging.Formatter):
         if use_colors in (True, False):
             self.use_colors = use_colors
         else:
-            self.use_colors = sys.stdout.isatty()
+            self.use_colors = use_color(sys.stdout)
         super().__init__(fmt=fmt, datefmt=datefmt, style=style)
 
     def formatMessage(self, record: logging.LogRecord) -> str:  # noqa: N802
@@ -349,7 +365,9 @@ def patch_log_config(log_config, *, access_log: bool = True):  # noqa: ANN001, A
     # watchfiles logs "N changes detected" to its own logger at INFO; only the
     # WARNING "Reloading..." line (uvicorn.error) should show.
     with suppress(Exception):
-        config.setdefault("loggers", {}).setdefault("watchfiles.main", {}).setdefault("level", "WARNING")
+        config.setdefault("loggers", {}).setdefault("watchfiles.main", {}).setdefault(
+            "level", "WARNING"
+        )
 
     # kanta-style output (diffs, colored headers) prints without prefixes,
     # like our access log.  A user-supplied "kanta" logger entry wins.

fastapi-vue/fastapi_vue/server.py

@@ -18,11 +18,17 @@ from .logging import (
     patch_lifespan_logging,
     patch_log_config,
     patch_server_error_middleware,
+    use_color,
 )
 from .startupbox import print_box
 
 tracerite.load()  # Early load on CLI load (import server); uvicorn workers reload via log config
 
+# Install force color to aid tracerite and any external software to use full color when available
+# Define NO_COLOR or FORCE_COLOR beforehand to avoid this
+if "FORCE_COLOR" not in os.environ and use_color():
+    os.environ["FORCE_COLOR"] = "3"
+
 logger = logging.getLogger(__name__)
 
 _WILDCARD_HOSTS = frozenset({"0.0.0.0", "::"})  # noqa: S104

fastapi_vue_setup.py

@@ -9,6 +9,7 @@ Options:
     --module-name NAME      Python module name (auto-detected from pyproject.toml)
     --ports DEFAULT,VITE,DEV  Port configuration (default: 3100,3100,3200)
     --dry                   Show what would be done without making changes
+    -- ARGS                 Extra arguments forwarded to create-vue (e.g. -- --default)
 """
 
 import argparse
@@ -1211,7 +1212,9 @@ 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, *, vue_args: list[str] | None = None, 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"
@@ -1234,6 +1237,10 @@ def ensure_frontend(project_dir: Path, *, dry: bool = False) -> bool:
         "bun": [js_tool, "create", "vue@latest", "frontend"],
     }
     create_cmd = create_vue_commands[js_name]
+    if vue_args:
+        # npm needs a `--` separator so it doesn't eat the arguments;
+        # create-vue runs non-interactively when given feature flags (e.g. --default)
+        create_cmd = [*create_cmd, *(["--"] if js_name == "npm" else []), *vue_args]
 
     if dry:
         print(f"🎨 Would run: {' '.join(create_cmd)}")
@@ -1241,7 +1248,8 @@ def ensure_frontend(project_dir: Path, *, dry: bool = False) -> bool:
 
     print("🎨 No frontend/ found, creating Vue project...")
     print(f">>> {' '.join(create_cmd)}")
-    print("(Follow the prompts to configure your Vue app)")
+    if not vue_args:
+        print("(Follow the prompts to configure your Vue app)")
     print()
     result = subprocess.run(create_cmd, cwd=project_dir, check=False)  # noqa: S603
     if result.returncode != 0:
@@ -1285,7 +1293,7 @@ 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=dry):
+    if not ensure_frontend(project_dir, vue_args=args.vue_args, dry=dry):
         return 1
 
     # Step 2: Ensure Python project exists
@@ -1663,6 +1671,8 @@ Examples:
   fastapi-vue-setup .                  Set up integration in current directory
   fastapi-vue-setup . --dry            Preview what would be done
   fastapi-vue-setup . --ports=8000,5173,8080  Change default ports (backend, vite dev, backend dev)
+  fastapi-vue-setup my-app -- --default  Non-interactive create-vue (extra args after --
+                                         are forwarded to create-vue, e.g. --default, --ts)
 """,
     )
     parser.add_argument(
@@ -1685,7 +1695,15 @@ Examples:
     )
     parser.add_argument("--dry", "--dry-run", action="store_true", help="Show what would be done")
 
-    args = parser.parse_args()
+    # Everything after a standalone `--` is forwarded verbatim to create-vue
+    argv = sys.argv[1:]
+    if "--" in argv:
+        split = argv.index("--")
+        ours, vue_args = argv[:split], argv[split + 1 :]
+    else:
+        ours, vue_args = argv, []
+    args = parser.parse_args(ours)
+    args.vue_args = vue_args
 
     if args.project_dir is None:
         parser.print_help()

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
@@ -48,11 +48,11 @@ async def run_devserver(
     os.environ["ENVPREFIX_DEV"] = "1"
 
     async with ProcessGroup() as pg:
+        pg.create_task(check_ports_free(viteurl, backurl))
         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: normal exit
+    except* (subprocess.SubprocessError, RuntimeError):
+        raise SystemExit(1) from None  # logged in devutil already; exit 1
 
 
 HELP_EPILOG = """

template/scripts/fastapi-vue/devutil.py

@@ -2,106 +2,78 @@
 """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():
+                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
@@ -127,29 +99,30 @@ 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("Conflicting %s already running at %s", server or "server", url)
-            raise SystemExit(1)
-
-    await asyncio.gather(*[check(url) for url in urls])
+            logger.error("Conflicting %s already running at %s", server or "server", url)
+            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.
+    Raises TimeoutError if server doesn't start in time.
     """
     if not path:
         return
@@ -159,8 +132,7 @@ 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)
+            raise TimeoutError(f"Backend at {url} didn't start in time")  # noqa: EM102, TRY003
         await asyncio.sleep(0.1)