"""App-Loader: discovers apps//manifest.yaml, resolves deps, imports and registers.""" from __future__ import annotations import importlib from dataclasses import dataclass, field from pathlib import Path import yaml from fastapi import APIRouter, FastAPI APPS_DIR = Path(__file__).resolve().parent.parent / "apps" @dataclass class AppManifest: name: str version: str = "0.1.0" depends_on: list[str] = field(default_factory=list) conflicts_with: list[str] = field(default_factory=list) required: bool = False provides: list[str] = field(default_factory=list) def discover_manifests() -> list[AppManifest]: manifests: list[AppManifest] = [] if not APPS_DIR.exists(): return manifests for entry in sorted(APPS_DIR.iterdir()): mf_path = entry / "manifest.yaml" if not mf_path.exists(): continue data = yaml.safe_load(mf_path.read_text()) or {} manifests.append( AppManifest( name=data["name"], version=data.get("version", "0.1.0"), depends_on=list(data.get("depends_on", [])), conflicts_with=list(data.get("conflicts_with", [])), required=bool(data.get("required", False)), provides=list(data.get("provides", [])), ) ) return manifests def _topo_sort(manifests: list[AppManifest]) -> list[AppManifest]: by_name = {m.name: m for m in manifests} visited: set[str] = set() temp: set[str] = set() order: list[AppManifest] = [] def visit(name: str) -> None: if name in visited: return if name in temp: raise RuntimeError(f"Circular app dependency at {name}") m = by_name.get(name) if not m: return # dependency on core or non-app (e.g. 'core') temp.add(name) for d in m.depends_on: if d == "core": continue if d not in by_name: raise RuntimeError(f"App {name} depends on missing app: {d}") visit(d) temp.discard(name) visited.add(name) order.append(m) for m in manifests: visit(m.name) return order def _check_conflicts(manifests: list[AppManifest]) -> None: names = {m.name for m in manifests} for m in manifests: for c in m.conflicts_with: if c in names: raise RuntimeError(f"App conflict: {m.name} conflicts with {c}") def load_apps(app: FastAPI) -> list[AppManifest]: manifests = discover_manifests() _check_conflicts(manifests) ordered = _topo_sort(manifests) loaded: list[AppManifest] = [] for m in ordered: module_name = f"apps.{m.name}" mod = importlib.import_module(module_name) # Optional lifecycle hook if hasattr(mod, "on_load") and callable(mod.on_load): mod.on_load() if hasattr(mod, "router") and isinstance(mod.router, APIRouter): app.include_router(mod.router, prefix=f"/api/{m.name}", tags=[m.name]) loaded.append(m) print(f"[app-loader] loaded {m.name} v{m.version}") return loaded