File size: 3,503 Bytes
cf5e005 79c4bf9 cf5e005 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """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__)
# ── Canonical router list (single source of truth) ─────────────────
# Order matters for path resolution. v1 routers are mounted at /api/v1/*.
# Domain routers (news, reports, x402) self-prefix.
ROUTER_MODULES: Final[list[str]] = [
# ── Core system routes ─────────────────────────────────
"app.core.health_route", # /health, /live, /ready
"app.core.metrics", # /metrics
"app.homepage", # /, /version
# ── v1 API (thin HTTP layer) ───────────────────────────
"app.api.v1.auth.alerts", # /api/v1/alerts/*
"app.api.v1.public.wallet", # /api/v1/wallet/*
"app.api.v1.public.token", # /api/v1/token/*
"app.api.v1.public.scanner", # /api/v1/scanner/*
"app.api.v1.rag.search", # /api/v1/rag/v2/*
"app.api.v1.admin.alerts_webhook", # /api/v1/admin/alerts/webhook
"app.api.v1.admin.glitchtip_test", # /api/v1/_test/* (T07)
"app.api.v1.catalog", # /api/v1/catalog/*
"app.api.v1.mcp", # /mcp/* (JSON-RPC + plain JSON)
# ── Domain facades (per v4.0 §T28-T34) ─────────────────
"app.domain.news", # /api/v1/news/*
"app.domain.news.admin_router", # /api/v1/news/_admin/*
"app.domain.reports", # /api/v1/reports/*
"app.domain.x402", # /api/v1/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)
|