Updated fastapi-vue-setup 1.7.2
release / gui-build (linux, bash) (push) Successful in 55s
release / gui-build (windows, cmd) (push) Successful in 1m19s
release / gui-build (macos, bash) (push) Failing after 1m14s

- DEVMODE removed; FASTAPI_VUE=MEDIAHIVE is set at entrypoints and
  fastapi_vue.env drives dev-mode checks
- CLI roots now reach the server via fastapi-vue's env(Config) teleport
  (mediahive.config.config) instead of MEDIAHIVE_ROOTS
- Legacy media_folder config field and migration removed
- mediahive.* logger level set via log_config (DEBUG in dev, INFO in prod)
  in both server.run() and winmain's patched uvicorn config
This commit is contained in:
2026-09-23 18:41:51 +00:00
parent 1477c240a1
commit f2fc6f657f
11 changed files with 165 additions and 171 deletions
+1 -1
View File
@@ -32,7 +32,7 @@ All media paths are scoped to a **root**, identified by a friendly `root_id`
- `GET /api/assets/{root_id}/{asset_type}/{asset_path:path}` is constrained to `<root>/.mediahive/{asset_type}` where `asset_type` is one of `movies`, `series`, `people`. - `GET /api/assets/{root_id}/{asset_type}/{asset_path:path}` is constrained to `<root>/.mediahive/{asset_type}` where `asset_type` is one of `movies`, `series`, `people`.
- `GET /api/meta/{root_id}/{meta_key}` supports metadata keys currently limited to `playback-state` and `scanignore`. - `GET /api/meta/{root_id}/{meta_key}` supports metadata keys currently limited to `playback-state` and `scanignore`.
- `GET /api/player/status` returns `{ "remote": true|false }`. - `GET /api/player/status` returns `{ "remote": true|false }`.
- Roots may also be provided at startup via the `MEDIAHIVE_ROOTS` environment variable (JSON dict of name → path), which overrides the persisted configuration. - Roots may also be provided at startup via CLI arguments (`mediahive /path/to/media ...`), which are passed to the server through fastapi-vue's env config (`mediahive.config.config`) and override the persisted configuration.
- Assumed playback: after `POST /api/play/{root_id}` the launched item is assumed to be playing while the frontend reports no input activity. On the next `POST /api/activity` the guessed position (`resume base + elapsed`, capped at the TMDb runtime) is written once; watches under 5 minutes are discarded (a peek is not progress). A resume entry written by another tracker (e.g. the GUI's MPC-BE tracker) during the session overrides the guess. The MPC-BE tracker likewise ignores sessions shorter than 5 minutes. - Assumed playback: after `POST /api/play/{root_id}` the launched item is assumed to be playing while the frontend reports no input activity. On the next `POST /api/activity` the guessed position (`resume base + elapsed`, capped at the TMDb runtime) is written once; watches under 5 minutes are discarded (a peek is not progress). A resume entry written by another tracker (e.g. the GUI's MPC-BE tracker) during the session overrides the guess. The MPC-BE tracker likewise ignores sessions shorter than 5 minutes.
## WebSocket ## WebSocket
+5 -5
View File
@@ -8,11 +8,11 @@
* - Disables Vite's screen clearing on startup * - Disables Vite's screen clearing on startup
* *
* Options: * Options:
* paths - Array of paths to proxy (default: ["/api"]) * paths - Array of paths to proxy (default: ['/api'])
*/ */
export default function fastapiVue({ paths = ["/api"] } = {}) { export default function fastapiVue({ paths = ['/api'] } = {}) {
const backendUrl = process.env.MEDIAHIVE_BACKEND_URL || "http://localhost:8421" const backendUrl = process.env.MEDIAHIVE_BACKEND_URL || 'http://localhost:8421'
// Build proxy configuration for each path // Build proxy configuration for each path
const proxy = {} const proxy = {}
@@ -25,12 +25,12 @@ export default function fastapiVue({ paths = ["/api"] } = {}) {
} }
return { return {
name: "vite-plugin-fastapi-mediahive", name: 'vite-plugin-fastapi-mediahive',
config: () => ({ config: () => ({
clearScreen: false, clearScreen: false,
server: { proxy }, server: { proxy },
build: { build: {
outDir: "../mediahive/frontend-build", outDir: '../mediahive/frontend-build',
emptyOutDir: true, emptyOutDir: true,
}, },
}), }),
+17 -7
View File
@@ -1,16 +1,20 @@
"""MediaHive CLI entrypoint.""" """MediaHive CLI entrypoint."""
import os
# Must be set before fastapi_vue env bindings are created (mediahive.config).
os.environ["FASTAPI_VUE"] = "MEDIAHIVE"
import argparse import argparse
import asyncio import asyncio
import json
import os
import sys import sys
from pathlib import Path from pathlib import Path
from fastapi_vue import server from fastapi_vue import env, server
from mediahive.config import config
DEFAULT_PORT = 8420 DEFAULT_PORT = 8420
DEVMODE = os.getenv("MEDIAHIVE_DEV") == "1"
def _configure_windows_event_loop_policy() -> None: def _configure_windows_event_loop_policy() -> None:
@@ -140,10 +144,11 @@ def main() -> None:
name = f"{base_name}{suffix}" name = f"{base_name}{suffix}"
suffix += 1 suffix += 1
roots[name] = p.as_posix() roots[name] = p.as_posix()
os.environ["MEDIAHIVE_ROOTS"] = json.dumps(roots) # Teleported to the server process by fastapi-vue's server.run().
config.roots = roots
if ( if (
DEVMODE env.dev
and sys.platform == "win32" and sys.platform == "win32"
and os.environ.get("MEDIAHIVE_DEV_CHILD") != "1" and os.environ.get("MEDIAHIVE_DEV_CHILD") != "1"
): ):
@@ -156,7 +161,12 @@ def main() -> None:
default_port=DEFAULT_PORT, default_port=DEFAULT_PORT,
server_header=False, server_header=False,
loop="none" if sys.platform == "win32" else "auto", loop="none" if sys.platform == "win32" else "auto",
reload=Path(__file__).parent if DEVMODE and sys.platform != "win32" else False, reload=Path(__file__).parent if env.dev and sys.platform != "win32" else False,
# fastapi-vue routes the root logger at INFO in dev / WARNING in prod;
# keep our own loggers visible in production too.
log_config={
"loggers": {"mediahive": {"level": "DEBUG" if env.dev else "INFO"}}
},
) )
+8 -16
View File
@@ -13,14 +13,20 @@ from pathlib import Path
import msgspec import msgspec
import msgspec.toml import msgspec.toml
from fastapi_vue import env
from platformdirs import user_config_path, user_log_path from platformdirs import user_config_path, user_log_path
class Config(msgspec.Struct, omit_defaults=True): class Config(msgspec.Struct, omit_defaults=True):
media_folder: str | None = None
roots: dict[str, str] | None = None roots: dict[str, str] | None = None
# Runtime config shared between the CLI entrypoint and the server process via
# fastapi-vue's env teleport (MEDIAHIVE_CONFIG). Values set here take
# precedence over the persisted config file.
config = env(Config)
def config_dir() -> Path: def config_dir() -> Path:
# appauthor=False: avoid the doubled %LOCALAPPDATA%\mediahive\mediahive. # appauthor=False: avoid the doubled %LOCALAPPDATA%\mediahive\mediahive.
# roaming=False: config is machine-specific state, not something to sync # roaming=False: config is machine-specific state, not something to sync
@@ -37,25 +43,11 @@ def config_path() -> Path:
return config_dir() / "config.toml" return config_dir() / "config.toml"
def _migrate_legacy_media_folder(cfg: Config) -> Config:
"""If roots is empty but media_folder exists, seed roots with it."""
if cfg.roots:
return cfg
if not cfg.media_folder:
return cfg
path = Path(cfg.media_folder)
name = path.name or path.anchor.strip("/\\").lower() or "media"
# Resolve collisions simply by using the basename; if user had weird layout
# they can rename via the UI later.
return msgspec.structs.replace(cfg, roots={name: cfg.media_folder})
def load_config() -> Config: def load_config() -> Config:
path = config_path() path = config_path()
if path.exists(): if path.exists():
try: try:
cfg = msgspec.toml.decode(path.read_bytes(), type=Config) return msgspec.toml.decode(path.read_bytes(), type=Config)
return _migrate_legacy_media_folder(cfg)
except OSError, msgspec.DecodeError, msgspec.ValidationError: except OSError, msgspec.DecodeError, msgspec.ValidationError:
return Config() return Config()
return Config() return Config()
+9 -6
View File
@@ -1,13 +1,18 @@
"""Hivescan CLI entrypoint.""" """Hivescan CLI entrypoint."""
import os
# Must be set before fastapi_vue env bindings are created (mediahive.config).
os.environ.setdefault("FASTAPI_VUE", "MEDIAHIVE")
import argparse import argparse
import asyncio import asyncio
import json
import logging import logging
import os
import sys import sys
from pathlib import Path from pathlib import Path
from mediahive.config import config
def _configure_windows_event_loop_policy() -> None: def _configure_windows_event_loop_policy() -> None:
"""Ensure Windows uses Proactor loop so asyncio subprocess APIs are available.""" """Ensure Windows uses Proactor loop so asyncio subprocess APIs are available."""
@@ -55,11 +60,9 @@ The server exposes a unified endpoint:
args = parser.parse_args() args = parser.parse_args()
# Defer filesystem validation to the server; pass raw path via env. # Defer filesystem validation to the server; pass raw path via env config.
media_root = Path(args.media_folder).expanduser() media_root = Path(args.media_folder).expanduser()
os.environ["MEDIAHIVE_ROOTS"] = json.dumps({ config.roots = {media_root.name or "media": media_root.as_posix()}
media_root.name or "media": media_root.as_posix()
})
logging.basicConfig( logging.basicConfig(
level=logging.INFO, level=logging.INFO,
+10 -20
View File
@@ -37,10 +37,9 @@ from fastapi.responses import (
Response, Response,
StreamingResponse, StreamingResponse,
) )
from fastapi_vue import Frontend from fastapi_vue import Frontend, env
from mediahive.__main__ import DEVMODE from mediahive.config import config, load_config, log_dir
from mediahive.config import load_config, log_dir
from mediahive.hivescan.images import close_image_client from mediahive.hivescan.images import close_image_client
from mediahive.hivescan.scanner import RootScanner from mediahive.hivescan.scanner import RootScanner
from mediahive.hivescan.tmdb_client import close_http_client from mediahive.hivescan.tmdb_client import close_http_client
@@ -974,23 +973,14 @@ async def _activate_all_roots() -> None:
""" """
desired: dict[str, str] = {} desired: dict[str, str] = {}
# 1. CLI roots via MEDIAHIVE_ROOTS (JSON dict) # 1. CLI roots (teleported via fastapi-vue's env config) take precedence
env_roots_raw = os.environ.get("MEDIAHIVE_ROOTS")
env_roots: dict[str, str] | None = None
if env_roots_raw:
try:
parsed = json.loads(env_roots_raw)
if isinstance(parsed, dict):
env_roots = parsed
except Exception:
logger.exception("Failed to parse MEDIAHIVE_ROOTS")
# 2. Persisted config roots (used only when CLI roots are not provided) # 2. Persisted config roots (used only when CLI roots are not provided)
cfg = load_config() if config.roots:
if env_roots is not None: desired.update(config.roots)
desired.update(env_roots) else:
elif cfg.roots: cfg = load_config()
desired.update(cfg.roots) if cfg.roots:
desired.update(cfg.roots)
if not desired: if not desired:
logger.info("No roots configured; waiting for PUT /api/config/roots") logger.info("No roots configured; waiting for PUT /api/config/roots")
@@ -1056,7 +1046,7 @@ async def lifespan(_app: FastAPI):
await close_image_client() await close_image_client()
app = FastAPI(title="MediaHive Server", lifespan=lifespan, debug=DEVMODE) app = FastAPI(title="MediaHive Server", lifespan=lifespan, debug=env.dev)
# Allow CORS for development # Allow CORS for development
app.add_middleware( app.add_middleware(
+18 -10
View File
@@ -24,15 +24,20 @@ import urllib.request
from concurrent.futures import Future, ThreadPoolExecutor from concurrent.futures import Future, ThreadPoolExecutor
from pathlib import Path from pathlib import Path
# Must be set before fastapi_vue env bindings are created (mediahive.config);
# this module is the PyInstaller entry point and may run without __main__.
os.environ.setdefault("FASTAPI_VUE", "MEDIAHIVE")
import msgspec.structs import msgspec.structs
import uvicorn import uvicorn
import velopack import velopack
import webview import webview
from fastapi_vue import env
from fastapi_vue.logging import patch_log_config from fastapi_vue.logging import patch_log_config
from fastapi_vue.startupbox import print_box from fastapi_vue.startupbox import print_box
from tracerite.html import html_traceback from tracerite.html import html_traceback
from mediahive.config import load_config, log_dir, save_config from mediahive.config import config, load_config, log_dir, save_config
from mediahive.volume_control import get_volume, set_volume, volume_max from mediahive.volume_control import get_volume, set_volume, volume_max
logger = logging.getLogger("mediahive.winmain") logger = logging.getLogger("mediahive.winmain")
@@ -1202,10 +1207,6 @@ def winmain() -> None:
initial_roots[name] = p.as_posix() initial_roots[name] = p.as_posix()
elif cfg.roots: elif cfg.roots:
initial_roots = cfg.roots initial_roots = cfg.roots
elif cfg.media_folder:
p = _normalize_media_root_input(cfg.media_folder)
name = p.name or "media"
initial_roots[name] = p.as_posix()
if not initial_roots: if not initial_roots:
folder = _run_initial_setup() folder = _run_initial_setup()
@@ -1219,8 +1220,9 @@ def winmain() -> None:
if cfg.roots != initial_roots: if cfg.roots != initial_roots:
save_config(msgspec.structs.replace(cfg, roots=initial_roots)) save_config(msgspec.structs.replace(cfg, roots=initial_roots))
# Pass roots to the server via env (validation deferred to server startup) # Pass roots to the in-process server via the shared env config
os.environ["MEDIAHIVE_ROOTS"] = json.dumps(initial_roots) # (validation deferred to server startup)
config.roots = initial_roots
backend_port = _reserve_backend_port() backend_port = _reserve_backend_port()
backend_url = f"http://{BACKEND_HOST}:{backend_port}" backend_url = f"http://{BACKEND_HOST}:{backend_port}"
@@ -1238,7 +1240,13 @@ def winmain() -> None:
# log config wires up its access-log middleware, emoji level prefixes and # log config wires up its access-log middleware, emoji level prefixes and
# tracerite tracebacks (colors are auto-disabled when stderr is not a tty, # tracerite tracebacks (colors are auto-disabled when stderr is not a tty,
# e.g. redirected to the log file in frozen builds). # e.g. redirected to the log file in frozen builds).
config = uvicorn.Config( log_config = patch_log_config(uvicorn.config.LOGGING_CONFIG)
# fastapi-vue routes the root logger at INFO in dev / WARNING in prod;
# keep our own loggers visible in production too.
log_config.setdefault("loggers", {})["mediahive"] = {
"level": "DEBUG" if env.dev else "INFO"
}
uvicorn_config = uvicorn.Config(
"mediahive.server:app", "mediahive.server:app",
host=BACKEND_HOST, host=BACKEND_HOST,
port=backend_port, port=backend_port,
@@ -1246,9 +1254,9 @@ def winmain() -> None:
server_header=False, server_header=False,
timeout_graceful_shutdown=0, timeout_graceful_shutdown=0,
access_log=False, # fastapi-vue's middleware replaces uvicorn's access_log=False, # fastapi-vue's middleware replaces uvicorn's
log_config=patch_log_config(uvicorn.config.LOGGING_CONFIG), log_config=log_config,
) )
server = uvicorn.Server(config) server = uvicorn.Server(uvicorn_config)
backend_thread = threading.Thread( backend_thread = threading.Thread(
target=server.run, daemon=True, name="mediahive-backend" target=server.run, daemon=True, name="mediahive-backend"
) )
+1 -1
View File
@@ -8,7 +8,7 @@ dependencies = [
"aiofiles>=25.1.0", "aiofiles>=25.1.0",
"aiopathlib>=0.6.0", "aiopathlib>=0.6.0",
"bencodepy>=0.9.5", "bencodepy>=0.9.5",
"fastapi-vue>=1.4.1", "fastapi-vue~=1.7.2",
"fastapi[standard]>=0.128.0", "fastapi[standard]>=0.128.0",
"httpx[http2]>=0.28.1", "httpx[http2]>=0.28.1",
"msgspec>=0.19", "msgspec>=0.19",
+9 -5
View File
@@ -5,8 +5,8 @@
import argparse import argparse
import asyncio import asyncio
import os import os
import subprocess
import sys import sys
from contextlib import suppress
from pathlib import Path from pathlib import Path
import tracerite import tracerite
@@ -48,11 +48,11 @@ async def run_devserver(
os.environ["MEDIAHIVE_DEV"] = "1" os.environ["MEDIAHIVE_DEV"] = "1"
async with ProcessGroup() as pg: async with ProcessGroup() as pg:
pg.create_task(check_ports_free(viteurl, backurl))
npm_i = await pg.spawn(*npm_install, cwd=front) npm_i = await pg.spawn(*npm_install, cwd=front)
await check_ports_free(viteurl, backurl) await pg.spawn(*mediahive, *(extra_args or []), vital=True)
await pg.spawn(*mediahive, *(extra_args or []))
await pg.wait(npm_i, ready(backurl, path=HEALTH)) 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: def main() -> None:
@@ -75,8 +75,12 @@ def main() -> None:
help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})", help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})",
) )
args, extra_args = parser.parse_known_args() args, extra_args = parser.parse_known_args()
with suppress(KeyboardInterrupt): try:
asyncio.run(run_devserver(args.listen, args.backend, extra_args)) 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 = """ HELP_EPILOG = """
+11 -4
View File
@@ -10,20 +10,27 @@ from pathlib import Path
MIN_NODE_VERSION = 20 MIN_NODE_VERSION = 20
class _PrefixFormatter(logging.Formatter): class _Formatter(logging.Formatter):
"""Formatter that adds prefix based on log level.""" """Prefix formatter, intentionally different from fastapi_vue.logging.
INFO and below pass through unprefixed so messages can use their own
markings (>>>, ###); WARNING and above get an emoji prefix.
"""
def format(self, record: logging.LogRecord) -> str: def format(self, record: logging.LogRecord) -> str:
if record.levelno >= logging.ERROR:
return f"🛑 {record.getMessage()}"
if record.levelno >= logging.WARNING: if record.levelno >= logging.WARNING:
return f"⚠️ {record.getMessage()}" return f"💣 {record.getMessage()}"
return record.getMessage() return record.getMessage()
_handler = logging.StreamHandler() _handler = logging.StreamHandler()
_handler.setFormatter(_PrefixFormatter()) _handler.setFormatter(_Formatter())
logger = logging.getLogger("fastapi-vue") logger = logging.getLogger("fastapi-vue")
logger.addHandler(_handler) logger.addHandler(_handler)
logger.setLevel(logging.INFO) 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: def _check_node_version(node_path: str) -> None:
+76 -96
View File
@@ -1,110 +1,89 @@
"""Utilities meant for devserver script, used only in source repository with dev deps.""" """Utilities meant for devserver script, used only in source repository with dev deps."""
from __future__ import annotations
import asyncio import asyncio
import subprocess
import sys import sys
from asyncio.subprocess import Process
from contextlib import suppress from contextlib import suppress
from pathlib import Path 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 urllib.parse import urlsplit
from buildutil import find_dev_tool, find_install_tool, logger from buildutil import find_dev_tool, find_install_tool, logger
from fastapi_vue.hostutil import parse_endpoint from fastapi_vue.hostutil import parse_endpoint
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Coroutine from collections.abc import Awaitable
class ProcessGroup: class ProcessGroup(asyncio.TaskGroup):
"""Manage async subprocesses with automatic cleanup, like TaskGroup for processes.""" """TaskGroup with structured ownership of async subprocesses."""
def __init__(self) -> None: def __init__(self, *, terminate_timeout: float = 10) -> None:
"""Initialize empty process tracking.""" """Set the grace period before terminate() escalates to kill()."""
self._procs: list[asyncio.subprocess.Process] = [] super().__init__()
self._cmds: dict[int, str] = {} # pid -> command name self._terminate_timeout = terminate_timeout
self._cmds: dict[Process, tuple[str, ...]] = {}
async def spawn( async def spawn(
self, self, *cmd: str, cwd: str | None = None, vital: bool = False
*cmd: str, ) -> Process:
cwd: str | None = None, """Spawn and own a subprocess. If a vital process exits, the group cancels."""
) -> 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 wait( async def run() -> None:
self, name = Path(cmd[0]).stem
*waitables: asyncio.subprocess.Process | Coroutine[Any, Any, Any], logger.info(">>> %s", " ".join([name, *cmd[1:]]))
) -> None: try:
"""Wait for processes/coroutines to complete, raise SystemExit on failure.""" proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd)
self._cmds[proc] = cmd
started.set_result(proc)
except Exception as e: # ruff: ignore[blind-except]
started.set_exception(e)
return
async def wait_proc(proc: asyncio.subprocess.Process) -> None: try:
returncode = await proc.wait() returncode = await proc.wait()
if returncode != 0: finally:
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:
with suppress(ProcessLookupError): with suppress(ProcessLookupError):
p.terminate() proc.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):
try: try:
await asyncio.shield( await asyncio.wait_for(proc.wait(), self._terminate_timeout)
asyncio.wait_for(
asyncio.gather(*[p.wait() for p in still_running]),
timeout=10,
),
)
except TimeoutError: except TimeoutError:
for p in self._procs: with suppress(ProcessLookupError):
if p.returncode is None: proc.kill()
with suppress(ProcessLookupError): await proc.wait()
p.kill()
await p.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:
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 async def http_get_server(url: str, timeout: float) -> str | None: # ruff: ignore[async-function-with-timeout]
"""GET url with plain asyncio streams, return the response Server header. """GET url with plain asyncio streams, return the response Server header.
Returns an empty string when the server responds without a Server header, Returns an empty string when the server responds without a Server header,
@@ -127,42 +106,43 @@ async def http_get_server(url: str, timeout: float) -> str | None: # noqa: ASYN
writer.close() writer.close()
except OSError, EOFError, ValueError, TimeoutError: except OSError, EOFError, ValueError, TimeoutError:
return None 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:"): if line.lower().startswith("server:"):
return line.split(":", 1)[1].strip() return line[7:].strip()
return "" return ""
async def check_ports_free(*urls: str) -> None: 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: Meant to run as a task inside a TaskGroup. Logs the conflict and raises
server = await http_get_server(url, timeout=0.1) 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: if server is not None:
logger.warning( logger.error(
"Conflicting %s already running at %s", server or "server", url "Conflicting %s already running at %s", server or "server", url
) )
raise SystemExit(1) raise RuntimeError(url)
await asyncio.gather(*[check(url) for url in urls])
async def ready(url: str, path: str = "", max_attempts: int = 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. """Wait for the server to be ready by polling an endpoint.
Use empty path to disable the check and make this return immediately. 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: if not path:
return return
for attempt in range(max_attempts): for attempt in range(max_attempts):
if await http_get_server(f"{url}{path}", timeout=1.0) is not None: if await http_get_server(f"{url}{path}", timeout=1.0) is not None:
logger.info(" Backend ready!") logger.info("🟢 Backend ready!")
return return
if attempt == max_attempts - 1: if attempt == max_attempts - 1:
logger.warning("Backend didn't start in time") logger.error("Backend at %s didn't start in time", url)
raise SystemExit(1) raise RuntimeError(url)
await asyncio.sleep(0.1) await asyncio.sleep(0.1)