Compare commits

...
9 Commits
7 changed files with 217 additions and 65 deletions
+17
View File
@@ -12,6 +12,7 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
Returns:
List of dicts with uvicorn bind kwargs (host/port or uds).
Two entries may be returned for IPv4 and IPv6 (all interaces).
Supported forms:
- None or empty -> [{host: "localhost", port: default_port}]
@@ -65,3 +66,19 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
ipaddress.ip_address(host)
return [{"host": host, "port": port}]
def parse_endpoints(
listen: str | list[str] | None = None, default_port: int = 8000
) -> list[dict]:
"""Parse listen strings into a list of endpoint dicts.
Args:
listen: Endpoint string(s) (see parse_endpoint for formats).
default_port: Port to use when not specified in listen args.
"""
if listen is None:
listen = [f"localhost:{default_port}"]
elif isinstance(listen, str):
listen = [listen]
return [ep for s in listen for ep in parse_endpoint(s, default_port)]
+5 -9
View File
@@ -6,7 +6,7 @@ from contextlib import suppress
import uvicorn
from uvicorn import Config, Server
from .hostutil import parse_endpoint
from .hostutil import parse_endpoints
logger = logging.getLogger(__name__)
@@ -30,13 +30,9 @@ def run(
workers: Number of worker processes (requires uvicorn.run, single endpoint only).
**uvicorn_config: Additional uvicorn config options (overrides all other settings).
"""
if listen is None:
listen = [f"localhost:{default_port}"]
elif isinstance(listen, str):
listen = [listen]
endpoints: list[dict] = []
for ep in listen:
endpoints.extend(parse_endpoint(ep, default_port))
endpoints = parse_endpoints(listen, default_port)
if not endpoints:
raise ValueError("No endpoints to serve; check listen configuration")
conf: dict[str, object] = {"app": app, "reload": reload, "workers": workers}
proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1")
@@ -45,7 +41,7 @@ def run(
conf["forwarded_allow_ips"] = proxy
conf.update(uvicorn_config)
with suppress(KeyboardInterrupt):
with suppress(KeyboardInterrupt, asyncio.CancelledError):
if reload or workers:
serve_multiprocess(endpoints, **conf)
else:
+181 -44
View File
@@ -21,6 +21,7 @@ import subprocess
import sys
from pathlib import Path
from textwrap import indent
from typing import Literal
import tomlkit
@@ -38,12 +39,16 @@ def print_boxed(text: str) -> None:
print(f"{'' * width}")
def ruff_format_content(content: str, target_path: Path) -> str:
def ruff_format_content(
content: str, target_path: Path, *, mode: Literal["isort", "full"] = "full"
) -> str:
"""Format Python content using ruff with project settings.
Writes to a temp file (.new.py) next to target, runs ruff check (import sorting)
and ruff format on it, reads back the result, and cleans up.
Writes to a temp file next to target, runs ruff check (import sorting)
and optionally ruff format on it, reads back the result, and cleans up.
Returns the formatted content, or original if ruff fails.
mode='isort' only sorts imports; mode='full' also formats.
"""
temp_file = target_path.with_suffix(".new.py")
try:
@@ -65,6 +70,8 @@ def ruff_format_content(content: str, target_path: Path) -> str:
cwd=target_path.parent,
capture_output=True,
)
if mode == "isort":
return temp_file.read_text("UTF-8")
# Then format
result = subprocess.run(
["uv", "run", "--with", "ruff", "ruff", "format", str(temp_file)],
@@ -103,6 +110,9 @@ def uv_add_packages(
# If vite == dev, dev is incremented by 100
DEFAULT_PORTS = (3100, 3100, 3200)
# Default health check path for devserver backend readiness check
DEFAULT_HEALTH = "/api/health?from=devserver.py"
# Marker comment indicating file can be auto-upgraded
# Users should remove this line to prevent automatic updates
UPGRADE_MARKER = "auto-upgrade@fastapi-vue-setup"
@@ -253,23 +263,20 @@ def find_import_insertion_line(source: str) -> int:
return 2 if source.startswith("#!") else 1
def extract_existing_ports(project_dir: Path) -> tuple[int, int, int] | None:
def extract_existing_ports(
project_dir: Path, main: Path
) -> tuple[int, int, int] | None:
"""Extract existing port configuration from project files.
Returns (default, vite, dev) or None if not found.
"""
module_name = find_module_name(project_dir)
if not module_name:
return None
default_port = None
vite_port = None
dev_port = None
# Try to extract DEFAULT_PORT from __main__.py
main_file = project_dir / module_name / "__main__.py"
if main_file.exists():
content = main_file.read_text("UTF-8")
# Try to extract DEFAULT_PORT from the CLI main module
if main.exists():
content = main.read_text("UTF-8")
match = re.search(r"DEFAULT_PORT\s*=\s*(\d+)", content)
if match:
default_port = int(match.group(1))
@@ -296,6 +303,29 @@ def extract_existing_ports(project_dir: Path) -> tuple[int, int, int] | None:
return None
# Sentinel for "not found" in extract_existing_health
_HEALTH_NOT_FOUND = object()
def extract_existing_health(project_dir: Path) -> str | object:
"""Extract existing health path configuration from devserver.py.
Returns:
- The path string (may be empty to disable)
- _HEALTH_NOT_FOUND sentinel if not found or file doesn't exist
"""
devserver_file = project_dir / "scripts" / "devserver.py"
if not devserver_file.exists():
return _HEALTH_NOT_FOUND
content = devserver_file.read_text("UTF-8")
# Match HEALTH = "/path" or HEALTH = ""
match = re.search(r'^HEALTH\s*=\s*"([^"]*)"', content, re.MULTILINE)
if match:
return match.group(1)
return _HEALTH_NOT_FOUND
def load_template(path: str) -> str:
"""Load a template file from the template directory."""
return (TEMPLATE_DIR / path).read_text("UTF-8")
@@ -427,10 +457,43 @@ def _find_app_in_subpackage(subpkg_dir: Path) -> tuple[Path, str] | None:
return None
def _find_existing_cli_entrypoint(project_dir: Path, module_name: str) -> str | None:
def _add_devmode_to_main(content: str) -> str:
"""Add DEVMODE variable to an existing main module."""
lines = content.splitlines()
# Check if os is imported
has_os_import = any("import os" in line for line in lines)
# Find the first import line
insert_idx = 0
for i, line in enumerate(lines):
stripped = line.strip()
if stripped.startswith("import ") or stripped.startswith("from "):
insert_idx = i + 1
elif stripped and not stripped.startswith("#"):
break
# Insert imports and DEVMODE after existing imports
new_lines = []
if not has_os_import:
new_lines.append("import os")
new_lines.extend(
[
"",
"# Added by fastapi-vue-setup",
'DEVMODE = os.getenv("ENVPREFIX_DEV") == "1"',
"",
]
)
lines[insert_idx:insert_idx] = new_lines
return "\n".join(lines)
def _find_existing_cli_module_path(project_dir: Path, module_name: str) -> str | None:
"""Check if pyproject.toml already has a CLI entrypoint for this module.
Returns the entrypoint string if found, None otherwise.
Returns the module path (e.g., 'module.subpkg.__main__') if found, None otherwise.
"""
pyproject = project_dir / "pyproject.toml"
if not pyproject.exists():
@@ -448,7 +511,10 @@ def _find_existing_cli_entrypoint(project_dir: Path, module_name: str) -> str |
# Look for any script that references our module
for script_name, entry in scripts.items():
if isinstance(entry, str) and entry.startswith(f"{module_name}."):
return f'{script_name} = "{entry}"'
# Extract module path from "module.subpkg.__main__:main"
if ":" in entry:
module_path, _ = entry.rsplit(":", 1)
return module_path
return None
@@ -517,7 +583,7 @@ def render_template(template: str, **kwargs) -> str:
def patch_app_file(
path: Path, module_name: str, app_var: str, dry: bool = False
path: Path, main_module_path: str, app_var: str, dry: bool = False
) -> bool:
"""Patch an existing app.py with frontend integration.
@@ -535,7 +601,7 @@ def patch_app_file(
# Check what's already patched
has_frontend = "from fastapi_vue import Frontend" in content
has_devmode = f"from {module_name}.__main__ import DEVMODE" in content
has_devmode = f"from {main_module_path} import DEVMODE" in content
has_debug_arg = re.search(r"FastAPI\s*\([^)]*debug\s*=", content) is not None
has_lifespan = "await frontend.load()" in content
@@ -550,7 +616,7 @@ def patch_app_file(
if not has_frontend:
imports.extend(["from pathlib import Path", "from fastapi_vue import Frontend"])
if not has_devmode:
imports.append(f"from {module_name}.__main__ import DEVMODE")
imports.append(f"from {main_module_path} import DEVMODE")
if imports:
insert_line = find_import_insertion_line(content)
lines = content.splitlines(keepends=True)
@@ -679,6 +745,8 @@ def patch_app_file(
print(f"✅ Would patch {path}")
return True
# Sort imports only (avoid full formatting of user code)
content = ruff_format_content(content, path, mode="isort")
path.write_text(content, "UTF-8", newline="\n")
print(f"✅ Patched {path}")
@@ -800,7 +868,7 @@ def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool:
target_file = app_vue
if target_file is None:
print("️ No Vue file found to patch, skipping frontend health check")
print("️ No default App.vue found to patch, not adding /api/health check")
return False
original_content = target_file.read_text("UTF-8")
@@ -845,7 +913,7 @@ def patch_frontend_health_check(frontend_dir: Path, dry: bool = False) -> bool:
else:
# Minimal App.vue - only patch if it contains the default welcome message
if "<h1>You did it!</h1>" not in content:
print(f"️ Skipping {target_file} (not a default Vue template)")
print(f"️ Skipping {target_file} (not a Vue demo app)")
return False
# Insert before the </p> tag
template_end = content.find("</template>")
@@ -987,6 +1055,7 @@ def _write_fallback_file(
existing_content = fallback_path.read_text("UTF-8")
if existing_content == content:
print(f"✔️ {fallback_path} (already up to date)")
_new_files_written.append((fallback_path, original_path))
return False
if dry:
@@ -1247,13 +1316,28 @@ def cmd_setup(args: argparse.Namespace) -> int:
else:
print(f"📦 Module: {module_name}")
module_dir = project_dir / module_name
scripts_dir = project_dir / "scripts"
fastapi_vue_scripts = scripts_dir / "fastapi-vue"
# Check if project already has a CLI entrypoint in pyproject.toml
existing_cli_module = _find_existing_cli_module_path(project_dir, module_name)
# Determine main module path for DEVMODE import
main_module_path = existing_cli_module or f"{module_name}.__main__"
if existing_cli_module:
print(f"️ Using existing CLI: {existing_cli_module}")
# Resolve the CLI main module to a file path
main = project_dir / Path(*main_module_path.split(".")).with_suffix(".py")
# Determine port configuration
# Priority: --ports argument > existing project values > defaults
if args.ports:
default_port, vite_port, dev_port = parse_ports(args.ports)
ports_note = "(--ports)"
else:
existing_ports = extract_existing_ports(project_dir)
existing_ports = extract_existing_ports(project_dir, main)
if existing_ports:
default_port, vite_port, dev_port = existing_ports
ports_note = "(kept for upgrade)"
@@ -1265,9 +1349,35 @@ def cmd_setup(args: argparse.Namespace) -> int:
f"📡 Ports: default={default_port}, vite={vite_port}, dev={dev_port} {ports_note}"
)
# Determine health path configuration
# Priority: --health argument > existing project value > default
if args.health is not None:
health = args.health
health_note = "(--health)" if health else "(disabled via --health)"
else:
existing_health = extract_existing_health(project_dir)
if existing_health is not _HEALTH_NOT_FOUND:
# Explicitly configured (path string, possibly empty to disable)
health = existing_health
health_note = "(kept for upgrade)"
else:
# Not found - use default
health = DEFAULT_HEALTH
health_note = "(--health to override)"
if health:
print(f"🏥 Health check: {health} {health_note}")
else:
print(f"🏥 Health check: disabled {health_note}")
# Title for templates
project_title = module_name.replace("_", " ").title()
# Find existing FastAPI app
app_info = (
find_fastapi_app(module_dir, project_dir) if module_dir.exists() else None
)
# Template variables
tpl_vars = {
"MODULE_NAME": module_name,
@@ -1275,30 +1385,25 @@ def cmd_setup(args: argparse.Namespace) -> int:
"TEMPLATE_DEFAULT_PORT": str(default_port),
"TEMPLATE_VITE_PORT": str(vite_port),
"TEMPLATE_DEV_PORT": str(dev_port),
"TEMPLATE_HEALTH": f'"{health}"',
"ENVPREFIX": module_name.upper(),
"PROJECT_CLI": module_name,
"MAIN_MODULE": main_module_path,
}
module_dir = project_dir / module_name
scripts_dir = project_dir / "scripts"
fastapi_vue_scripts = scripts_dir / "fastapi-vue"
# Find existing FastAPI app
app_info = (
find_fastapi_app(module_dir, project_dir) if module_dir.exists() else None
)
if app_info:
app_file, app_var = app_info
print(f"📍 Found FastAPI app: {app_var} in {app_file.name}")
tpl_vars["APP_VAR"] = app_var
tpl_vars["APP_MODULE"] = app_file.stem
# Dotted module path relative to project dir (e.g. "paskia.fastapi.mainapp")
app_module = ".".join(app_file.relative_to(project_dir).with_suffix("").parts)
tpl_vars["APP_MODULE"] = app_module
else:
print("📍 No existing FastAPI app found, will create new one")
app_file = None
app_var = "app"
tpl_vars["APP_VAR"] = app_var
tpl_vars["APP_MODULE"] = "app"
tpl_vars["APP_MODULE"] = f"{module_name}.app"
# Create directories
if not dry:
@@ -1351,7 +1456,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
# === Handle app module ===
if app_file:
# Existing app: patch with import, route, and try to patch lifespan
patch_app_file(app_file, module_name, app_var, dry=dry)
patch_app_file(app_file, main_module_path, app_var, dry=dry)
else:
# No app: create full app.py
# Create __init__.py if missing
@@ -1368,14 +1473,16 @@ def cmd_setup(args: argparse.Namespace) -> int:
write_file(app_file_path, content, overwrite=False, dry=dry)
# === Handle __main__.py ===
if existing_cli_module:
# Existing CLI entrypoint - write .new.py beside the existing main module
main_fallback = main.with_suffix(".new.py")
else:
main_fallback = module_dir / "__main__.new.py"
main_file = module_dir / "__main__.py"
main_fallback = module_dir / "__main__.new.py"
template = load_template("backend/__main__.py")
main_content = render_template(template, **tpl_vars)
# Check if project already has a CLI entrypoint in pyproject.toml
existing_cli = _find_existing_cli_entrypoint(project_dir, module_name)
if main_file.exists():
# File exists: update if it has the auto-upgrade marker, otherwise use fallback
write_file(
@@ -1385,7 +1492,7 @@ def cmd_setup(args: argparse.Namespace) -> int:
dry=dry,
fallback_path=main_fallback,
)
elif not existing_cli:
elif not existing_cli_module:
# No file and no existing entrypoint: create new __main__.py
write_file(
main_file,
@@ -1394,8 +1501,29 @@ def cmd_setup(args: argparse.Namespace) -> int:
dry=dry,
)
else:
# No file but has existing entrypoint - don't create (user has custom CLI setup)
print(f"️ Skipping __main__.py (package already has CLI: {existing_cli})")
# Existing CLI entrypoint: write our template as .new.py beside the existing module
# and also patch the existing module with DEVMODE if needed
_write_fallback_file(
main,
main_fallback,
main_content,
dry=dry,
executable=False,
)
if main.exists():
content = main.read_text("UTF-8")
if "DEVMODE" not in content:
new_content = _add_devmode_to_main(content)
new_file = main.with_suffix(".new.py")
_write_fallback_file(
main,
new_file,
new_content,
dry=dry,
executable=False,
)
else:
print(f"⚠️ Existing CLI main module {main} not found")
# === Update vite.config.js/ts ===
frontend_dir = project_dir / "frontend"
@@ -1493,12 +1621,16 @@ def cmd_setup(args: argparse.Namespace) -> int:
# Show merge note if any .new.py files were written
if _new_files_written:
print()
print(
"⚠️ Note: Some files could not be auto-upgraded because you customized them."
)
print("⚠️ Some files could not be auto-upgraded because you customized them.")
print(" Please manually merge the following files:")
for new_file, original_file in _new_files_written:
print(f"{new_file.name}{original_file.name}")
try:
new_rel = new_file.relative_to(project_dir)
orig_rel = original_file.relative_to(project_dir)
except ValueError:
new_rel = new_file.name
orig_rel = original_file.name
print(f"{new_rel}{orig_rel}")
print()
# Clear the list for potential subsequent runs
_new_files_written.clear()
@@ -1551,6 +1683,11 @@ Examples:
metavar="BACKEND,VITE,DEV",
help="Port configuration as comma-separated values (default: 3100,3100,3200)",
)
parser.add_argument(
"--health",
metavar="PATH",
help="Health check path for devserver (default: /api/health?from=devserver.py, '' to disable)",
)
parser.add_argument(
"--dry", "--dry-run", action="store_true", help="Show what would be done"
)
+3 -3
View File
@@ -17,12 +17,12 @@ def main():
help=(f"Endpoint (default: localhost:{DEFAULT_PORT})."),
)
args = parser.parse_args()
dev = {"reload": True, "reload_dirs": ["MODULE_NAME"]}
dev = {"reload": True, "reload_dirs": ["paskia"]} if DEVMODE else {}
server.run(
"MODULE_NAME.APP_MODULE:APP_VAR",
"APP_MODULE:APP_VAR",
listen=args.listen,
default_port=DEFAULT_PORT,
**(dev if DEVMODE else {}),
**dev,
)
+1 -1
View File
@@ -3,7 +3,7 @@ from pathlib import Path
from fastapi import FastAPI
from fastapi_vue import Frontend
from MODULE_NAME.__main__ import DEVMODE
from MAIN_MODULE import DEVMODE
# Vue Frontend static files
frontend = Frontend(Path(__file__).with_name("frontend-build"))
+5 -4
View File
@@ -22,6 +22,7 @@ from devutil import ( # type: ignore
DEFAULT_VITE_PORT = TEMPLATE_VITE_PORT
DEFAULT_DEV_PORT = TEMPLATE_DEV_PORT
HEALTH = TEMPLATE_HEALTH
async def run_devserver(
@@ -45,7 +46,7 @@ async def run_devserver(
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.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)
@@ -58,12 +59,12 @@ def main():
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 +73,7 @@ def main():
HELP_EPILOG = """
scripts/devserver.py [args to PROJECT_CLI]
Other options are forwarded to PROJECT_CLI [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).
+5 -4
View File
@@ -111,18 +111,19 @@ async def check_ports_free(*urls: str) -> None:
await asyncio.gather(*[check(client, url) for url in urls])
async def ready(url: str, path: str = "") -> None:
async def ready(url: str, path: str = "", max_attempts=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)
await client.get(f"{url}{path}", timeout=1.0)
logger.info("✓ Backend ready!")
return
except httpx.RequestError: