269 lines
8.6 KiB
Python
269 lines
8.6 KiB
Python
#!/usr/bin/env -S uv run
|
|
"""Run Vite development server for frontend and FastAPI backend with auto-reload.
|
|
|
|
Usage:
|
|
uv run scripts/devserver.py [host:port]
|
|
|
|
The optional host:port argument sets where the Vite frontend listens.
|
|
Supported forms: host[:port], :port (all interfaces), or just port.
|
|
Backend always listens on localhost:{{BACKEND_PORT}}.
|
|
|
|
Environment:
|
|
JS_RUNTIME Path or name of JS runtime to use (deno, npm/node or bun).
|
|
FASTAPI_VUE_FRONTEND_URL Set by this script for the backend to know where Vite is.
|
|
"""
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import ipaddress
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from sys import stderr
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
|
|
exec((Path(__file__).parent / "fastapi-vue/util.py").read_text("UTF-8")) # noqa: S102
|
|
|
|
DEFAULT_HOST = "localhost"
|
|
DEFAULT_VITE_PORT = {{VITE_PORT}}
|
|
BACKEND_PORT = {{BACKEND_PORT}}
|
|
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.
|
|
"""
|
|
|
|
|
|
def parse_endpoint(value: str | None) -> tuple[str | None, int, bool]:
|
|
"""Parse an endpoint for Vite (no unix socket support).
|
|
|
|
Returns (host, port, all_ifaces).
|
|
"""
|
|
if not value:
|
|
return DEFAULT_HOST, DEFAULT_VITE_PORT, False
|
|
|
|
# Port only (numeric) -> localhost:port
|
|
if value.isdigit():
|
|
return DEFAULT_HOST, int(value), False
|
|
|
|
# Leading colon :port -> bind all interfaces
|
|
if value.startswith(":") and value != ":":
|
|
port_part = value[1:]
|
|
if not port_part.isdigit():
|
|
raise SystemExit(f"Invalid port in '{value}'")
|
|
return None, int(port_part), True
|
|
|
|
# Unbracketed IPv6 (cannot safely contain a port)
|
|
if value.count(":") > 1 and not value.startswith("["):
|
|
try:
|
|
ipaddress.IPv6Address(value)
|
|
except ValueError as e:
|
|
raise SystemExit(f"Invalid IPv6 address '{value}': {e}") from e
|
|
return value, DEFAULT_VITE_PORT, False
|
|
|
|
# Use urllib.parse for everything else
|
|
parsed = urlparse(f"//{value}")
|
|
host = parsed.hostname or DEFAULT_HOST
|
|
port = parsed.port or DEFAULT_VITE_PORT
|
|
|
|
return host, port, False
|
|
|
|
|
|
def resolve_frontend_tools(
|
|
vite_host: str | None, vite_port: int, all_ifaces: bool
|
|
) -> tuple[list[str], list[str], str]:
|
|
"""Resolve frontend install and dev commands.
|
|
|
|
Returns (install_cmd, dev_cmd, tool_name).
|
|
Raises SystemExit if tools are not available.
|
|
"""
|
|
if not (FRONTEND_PATH / "package.json").exists():
|
|
stderr.write(f"┃ ⚠️ Frontend source not found at {FRONTEND_PATH}\n")
|
|
raise SystemExit(1)
|
|
|
|
result = find_js_runtime()
|
|
if result 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)
|
|
|
|
tool, name = result
|
|
|
|
install_args = {
|
|
"deno": ("install", "--quiet", "--allow-scripts=npm:vue-demi"),
|
|
"npm": ("install", "--silent"),
|
|
"bun": ("install", "--silent"),
|
|
}
|
|
dev_args = {
|
|
"deno": ("run", "dev", "--"),
|
|
"npm": ("--silent", "run", "dev", "--"),
|
|
"bun": ("run", "dev", "--"),
|
|
}
|
|
|
|
install_cmd = [tool, *install_args[name]]
|
|
dev_cmd = [tool, *dev_args[name], "--port", str(vite_port)]
|
|
|
|
if all_ifaces:
|
|
dev_cmd.append("--host")
|
|
elif vite_host:
|
|
dev_cmd.extend(["--host", vite_host])
|
|
|
|
if name == "bun":
|
|
stderr.write(BUN_BUG)
|
|
|
|
return install_cmd, dev_cmd, name
|
|
|
|
|
|
async def wait_for_backend():
|
|
"""Wait for the backend to be ready by polling the health endpoint."""
|
|
max_attempts = 50
|
|
async with httpx.AsyncClient() as client:
|
|
for attempt in range(max_attempts):
|
|
try:
|
|
await client.get(f"http://localhost:{BACKEND_PORT}", 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(
|
|
vite_host: str | None, vite_port: int, all_ifaces: bool
|
|
) -> None:
|
|
"""Run the development server with install, backend, and frontend."""
|
|
install_cmd, dev_cmd, tool_name = resolve_frontend_tools(
|
|
vite_host, vite_port, all_ifaces
|
|
)
|
|
|
|
# Tell the backend where the Vite dev server is
|
|
os.environ["FASTAPI_VUE_FRONTEND_URL"] = (
|
|
f"http://{vite_host or 'localhost'}:{vite_port}"
|
|
)
|
|
# Tell Vite where the backend is (for proxying /api requests)
|
|
os.environ["VITE_BACKEND_URL"] = f"http://localhost:{BACKEND_PORT}"
|
|
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)
|
|
backend_cmd = [
|
|
"uvicorn",
|
|
"{{MODULE_NAME}}.app:app",
|
|
"--host",
|
|
"localhost",
|
|
"--port",
|
|
str(BACKEND_PORT),
|
|
"--reload",
|
|
]
|
|
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(), 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")
|
|
|
|
|
|
def main():
|
|
hostport = sys.argv[1] if len(sys.argv) > 1 else None
|
|
vite_host, vite_port, all_ifaces = parse_endpoint(hostport)
|
|
|
|
with contextlib.suppress(KeyboardInterrupt):
|
|
asyncio.run(run_devserver(vite_host, vite_port, all_ifaces))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|