456 lines
17 KiB
Python
456 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import importlib.util
|
|
from dataclasses import dataclass
|
|
import inspect
|
|
import os
|
|
import sys
|
|
from collections.abc import Iterable, Sequence
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy.engine import make_url
|
|
from sqlalchemy.exc import ArgumentError
|
|
|
|
from govoplan_core.core.discovery import iter_module_entry_points
|
|
from govoplan_core.core.module_management import load_startup_enabled_modules, startup_candidate_module_ids
|
|
from govoplan_core.core.modules import ModuleManifest
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_core.db.session import configure_database
|
|
from govoplan_core.server.config import GovoplanServerConfig, import_object, load_server_config
|
|
from govoplan_core.server.registry import available_module_manifests, build_platform_registry
|
|
|
|
DEFAULT_APP = "govoplan_core.server.app:app"
|
|
DEFAULT_CONFIG = "govoplan_core.server.default_config:get_server_config"
|
|
DEFAULT_DEV_DATABASE_URL = "postgresql+psycopg://govoplan_dev@127.0.0.1:5432/govoplan_dev"
|
|
DEFAULT_DEV_DATABASE_URL_PGTOOLS = "postgresql://govoplan_dev@127.0.0.1:5432/govoplan_dev"
|
|
DEFAULT_SQLITE_DATABASE_NAME = "govoplan-dev.db"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class DevserverState:
|
|
config_path: str | None
|
|
runtime_root: Path | None
|
|
database_url: str
|
|
bootstrap_db_path: Path | None
|
|
config: GovoplanServerConfig
|
|
registry: PlatformRegistry
|
|
reload_dirs: list[str]
|
|
reload_module_ids: tuple[str, ...] | None
|
|
|
|
|
|
def _config_module_runtime_root(config_path: str | None) -> Path | None:
|
|
object_path = config_path or os.getenv("GOVOPLAN_SERVER_CONFIG")
|
|
if not object_path:
|
|
return None
|
|
module_name = object_path.split(":", 1)[0]
|
|
try:
|
|
spec = importlib.util.find_spec(module_name)
|
|
except (ImportError, AttributeError, ValueError):
|
|
return None
|
|
if spec is None or not spec.origin:
|
|
return None
|
|
|
|
source = Path(spec.origin).resolve()
|
|
if source.name == "__init__.py":
|
|
package_dir = source.parent
|
|
else:
|
|
package_dir = source.parent
|
|
if package_dir.name == "app":
|
|
return package_dir.parent
|
|
return None
|
|
|
|
|
|
def apply_runtime_defaults(config_path: str | None) -> Path | None:
|
|
runtime_root = _config_module_runtime_root(config_path) or Path.cwd().resolve()
|
|
(runtime_root / "runtime").mkdir(parents=True, exist_ok=True)
|
|
(runtime_root / "runtime" / "files").mkdir(parents=True, exist_ok=True)
|
|
(runtime_root / "runtime" / "mock-mailbox").mkdir(parents=True, exist_ok=True)
|
|
|
|
dev_database_backend = os.getenv("GOVOPLAN_DEV_DATABASE_BACKEND", "postgres").strip().lower()
|
|
default_database_url = DEFAULT_DEV_DATABASE_URL
|
|
default_pgtools_url = DEFAULT_DEV_DATABASE_URL_PGTOOLS
|
|
if dev_database_backend == "sqlite":
|
|
default_database_url = f"sqlite:///{runtime_root / 'runtime' / DEFAULT_SQLITE_DATABASE_NAME}"
|
|
default_pgtools_url = ""
|
|
|
|
defaults = {
|
|
"DATABASE_URL": os.getenv("GOVOPLAN_DEV_DATABASE_URL", default_database_url),
|
|
"DEV_BOOTSTRAP_ENABLED": "true",
|
|
"FILE_STORAGE_LOCAL_ROOT": str(runtime_root / "runtime" / "files"),
|
|
"MOCK_MAILBOX_DIR": str(runtime_root / "runtime" / "mock-mailbox"),
|
|
}
|
|
for key, value in defaults.items():
|
|
os.environ.setdefault(key, value)
|
|
if default_pgtools_url:
|
|
os.environ.setdefault("GOVOPLAN_DATABASE_URL_PGTOOLS", os.getenv("GOVOPLAN_DEV_DATABASE_URL_PGTOOLS", default_pgtools_url))
|
|
return runtime_root
|
|
|
|
|
|
def _sqlite_database_path(database_url: str) -> Path | None:
|
|
if not database_url.startswith("sqlite:///"):
|
|
return None
|
|
raw_path = database_url[len("sqlite:///"):]
|
|
if not raw_path or raw_path == ":memory:":
|
|
return None
|
|
path = Path(raw_path).expanduser()
|
|
return path if path.is_absolute() else Path.cwd() / path
|
|
|
|
|
|
def validate_sqlite_database_url(database_url: str) -> None:
|
|
db_path = _sqlite_database_path(database_url)
|
|
if db_path is None:
|
|
return
|
|
if db_path.name.startswith(".fuse"):
|
|
raise SystemExit(
|
|
"DATABASE_URL points at a FUSE-hidden SQLite file. "
|
|
"Stop the process still holding the deleted database, unset DATABASE_URL, "
|
|
"or set DATABASE_URL to a normal .db path under the core runtime directory."
|
|
)
|
|
if not db_path.parent.exists():
|
|
raise SystemExit(
|
|
f"DATABASE_URL points to {db_path}, but its parent directory does not exist. "
|
|
"Unset stale DATABASE_URL or set it to "
|
|
f"sqlite:///{Path.cwd().resolve() / 'runtime' / DEFAULT_SQLITE_DATABASE_NAME}."
|
|
)
|
|
|
|
|
|
def redacted_database_url(database_url: str) -> str:
|
|
try:
|
|
return make_url(database_url).render_as_string(hide_password=True)
|
|
except (ArgumentError, TypeError, ValueError):
|
|
scheme = database_url.partition(":")[0].strip()
|
|
return f"{scheme}:<redacted>" if scheme else "<redacted>"
|
|
|
|
|
|
def _env_truthy(value: str | None) -> bool:
|
|
return value is not None and value.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def enable_dev_bootstrap_for_missing_sqlite(database_url: str) -> Path | None:
|
|
db_path = _sqlite_database_path(database_url)
|
|
if db_path is None or db_path.name.startswith(".fuse") or not db_path.parent.exists():
|
|
return None
|
|
if db_path.exists() and db_path.stat().st_size > 0:
|
|
return None
|
|
os.environ.setdefault("DEV_BOOTSTRAP_ENABLED", "true")
|
|
return db_path
|
|
|
|
|
|
def apply_detected_dev_bootstrap(config: GovoplanServerConfig, bootstrap_db_path: Path | None) -> None:
|
|
if bootstrap_db_path is None or not _env_truthy(os.getenv("DEV_BOOTSTRAP_ENABLED")):
|
|
return
|
|
if config.settings is not None and hasattr(config.settings, "dev_bootstrap_enabled"):
|
|
setattr(config.settings, "dev_bootstrap_enabled", True)
|
|
|
|
|
|
def _source_root_for_path(path: str | os.PathLike[str]) -> Path | None:
|
|
source_path = Path(path).resolve()
|
|
directory = source_path if source_path.is_dir() else source_path.parent
|
|
if not directory.exists():
|
|
return None
|
|
|
|
for candidate in (directory, *directory.parents):
|
|
if candidate.name == "src":
|
|
return candidate
|
|
return directory
|
|
|
|
|
|
def _source_roots_for_object(value: Any) -> tuple[Path, ...]:
|
|
candidates: list[Any] = [value]
|
|
if not inspect.isclass(value) and not inspect.ismodule(value):
|
|
candidates.append(type(value))
|
|
|
|
roots: list[Path] = []
|
|
for candidate in candidates:
|
|
try:
|
|
source = inspect.getsourcefile(candidate) or inspect.getfile(candidate)
|
|
except (TypeError, OSError):
|
|
continue
|
|
root = _source_root_for_path(source)
|
|
if root is not None:
|
|
roots.append(root)
|
|
return tuple(roots)
|
|
|
|
|
|
def _manifest_source_roots(manifest: ModuleManifest) -> tuple[Path, ...]:
|
|
roots: list[Path] = []
|
|
|
|
if manifest.route_factory is not None:
|
|
roots.extend(_source_roots_for_object(manifest.route_factory))
|
|
|
|
if manifest.migration_spec is not None and manifest.migration_spec.script_location:
|
|
root = _source_root_for_path(manifest.migration_spec.script_location)
|
|
if root is not None:
|
|
roots.append(root)
|
|
|
|
for provider in manifest.resource_acl_providers:
|
|
roots.extend(_source_roots_for_object(provider))
|
|
|
|
for provider in manifest.tenant_summary_providers:
|
|
roots.extend(_source_roots_for_object(provider))
|
|
|
|
for provider in manifest.tenant_summary_batch_providers:
|
|
roots.extend(_source_roots_for_object(provider))
|
|
|
|
for providers in manifest.delete_veto_providers.values():
|
|
for provider in providers:
|
|
roots.extend(_source_roots_for_object(provider))
|
|
|
|
return tuple(roots)
|
|
|
|
|
|
def _entry_point_source_roots(enabled_module_ids: set[str]) -> tuple[Path, ...]:
|
|
roots: list[Path] = []
|
|
for entry_point in iter_module_entry_points():
|
|
loaded = entry_point.load()
|
|
manifest = loaded() if callable(loaded) else loaded
|
|
if not isinstance(manifest, ModuleManifest) or manifest.id not in enabled_module_ids:
|
|
continue
|
|
roots.extend(_source_roots_for_object(loaded))
|
|
return tuple(roots)
|
|
|
|
|
|
def _config_source_roots(config_path: str | None) -> tuple[Path, ...]:
|
|
object_path = config_path or os.getenv("GOVOPLAN_SERVER_CONFIG") or DEFAULT_CONFIG
|
|
try:
|
|
loaded = import_object(object_path)
|
|
except ModuleNotFoundError:
|
|
if config_path or os.getenv("GOVOPLAN_SERVER_CONFIG"):
|
|
raise
|
|
loaded = import_object(DEFAULT_CONFIG)
|
|
return _source_roots_for_object(loaded)
|
|
|
|
|
|
def unique_existing_dirs(paths: Iterable[Path | str]) -> list[str]:
|
|
seen: set[Path] = set()
|
|
result: list[str] = []
|
|
for item in paths:
|
|
path = Path(item).resolve()
|
|
if path in seen or not path.is_dir():
|
|
continue
|
|
seen.add(path)
|
|
result.append(str(path))
|
|
return result
|
|
|
|
|
|
def build_reload_dirs(
|
|
config: GovoplanServerConfig,
|
|
*,
|
|
config_path: str | None = None,
|
|
registry: PlatformRegistry | None = None,
|
|
extra_dirs: Sequence[str] = (),
|
|
module_ids: Sequence[str] | None = None,
|
|
) -> list[str]:
|
|
active_registry = registry or build_platform_registry(config.enabled_modules, manifest_factories=config.manifest_factories)
|
|
manifests = active_registry.manifests()
|
|
enabled_module_ids = {manifest.id for manifest in manifests}
|
|
selected_module_ids = (
|
|
enabled_module_ids
|
|
if module_ids is None
|
|
else {module_id.strip() for module_id in module_ids if module_id.strip()}
|
|
)
|
|
unknown_module_ids = selected_module_ids - enabled_module_ids
|
|
if unknown_module_ids:
|
|
raise SystemExit(
|
|
"Reload modules are not enabled: "
|
|
+ ", ".join(sorted(unknown_module_ids))
|
|
)
|
|
|
|
roots: list[Path | str] = []
|
|
roots.extend(_config_source_roots(config_path))
|
|
|
|
for factory in config.manifest_factories:
|
|
try:
|
|
manifest = factory()
|
|
except TypeError:
|
|
manifest = None
|
|
if (
|
|
module_ids is not None
|
|
and isinstance(manifest, ModuleManifest)
|
|
and manifest.id not in selected_module_ids
|
|
):
|
|
continue
|
|
roots.extend(_source_roots_for_object(factory))
|
|
|
|
roots.extend(_entry_point_source_roots(selected_module_ids))
|
|
|
|
for manifest in manifests:
|
|
if manifest.id not in selected_module_ids:
|
|
continue
|
|
roots.extend(_manifest_source_roots(manifest))
|
|
|
|
roots.extend(extra_dirs)
|
|
return unique_existing_dirs(roots)
|
|
|
|
|
|
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Run the GovOPlaN development server with module-aware reload directories.")
|
|
parser.add_argument("--config", help="GovoplanServerConfig object path, e.g. app.govoplan_config:get_server_config.")
|
|
parser.add_argument("--app", default=DEFAULT_APP, help=f"ASGI app import string passed to uvicorn. Default: {DEFAULT_APP}.")
|
|
parser.add_argument("--host", default="127.0.0.1", help="Host to bind. Default: 127.0.0.1.")
|
|
parser.add_argument("--port", type=int, default=8000, help="Port to bind. Default: 8000.")
|
|
parser.add_argument("--no-reload", action="store_true", help="Disable uvicorn reload.")
|
|
parser.add_argument("--reload-dir", action="append", default=[], help="Additional directory to watch. May be passed multiple times.")
|
|
reload_scope = parser.add_mutually_exclusive_group()
|
|
reload_scope.add_argument(
|
|
"--reload-module",
|
|
action="append",
|
|
default=None,
|
|
metavar="MODULE_ID",
|
|
help=(
|
|
"Watch only this enabled module in addition to core/config sources. "
|
|
"May be passed multiple times."
|
|
),
|
|
)
|
|
reload_scope.add_argument(
|
|
"--reload-core-only",
|
|
action="store_true",
|
|
help="Watch core/config sources but no optional module source trees.",
|
|
)
|
|
parser.add_argument("--smoke", action="store_true", help="Prepare runtime paths, run app startup, print effective paths, and exit without uvicorn.")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def prepare_devserver(
|
|
config_path: str | None,
|
|
*,
|
|
extra_reload_dirs: Sequence[str] = (),
|
|
reload_module_ids: Sequence[str] | None = None,
|
|
) -> DevserverState:
|
|
runtime_root = apply_runtime_defaults(config_path)
|
|
database_url = os.getenv("DATABASE_URL", "")
|
|
validate_sqlite_database_url(database_url)
|
|
bootstrap_db_path = enable_dev_bootstrap_for_missing_sqlite(database_url)
|
|
config = load_server_config(config_path)
|
|
database_url = str(getattr(config.settings, "database_url", database_url) or "")
|
|
validate_sqlite_database_url(database_url)
|
|
if database_url:
|
|
configure_database(database_url)
|
|
if bootstrap_db_path is None:
|
|
bootstrap_db_path = enable_dev_bootstrap_for_missing_sqlite(database_url)
|
|
apply_detected_dev_bootstrap(config, bootstrap_db_path)
|
|
raw_enabled_modules = load_startup_enabled_modules(config.enabled_modules)
|
|
candidate_modules = startup_candidate_module_ids(config.enabled_modules, raw_enabled_modules)
|
|
available_modules = available_module_manifests(
|
|
config.manifest_factories,
|
|
enabled_modules=candidate_modules,
|
|
ignore_load_errors=True,
|
|
)
|
|
enabled_modules = load_startup_enabled_modules(config.enabled_modules, available=available_modules)
|
|
registry = build_platform_registry(enabled_modules, manifest_factories=config.manifest_factories)
|
|
reload_dirs = build_reload_dirs(
|
|
config,
|
|
config_path=config_path,
|
|
registry=registry,
|
|
extra_dirs=extra_reload_dirs,
|
|
module_ids=reload_module_ids,
|
|
)
|
|
return DevserverState(
|
|
config_path=config_path,
|
|
runtime_root=runtime_root,
|
|
database_url=database_url,
|
|
bootstrap_db_path=bootstrap_db_path,
|
|
config=config,
|
|
registry=registry,
|
|
reload_dirs=reload_dirs,
|
|
reload_module_ids=(
|
|
None
|
|
if reload_module_ids is None
|
|
else tuple(sorted(set(reload_module_ids)))
|
|
),
|
|
)
|
|
|
|
|
|
def print_devserver_summary(state: DevserverState, *, app: str, no_reload: bool) -> None:
|
|
print(f"Starting GovOPlaN dev server: {app}")
|
|
print(f"Config: {state.config_path or DEFAULT_CONFIG}")
|
|
print(f"Runtime root: {state.runtime_root}")
|
|
if state.database_url:
|
|
print(f"Database: {redacted_database_url(state.database_url)}")
|
|
if state.database_url.startswith("postgresql"):
|
|
pgtools_url = os.getenv("GOVOPLAN_DATABASE_URL_PGTOOLS")
|
|
if pgtools_url:
|
|
print(f"PostgreSQL tools URL: {redacted_database_url(pgtools_url)}")
|
|
if state.bootstrap_db_path is not None:
|
|
bootstrap_state = "enabled" if getattr(state.config.settings, "dev_bootstrap_enabled", False) else "disabled by DEV_BOOTSTRAP_ENABLED"
|
|
print(f"Dev bootstrap for missing SQLite DB: {bootstrap_state} ({state.bootstrap_db_path})")
|
|
print("Modules: " + ", ".join(manifest.id for manifest in state.registry.manifests()))
|
|
if no_reload:
|
|
print("Reload: disabled")
|
|
else:
|
|
if state.reload_module_ids is None:
|
|
print("Reload scope: core/config plus all enabled modules")
|
|
elif state.reload_module_ids:
|
|
print(
|
|
"Reload scope: core/config plus "
|
|
+ ", ".join(state.reload_module_ids)
|
|
)
|
|
else:
|
|
print("Reload scope: core/config only")
|
|
print("Reload dirs:")
|
|
for directory in state.reload_dirs:
|
|
print(f" - {directory}")
|
|
|
|
|
|
def run_smoke_check(state: DevserverState) -> None:
|
|
try:
|
|
from fastapi.testclient import TestClient
|
|
except ModuleNotFoundError as exc:
|
|
raise SystemExit("fastapi.testclient is not available. Install govoplan-core dev/server requirements before running --smoke.") from exc
|
|
|
|
from govoplan_core.server.app import create_app
|
|
|
|
print("Smoke check: creating ASGI app and running startup")
|
|
app = create_app(state.config)
|
|
with TestClient(app) as client:
|
|
response = client.get("/health")
|
|
if response.status_code != 200:
|
|
raise SystemExit(f"Smoke check failed: /health returned {response.status_code}: {response.text}")
|
|
|
|
db_path = _sqlite_database_path(state.database_url)
|
|
if state.bootstrap_db_path is not None:
|
|
if db_path is None or not db_path.exists() or db_path.stat().st_size <= 0:
|
|
raise SystemExit(f"Smoke check failed: expected initialized SQLite DB at {state.bootstrap_db_path}")
|
|
print("Smoke check: OK")
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
if args.config:
|
|
os.environ["GOVOPLAN_SERVER_CONFIG"] = args.config
|
|
|
|
config_path = args.config or os.getenv("GOVOPLAN_SERVER_CONFIG")
|
|
reload_module_ids: Sequence[str] | None = args.reload_module
|
|
if args.reload_core_only:
|
|
reload_module_ids = ()
|
|
state = prepare_devserver(
|
|
config_path,
|
|
extra_reload_dirs=args.reload_dir,
|
|
reload_module_ids=reload_module_ids,
|
|
)
|
|
print_devserver_summary(state, app=args.app, no_reload=args.no_reload)
|
|
|
|
if args.smoke:
|
|
run_smoke_check(state)
|
|
return 0
|
|
|
|
try:
|
|
import uvicorn
|
|
except ModuleNotFoundError as exc:
|
|
raise SystemExit("uvicorn is not installed. Install govoplan-core with the server extra or the meta requirements-dev.txt.") from exc
|
|
|
|
uvicorn.run(
|
|
args.app,
|
|
host=args.host,
|
|
port=args.port,
|
|
reload=not args.no_reload,
|
|
reload_dirs=state.reload_dirs if not args.no_reload else None,
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|