Vasanko/cista-storage
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update scripts with latest fastapi-vue-setup.

Browse Source
This commit is contained in:
Leo Vasanko
2026-01-30 18:52:54 +00:00
parent 14f2177514
commit 938ece90fe
11 changed files with 467 additions and 306 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cista/__main__.py
+3 -1
View File
@@ -128,7 +128,9 @@ def _main():
auth_mode = args["--auth"]
if auth_mode:
if auth_mode not in ("none", "password", "paskia"):
raise ValueError(f"Invalid auth mode: {auth_mode}. Use: none, password, paskia")
raise ValueError(
f"Invalid auth mode: {auth_mode}. Use: none, password, paskia"
)
settings["authentication"] = auth_mode
elif not exists and not import_droppy:
# We have no users, so make it public
cista/api.py
+1 -3
View File
@@ -3,9 +3,7 @@ import typing
from secrets import token_bytes
import msgspec
from sanic import Blueprint
from sanic import json
from sanic import Blueprint, json
from sanic.exceptions import BadRequest
from cista import __version__, auth, config, watching
cista/sso.py
+3 -1
View File
@@ -18,7 +18,9 @@ from sanic.log import logger
from cista import config
# Auth backend URL for SSO validation (from env with default, no trailing slash)
PASKIA_BACKEND_URL = os.environ.get("PASKIA_BACKEND_URL", "http://localhost:4401").rstrip("/")
PASKIA_BACKEND_URL = os.environ.get(
"PASKIA_BACKEND_URL", "http://localhost:4401"
).rstrip("/")
# Shared httpx client for SSO requests (reused for connection pooling)
_client: httpx.AsyncClient | None = None
frontend/vite-plugin-fastapi.js
+35
View File
@@ -0,0 +1,35 @@
/**
* FastAPI-Vue Vite Plugin
*
* Configures Vite for FastAPI backend integration:
* - Proxies /api/* requests to the FastAPI backend
* - Builds to the Python module's frontend-build directory
*
* Environment variables (with defaults):
* FASTAPI_VUE_BACKEND_URL=http://localhost:5180 - Backend API URL for proxying
*/
const backendUrl = process.env.FASTAPI_VUE_BACKEND_URL || "http://localhost:5180"
export default function fastapiVue({ paths = ["/api"] } = {}) {
// Build proxy configuration for each path
const proxy = {}
for (const path of paths) {
proxy[path] = {
target: backendUrl,
changeOrigin: false,
ws: true,
}
}
return {
name: "fastapi-vite",
config: () => ({
server: { proxy },
build: {
outDir: "../cista/frontend-build",
emptyOutDir: true,
},
}),
}
}
frontend/vite.config.ts
+3 -19
View File
@@ -1,4 +1,5 @@
import { fileURLToPath, URL } from 'node:url'
import fastapiVue from './vite-plugin-fastapi.js'
import { defineConfig } from 'vite'
import vue from '@vitejs/plugin-vue'
@@ -7,15 +8,11 @@ import vue from '@vitejs/plugin-vue'
import svgLoader from 'vite-svg-loader'
import Components from 'unplugin-vue-components/vite'
const dev_backend = {
target: process.env.CISTA_BACKEND_URL || "http://localhost:8989",
changeOrigin: false, // Use frontend "host" to match "origin" from browser
ws: true,
}
// https://vitejs.dev/config/
// Note: fastapiVue() handles proxy and build output (uses FASTAPI_VUE_BACKEND_URL env)
export default defineConfig({
plugins: [
fastapiVue({ paths: ["/api", "/auth", "/files", "/zip", "/preview"] }),
vue(),
svgLoader(), // import svg files
Components(), // auto import components
@@ -33,17 +30,4 @@ export default defineConfig({
'@': fileURLToPath(new URL('./src', import.meta.url))
}
},
server: {
proxy: {
"/api": dev_backend,
"/auth": dev_backend,
"/files": dev_backend,
"/zip": dev_backend,
"/preview": dev_backend,
}
},
build: {
outDir: "../cista/frontend-build",
emptyOutDir: true,
}
})
pyproject.toml
+2 -1
View File
@@ -28,7 +28,8 @@ dependencies = [
"av>=15.0.0",
"blake3>=1.0.5",
"docopt>=0.6.2",
"fastapi-vue>=0.5.0",
"fastapi-vue>=0.5.1",
"fastapi[standard]>=0.128.0",
"html5tagger>=1.3.0",
"httpx>=0.28.0",
"inotify>=0.2.12",
scripts/devserver.py
+68 -171
View File
@@ -2,208 +2,105 @@
"""Run Vite development server for frontend and Cista backend with auto-reload.
Usage:
uv run scripts/devserver.py [-l <listen>]
uv run scripts/devserver.py [frontend] [--backend backend]
Options:
-l LISTEN Listen address for backend (default: from config, or :8000)
frontend Vite frontend endpoint (default: localhost:5173)
--backend Cista backend endpoint (default: from config, or :8000)
Environment:
JS_RUNTIME Path or name of JS runtime to use (deno, npm/node or bun).
"""
import argparse
import asyncio
import contextlib
import os
import sys
from pathlib import Path
from sys import stderr
import httpx
# Import devutil 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 ProcessGroup, logger, ready, setup_vite # type: ignore
from cista import config
from cista.serve import parse_listen
exec((Path(__file__).parent / "fastapi-vue/util.py").read_text("UTF-8")) # noqa: S102
DEFAULT_VITE_PORT = 5173
FRONTEND_PATH = Path(__file__).parent.parent / "frontend"
BUN_BUG = """\
┃ ⚠️ Bun cannot correctly proxy API requests to the backend.
┃ Bug report: https://github.com/oven-sh/bun/issues/9882
┃
┃ Consider using deno or npm instead for development.
"""
DEFAULT_BACKEND_PORT = 8000
def resolve_frontend_tools(vite_port: int) -> tuple[list[str], list[str], str]:
"""Resolve frontend install and dev commands.
def setup_sanic_backend(listen: str | None) -> tuple[str, list[str]]:
"""Parse backend listen address and build cista dev command.
Returns (install_cmd, dev_cmd, tool_name).
Raises SystemExit if tools are not available.
Returns (url, cmd).
"""
if not (FRONTEND_PATH / "package.json").exists():
stderr.write(f"┃ ⚠️ Frontend source not found at {FRONTEND_PATH}\n")
config.load_config()
listen = listen or config.config.listen or f":{DEFAULT_BACKEND_PORT}"
url, opts = parse_listen(listen)
port = opts.get("port", DEFAULT_BACKEND_PORT)
host = opts.get("host", "localhost") or "localhost"
cmd = ["cista", "--dev", "-l", listen]
return f"http://{host}:{port}", cmd
async def run_devserver(frontend: str | None, backend: str | None) -> None:
reporoot = Path(__file__).parent.parent
front = reporoot / "frontend"
if not (front / "package.json").exists():
logger.warning("Frontend source not found at %s", front)
raise SystemExit(1)
install_cmd, build_cmd = find_build_tool() # noqa # type: ignore
dev_cmd, name = find_dev_tool() # noqa # type: ignore
if dev_cmd is None:
if not os.environ.get("JS_RUNTIME"):
stderr.write("┃ ⚠️ deno, npm or bun needed to run the frontend server.\n")
raise SystemExit(1)
frontend_url, npm_install, vite = setup_vite(frontend or "")
backend_url, sanic_cmd = setup_sanic_backend(backend)
dev_cmd = [*dev_cmd, "--clearScreen=false", f"--port={vite_port}"]
# Tell vite where to proxy API requests
os.environ["FASTAPI_VUE_BACKEND_URL"] = backend_url
if name == "bun":
stderr.write(BUN_BUG)
async with ProcessGroup() as pg:
install_proc = await pg.spawn(*npm_install, cwd=str(front))
await asyncio.sleep(0.2) # reduce message overlap
await pg.spawn(*sanic_cmd, cwd=str(reporoot))
return install_cmd, dev_cmd, name
# Wait for both install and backend to be ready
async with asyncio.TaskGroup() as tg:
tg.create_task(pg.wait(install_proc))
tg.create_task(ready(backend_url, path="/api/health?from=devserver.py"))
async def wait_for_backend(host: str, port: int):
"""Wait for the backend to be ready by polling the health endpoint."""
max_attempts = 50
url = f"http://{host}:{port}"
async with httpx.AsyncClient() as client:
for attempt in range(max_attempts):
try:
await client.get(url, timeout=1.0)
stderr.write("✓ Backend ready!\n")
return True
except httpx.RequestError:
if attempt == max_attempts - 1:
stderr.write("┃ ⚠️ Backend didn't start in time\n")
return False
await asyncio.sleep(0.1)
return False
async def _terminate_process(proc: asyncio.subprocess.Process, name: str) -> None:
"""Gracefully terminate a subprocess."""
if proc.returncode is not None:
return
try:
proc.terminate()
except ProcessLookupError:
return
try:
await asyncio.wait_for(proc.wait(), timeout=2)
except TimeoutError:
try:
proc.kill()
except ProcessLookupError:
return
await proc.wait()
async def run_devserver(backend_port: int, cista_args: list[str]) -> None:
"""Run the development server with install, backend, and frontend."""
vite_port = DEFAULT_VITE_PORT
install_cmd, dev_cmd, tool_name = resolve_frontend_tools(vite_port)
# Tell the backend where the Vite dev server is (not used yet)
os.environ["CISTA_DEV_FRONTEND_URL"] = f"http://localhost:{vite_port}"
backend_cmd = ["cista", "--dev", *cista_args]
cwd = str(Path(__file__).parent.parent)
frontend_cwd = str(FRONTEND_PATH)
backend_proc: asyncio.subprocess.Process | None = None
install_proc: asyncio.subprocess.Process | None = None
frontend_proc: asyncio.subprocess.Process | None = None
try:
# Start install (concurrent with backend)
stderr.write(f">>> {tool_name} {' '.join(install_cmd[1:])}\n")
install_proc = await asyncio.create_subprocess_exec(
*install_cmd, cwd=frontend_cwd
)
await asyncio.sleep(0.1)
# Start backend (concurrent with install)
stderr.write(f">>> {' '.join(backend_cmd)}\n")
backend_proc = await asyncio.create_subprocess_exec(*backend_cmd, cwd=cwd)
# Wait for install to complete and backend to be ready
install_task = asyncio.create_task(install_proc.wait(), name="install")
backend_ready_task = asyncio.create_task(
wait_for_backend("localhost", backend_port), name="backend_ready"
)
done, pending = await asyncio.wait(
{install_task, backend_ready_task},
return_when=asyncio.FIRST_COMPLETED,
)
for task in done:
if task.get_name() == "install":
if task.result() != 0:
stderr.write("┃ ⚠️ Install failed\n")
raise SystemExit(1)
elif task.get_name() == "backend_ready" and not task.result():
raise SystemExit(1)
if pending:
done2, _ = await asyncio.wait(pending)
for task in done2:
if task.get_name() == "install":
if task.result() != 0:
stderr.write("┃ ⚠️ Install failed\n")
raise SystemExit(1)
elif task.get_name() == "backend_ready" and not task.result():
raise SystemExit(1)
install_proc = None
# Start Vite dev server
stderr.write(f">>> {tool_name} {' '.join(dev_cmd[1:])}\n")
frontend_proc = await asyncio.create_subprocess_exec(*dev_cmd, cwd=frontend_cwd)
# Wait for either process to exit
done, pending = await asyncio.wait(
{
asyncio.create_task(backend_proc.wait(), name="backend"),
asyncio.create_task(frontend_proc.wait(), name="frontend"),
},
return_when=asyncio.FIRST_COMPLETED,
)
for t in done:
t.result()
for t in pending:
t.cancel()
except asyncio.CancelledError:
stderr.write("\n✓ Shutting down...\n")
finally:
if frontend_proc is not None:
await _terminate_process(frontend_proc, "frontend")
if install_proc is not None:
await _terminate_process(install_proc, "install")
if backend_proc is not None:
await _terminate_process(backend_proc, "backend")
# Start Vite dev server (ProcessGroup waits for any exit, then terminates others)
await pg.spawn(*vite, cwd=str(front))
def main():
# Pass all arguments to cista, parse -l to determine backend port
cista_args = sys.argv[1:]
listen_arg = None
if "-l" in cista_args:
idx = cista_args.index("-l")
if idx + 1 < len(cista_args):
listen_arg = cista_args[idx + 1]
# Load config to get the backend port
config.load_config()
listen = listen_arg or config.config.listen or ":8000"
_, opts = parse_listen(listen)
backend_port = opts.get("port", 8000)
parser = argparse.ArgumentParser(
description="Run Vite and Cista (Sanic) development servers",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=HELP_EPILOG,
)
parser.add_argument(
"frontend",
nargs="?",
metavar="host:port",
help="Vite frontend endpoint (default: localhost:5173)",
)
parser.add_argument(
"--backend",
"-l",
metavar="host:port",
help="Cista backend endpoint (default: from config, or :8000)",
)
args = parser.parse_args()
with contextlib.suppress(KeyboardInterrupt):
asyncio.run(run_devserver(backend_port, cista_args))
asyncio.run(run_devserver(args.frontend, args.backend))
HELP_EPILOG = """
scripts/devserver.py # Default ports
scripts/devserver.py 3000 # Vite on localhost:3000
scripts/devserver.py :3000 --backend 8080 # Vite on *:3000, backend on :8080
JS_RUNTIME environment variable can be used to select the JS runtime
"""
if __name__ == "__main__":
scripts/fastapi-vue/build-frontend.py
+4 -23
View File
@@ -1,34 +1,15 @@
"""Hatch build hook for building Vue frontend during package build."""
import subprocess
import sys
from pathlib import Path
from sys import stderr
from hatchling.builders.hooks.plugin.interface import BuildHookInterface # type: ignore
exec(Path(__file__).with_name("util.py").read_text("UTF-8")) # noqa: S102
def run(cmd, **kwargs):
"""Run a command and display it."""
display_cmd = [Path(cmd[0]).name, *cmd[1:]]
stderr.write(f"### {' '.join(display_cmd)}\n")
subprocess.run(cmd, check=True, **kwargs)
sys.path.insert(0, str(Path(__file__).parent))
from buildutil import build
class CustomBuildHook(BuildHookInterface):
"""Build hook that compiles Vue frontend before packaging."""
def initialize(self, version, build_data):
super().initialize(version, build_data)
stderr.write(">>> Building the frontend\n")
install_cmd, build_cmd = find_build_tool() # noqa # type: ignore
try:
run(install_cmd, cwd="frontend")
stderr.write("\n")
run(build_cmd, cwd="frontend")
except Exception as e:
stderr.write(f"Error occurred while building frontend: {e}\n")
raise
build("frontend")
scripts/fastapi-vue/buildutil.py
+191
View File
@@ -0,0 +1,191 @@
"""Utilities used at build time and in devserver script. No dependencies."""
import logging
import os
import re
import shutil
import subprocess
from pathlib import Path
class _PrefixFormatter(logging.Formatter):
"""Formatter that adds prefix based on log level."""
def format(self, record: logging.LogRecord) -> str:
if record.levelno >= logging.WARNING:
return f"┃ ⚠️ {record.getMessage()}"
return record.getMessage()
_handler = logging.StreamHandler()
_handler.setFormatter(_PrefixFormatter())
logger = logging.getLogger("fastapi-vue")
logger.addHandler(_handler)
logger.setLevel(logging.INFO)
def _check_node_version(node_path: str) -> None:
"""Check if Node.js version is >= 20.
Raises RuntimeError if version is too old or cannot be determined.
"""
try:
result = subprocess.run(
[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:
return
raise RuntimeError(
f"Node.js {version_str} found, but v20+ required (install with nvm)"
)
except (subprocess.CalledProcessError, FileNotFoundError, ValueError):
pass
raise RuntimeError("Could not determine Node.js version")
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.
"""
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")
# 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")
def find_build_tool():
"""Find JavaScript runtime and construct install/build commands.
Returns (install_cmd, build_cmd) tuples of command lists.
Raises RuntimeError if no runtime is found.
"""
install = {
"deno": ("install", "--allow-scripts=npm:vue-demi"),
"npm": ("install",),
"bun": ("--bun", "install"),
}
# Run vite directly for deno to avoid npm-run-all2/run-p issues
build = {
"deno": ("run", "-A", "npm:vite", "build"),
"npm": ("run", "build"),
"bun": ("--bun", "run", "build"),
}
tool, name = find_js_runtime()
return [tool, *install[name]], [tool, *build[name]]
def find_dev_tool() -> list[str]:
"""Find JavaScript runtime and construct dev command.
Returns dev_cmd (without vite-specific args).
Raises RuntimeError if no runtime is found.
"""
dev_args = {
"deno": ("run", "dev", "--"),
"npm": ("--silent", "run", "dev", "--"),
"bun": ("run", "dev", "--"),
}
tool, name = find_js_runtime()
if name == "bun":
logger.warning(
"Bun has a bug in WS proxying (https://github.com/oven-sh/bun/issues/9882). Consider using npm instead."
)
return [tool, *dev_args[name]]
def find_install_tool() -> list[str]:
"""Find JavaScript runtime and construct install command.
Returns install_cmd.
Raises RuntimeError if no runtime is found.
"""
install_args = {
"deno": ("install", "--quiet", "--allow-scripts=npm:vue-demi"),
"npm": ("install", "--silent"),
"bun": ("install", "--silent"),
}
tool, name = find_js_runtime()
return [tool, *install_args[name]]
def build(folder: str = "frontend") -> None:
"""Build the frontend in the specified folder.
Raises SystemExit(1) on failure.
"""
logger.info(">>> Building %s", folder)
try:
install_cmd, build_cmd = find_build_tool()
except RuntimeError as e:
logger.warning(e)
raise SystemExit(1)
def run(cmd):
display_cmd = [Path(cmd[0]).name, *cmd[1:]]
logger.info("### %s", " ".join(display_cmd))
subprocess.run(cmd, check=True, cwd=folder)
try:
run(install_cmd)
logger.info("")
run(build_cmd)
except subprocess.CalledProcessError:
raise SystemExit(1)
scripts/fastapi-vue/devutil.py
+157
View File
@@ -0,0 +1,157 @@
"""Utilities meant for devserver script, used only in source repository with dev deps."""
import asyncio
from pathlib import Path
import httpx
from buildutil import find_dev_tool, find_install_tool, logger
from fastapi_vue.hostutil import parse_endpoint
DEFAULT_VITE_PORT = 5173
DEFAULT_BACKEND_PORT = 5180
class ProcessGroup:
"""Manage async subprocesses with automatic cleanup, like TaskGroup for processes."""
def __init__(self):
self._procs: list[asyncio.subprocess.Process] = []
async def spawn(
self, *cmd: str, cwd: str | None = None
) -> asyncio.subprocess.Process:
"""Spawn a subprocess and track it."""
logger.info(">>> %s", " ".join([Path(cmd[0]).name, *cmd[1:]]))
proc = await asyncio.create_subprocess_exec(*cmd, cwd=cwd)
self._procs.append(proc)
return proc
async def wait(self, proc: asyncio.subprocess.Process) -> None:
"""Wait for a process to complete, raise SystemExit(1) on failure."""
if await proc.wait() != 0:
logger.warning("Command failed")
raise SystemExit(1)
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, *_):
"""Wait for one process to exit, terminate others, then wait for all."""
cleanup_task = asyncio.create_task(self._cleanup())
try:
await asyncio.shield(cleanup_task)
except asyncio.CancelledError:
# Shield was cancelled but cleanup_task continues - wait for it
await cleanup_task
async def _cleanup(self):
running = [p for p in self._procs if p.returncode is None]
if not running:
return
# Wait for any one process to exit
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:
p.terminate()
except ProcessLookupError:
pass
# Wait for all to finish (with overall timeout)
still_running = [p for p in self._procs if p.returncode is None]
if still_running:
try:
await asyncio.wait_for(
asyncio.gather(*[p.wait() for p in still_running]),
timeout=10,
)
except TimeoutError:
for p in self._procs:
if p.returncode is None:
try:
p.kill()
except ProcessLookupError:
pass
await p.wait()
async def ready(url: str, path: str = "") -> None:
"""Wait for the server to be ready by polling an endpoint.
Raises SystemExit(1) if server doesn't start in time.
"""
max_attempts = 50
full_url = f"{url}{path}"
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)
def setup_vite(endpoint: str) -> tuple[str, list[str], list[str]]:
"""Parse frontend endpoint and build commands.
Returns (url, install_cmd, dev_cmd).
Raises SystemExit(1) on invalid config.
"""
endpoints = parse_endpoint(endpoint, DEFAULT_VITE_PORT)
if "uds" in endpoints[0]:
logger.warning("Unix sockets not supported with vite devserver")
raise SystemExit(1)
port = endpoints[0]["port"]
host = endpoints[0]["host"]
install_cmd = find_install_tool()
dev_cmd = find_dev_tool()
if host != "localhost":
dev_cmd.append("--host" if len(endpoints) > 1 else f"--host={host}")
if port != 5173:
dev_cmd.append(f"--port={port}")
return f"http://{host}:{port}", install_cmd, dev_cmd
def setup_fastapi(
endpoint: str, module: str, default_port: int = DEFAULT_BACKEND_PORT
) -> tuple[str, list[str]]:
"""Parse backend endpoint and build fastapi dev command.
Returns (url, cmd).
Raises SystemExit(1) on invalid config.
"""
endpoints = parse_endpoint(endpoint, default_port)
if "uds" in endpoints[0]:
logger.warning("Unix sockets not supported with vite devserver")
raise SystemExit(1)
host = endpoints[0]["host"]
port = endpoints[0]["port"]
cmd = [
"fastapi",
"dev",
"--entrypoint",
module,
"--host",
host,
"--port",
str(port),
]
return f"http://{host}:{port}", cmd
scripts/fastapi-vue/util.py
-87
View File
@@ -1,87 +0,0 @@
"""Shared utilities for build and dev scripts."""
import os
import shutil
from pathlib import Path
from sys import stderr
def find_js_runtime() -> tuple[str, str] | None:
"""Find a JavaScript runtime from JS_RUNTIME env or auto-detect.
Returns (tool_path, tool_name) where tool_name is "deno", "npm", or "bun".
Returns None if no runtime is found.
"""
options = ["deno", "npm", "bun"]
# 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:
stderr.write(f"┃ ⚠️ JS_RUNTIME={js_runtime_env} not found\n")
return None
return tool, option
stderr.write(f"┃ ⚠️ JS_RUNTIME={js_runtime_env} not recognized\n")
return None
# Auto-detect
for option in options:
if tool := shutil.which(option):
return tool, option
return None
def find_build_tool():
"""Find JavaScript runtime and construct install/build commands.
Returns (install_cmd, build_cmd) tuples of command lists.
Raises RuntimeError if no runtime is found.
"""
install = {
"deno": ("install", "--allow-scripts=npm:vue-demi"),
"npm": ("install",),
"bun": ("--bun", "install"),
}
# Run vite directly for deno to avoid npm-run-all2/run-p issues
build = {
"deno": ("run", "-A", "npm:vite", "build"),
"npm": ("run", "build"),
"bun": ("--bun", "run", "build"),
}
result = find_js_runtime()
if result is None:
raise RuntimeError(
"Deno, npm or Bun is required for building but none was found"
)
tool, name = result
return [tool, *install[name]], [tool, *build[name]]
def find_dev_tool():
"""Find JavaScript runtime and construct dev command.
Returns (dev_cmd, tool_name) or (None, None) if not found.
"""
dev_args = {
"deno": ("run", "dev", "--"),
"npm": ("--silent", "run", "dev", "--"),
"bun": ("run", "dev", "--"),
}
result = find_js_runtime()
if result is None:
return None, None
tool, name = result
return [tool, *dev_args[name]], name