| """RMI Backend β router mounting. |
| |
| Single source of truth for every router. Adding a new domain: |
| 1. Create the module with a `router = APIRouter(...)` attribute |
| 2. Add the import path to ROUTER_MODULES below |
| 3. Done β factory.create_app() picks it up automatically |
| |
| Per v4.0 Β§T01 + ADR-0001, this replaces the hardcoded lists that |
| existed in the old main.py. Each mount is isolated β one failure |
| doesn't break the others. |
| """ |
| from __future__ import annotations |
|
|
| import importlib |
| import logging |
| from typing import Final |
|
|
| log = logging.getLogger(__name__) |
|
|
|
|
| |
| |
| |
| ROUTER_MODULES: Final[list[str]] = [ |
| |
| "app.core.health_route", |
| "app.core.metrics", |
| "app.homepage", |
|
|
| |
| "app.api.v1.auth.alerts", |
| "app.api.v1.public.wallet", |
| "app.api.v1.public.token", |
| "app.api.v1.public.scanner", |
| "app.api.v1.rag.search", |
| "app.api.v1.admin.alerts_webhook", |
| "app.api.v1.admin.glitchtip_test", |
| "app.api.v1.catalog", |
| "app.api.v1.mcp", |
|
|
| |
| "app.domain.news", |
| "app.domain.news.admin_router", |
| "app.domain.reports", |
| "app.domain.x402", |
| ] |
|
|
|
|
| def mount_all(app) -> int: |
| """Mount every router. Returns count of successful mounts. |
| |
| Each mount is isolated β one failure does not break the rest. |
| """ |
| mounted = 0 |
| for module_path in ROUTER_MODULES: |
| if _try_mount(app, module_path): |
| mounted += 1 |
| return mounted |
|
|
|
|
| def _try_mount(app, module_path: str) -> bool: |
| """Try to import + mount a single router. Returns True on success.""" |
| try: |
| module = importlib.import_module(module_path) |
| router = getattr(module, "router", None) |
| if router is None: |
| log.warning("router_missing path=%s (no `router` attribute)", module_path) |
| return False |
| app.include_router(router) |
| log.info("router_mounted path=%s", module_path) |
| return True |
| except Exception as exc: |
| log.warning( |
| "router_mount_failed path=%s err=%s: %s", |
| module_path, type(exc).__name__, exc, |
| ) |
| return False |
|
|
|
|
| def register_external_router(module_path: str) -> None: |
| """Allow third-party code to register additional routers at runtime. |
| |
| Plugins / MCP servers / worker modules can call this from their |
| own __init__.py to add routes to the main app. |
| """ |
| if module_path not in ROUTER_MODULES: |
| ROUTER_MODULES.append(module_path) |
| log.info("router_registered_external path=%s", module_path) |
|
|