LeoVasanko/mediahive
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

fastapi-vue-setup 1.4.1 upgrade, replaces our own access logging and more.

Browse Source
This commit is contained in:
LeoVasanko
2026-09-03 12:19:50 +00:00
parent e5bc736ffa
commit c2776d2e2d
10 changed files with 214 additions and 430 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/devserver.py
Regular → Executable
+15 -7
View File
@@ -9,9 +9,11 @@ import sys
from contextlib import suppress
from pathlib import Path
import tracerite
# 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[import-not-found]
from devutil import (
ProcessGroup,
check_ports_free,
logger,
@@ -22,11 +24,15 @@ from devutil import ( # type: ignore[import-not-found]
DEFAULT_VITE_PORT = 8420
DEFAULT_DEV_PORT = 8421
HEALTH = "/api/health?from=devserver.py"
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():
@@ -36,7 +42,7 @@ async def run_devserver(
viteurl, npm_install, vite = setup_vite(listen, DEFAULT_VITE_PORT)
backurl, mediahive = setup_cli("mediahive", backend, DEFAULT_DEV_PORT)
# Tell the everyone by environment (vite proxy and backend devmode use these)
# Tell everyone via environment (vite proxy and backend devmode use these)
os.environ["MEDIAHIVE_VITE_URL"] = viteurl
os.environ["MEDIAHIVE_BACKEND_URL"] = backurl
os.environ["MEDIAHIVE_DEV"] = "1"
@@ -45,11 +51,13 @@ async def run_devserver(
npm_i = await pg.spawn(*npm_install, cwd=front)
await check_ports_free(viteurl, backurl)
await pg.spawn(*mediahive, *(extra_args or []))
await pg.wait(npm_i, ready(backurl, path="/api/health?from=devserver.py"))
await pg.wait(npm_i, ready(backurl, path=HEALTH))
await pg.spawn(*vite, cwd=front)
def main() -> None:
"""Parse CLI arguments and run the devserver."""
tracerite.load()
parser = argparse.ArgumentParser(
description="Run Vite and FastAPI development servers",
formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -58,12 +66,12 @@ def main() -> None:
parser.add_argument(
"-l",
"--listen",
metavar="host:port",
metavar="addr",
help=f"Vite (default: localhost:{DEFAULT_VITE_PORT})",
)
parser.add_argument(
"--backend",
metavar="host:port",
metavar="addr",
help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})",
)
args, extra_args = parser.parse_known_args()
@@ -72,7 +80,7 @@ def main() -> None:
HELP_EPILOG = """
scripts/devserver.py [args to mediahive]
Other options are forwarded to mediahive [args]
JS_RUNTIME environment variable can be used to select the JS runtime:
npm, deno, bun, or full path to the runtime executable (node maps to npm).
scripts/fastapi-vue/build-frontend.py
-34
View File
@@ -1,34 +0,0 @@
"""Hatch build hook for building Vue frontend during package build."""
import sys
from pathlib import Path
from hatchling.builders.hooks.plugin.interface import ( # type: ignore[import-not-found]
BuildHookInterface,
)
sys.path.insert(0, str(Path(__file__).parent))
from buildutil import build
class CustomBuildHook(BuildHookInterface):
def initialize(self, version, build_data) -> None:
super().initialize(version, build_data)
root = Path(self.root)
frontend_src = root / "frontend"
frontend_build = root / "mediahive" / "frontend-build"
# When building a wheel from sdist, frontend sources may be omitted
# while prebuilt assets are already present in mediahive/frontend-build.
if frontend_src.exists():
build(str(frontend_src))
return
if frontend_build.exists():
return
msg = (
"Frontend build is missing. Expected either source directory "
f"'{frontend_src}' or prebuilt assets in '{frontend_build}'."
)
raise RuntimeError(msg)
scripts/fastapi-vue/buildhook.py
+18
View File
@@ -0,0 +1,18 @@
"""Hatch build hook for building Vue frontend during package build."""
import sys
from pathlib import Path
from hatchling.builders.hooks.plugin.interface import BuildHookInterface
sys.path.insert(0, str(Path(__file__).parent))
from buildutil import build
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")
scripts/fastapi-vue/buildutil.py
+95 -58
View File
@@ -7,6 +7,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."""
@@ -31,81 +33,118 @@ def _check_node_version(node_path: str) -> None:
"""
try:
result = subprocess.run(
[node_path, "--version"], capture_output=True, text=True, check=True
[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,9 +182,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]]
@@ -178,9 +215,9 @@ 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) -> None:
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)
@@ -190,4 +227,4 @@ def build(folder: str = "frontend") -> None:
logger.info("")
run(build_cmd)
except subprocess.CalledProcessError:
raise SystemExit(1)
raise SystemExit(1) from None
scripts/fastapi-vue/devutil.py
+76 -49
View File
@@ -1,33 +1,32 @@
"""Utilities for the devserver script in the source repository.
Used only with development dependencies.
"""
"""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, Self
from typing import TYPE_CHECKING, Any, Self
from urllib.parse import urlsplit
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.
Acts like TaskGroup for processes.
"""
"""Manage async subprocesses with automatic cleanup, like TaskGroup for processes."""
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
@@ -38,7 +37,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."""
@@ -59,18 +59,14 @@ class ProcessGroup:
raise SystemExit(1) from None
async def __aenter__(self) -> Self:
"""Return this process group context manager."""
"""Enter the async context manager."""
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
*_: object,
) -> None:
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:
async def _cleanup(self, *, immediate: bool = False) -> None:
running = [p for p in self._procs if p.returncode is None]
if not running:
return
@@ -98,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:
@@ -108,46 +104,71 @@ class ProcessGroup:
await p.wait()
async def check_ports_free(*urls: str) -> None:
"""Verify URLs are not responding (ports are free).
async def http_get_server(url: str, timeout: float) -> str | None: # noqa: ASYNC109
"""GET url with plain asyncio streams, return the response Server header.
Raise SystemExit if any endpoint responds.
Returns an empty string when the server responds without a Server header,
and None when the server is unreachable or doesn't answer in time.
"""
parts = urlsplit(url)
host = parts.hostname or "localhost"
port = parts.port or (443 if parts.scheme == "https" else 80)
path = parts.path or "/"
if parts.query:
path += f"?{parts.query}"
try:
async with asyncio.timeout(timeout):
reader, writer = await asyncio.open_connection(host, port)
try:
writer.write(f"GET {path} HTTP/1.0\r\nHost: {host}\r\n\r\n".encode())
await writer.drain()
data = await reader.readuntil(b"\r\n\r\n")
finally:
writer.close()
except OSError, EOFError, ValueError, TimeoutError:
return None
for line in data.decode("latin-1").split("\r\n"):
if line.lower().startswith("server:"):
return line.split(":", 1)[1].strip()
return ""
async def check(client: httpx.AsyncClient, url: str) -> None:
with suppress(httpx.RequestError):
res = await client.get(url, timeout=0.1)
server = res.headers.get("server", "server")
logger.warning("Conflicting %s already running at %s", server, url)
async def check_ports_free(*urls: str) -> None:
"""Verify URLs are not responding (ports are free). Raise SystemExit if any respond."""
async def check(url: str) -> None:
server = await http_get_server(url, timeout=0.1)
if server is not None:
logger.warning(
"Conflicting %s already running at %s", server or "server", url
)
raise SystemExit(1)
async with httpx.AsyncClient() as client:
await asyncio.gather(*[check(client, url) for url in urls])
await asyncio.gather(*[check(url) for url in urls])
async def ready(url: str, path: str = "") -> 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.
Raises SystemExit(1) if server doesn't start in time.
"""
max_attempts = 50
full_url = f"{url}{path}"
if not path:
return
async with httpx.AsyncClient() as client:
for attempt in range(max_attempts):
try:
await client.get(full_url, 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)
await asyncio.sleep(0.1)
for attempt in range(max_attempts):
if await http_get_server(f"{url}{path}", timeout=1.0) is not None:
logger.info("✓ Backend ready!")
return
if attempt == max_attempts - 1:
logger.warning("Backend didn't start in time")
raise SystemExit(1)
await asyncio.sleep(0.1)
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.
@@ -173,7 +194,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.
@@ -205,7 +228,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.
@@ -221,5 +246,7 @@ def setup_cli(
host = endpoints[0]["host"]
port = endpoints[0]["port"]
cmd = [cli, f"--listen={host}:{port}"]
# Run the package as a module with the current interpreter, instead of
# relying on a PATH-installed CLI entry point.
cmd = [sys.executable, "-m", cli, f"--listen={host}:{port}"]
return f"http://{host}:{port}", cmd