"""Реестр модулей хаба. Список собирается лениво: ядро не импортирует модули при импорте.""" from .modules import Module _cache: list[Module] | None = None def modules() -> list[Module]: global _cache if _cache is None: from ..modules import jira # внутри функции: ядро не зависит от модулей при импорте from . import settings mods = [jira.MODULE] # маршруты ядра проверяет core/api.py при сборке ROUTES: здесь их ещё нет check(mods, [], settings.CORE_DEFAULTS) _cache = mods return _cache DEFAULT_AGENT = "jira" # агент по умолчанию: GET /agent без module, agent publish без --module def agent_modules() -> list[Module]: """Модули с командами: у каждого свой агент (имя агента = имя модуля), в порядке реестра.""" return [m for m in modules() if m.commands] def module(name: str) -> Module | None: return next((m for m in modules() if m.name == name), None) def module_of(ctype: str) -> Module | None: """Модуль, владеющий типом команды.""" return next((m for m in modules() if m.commands and ctype in m.commands.types), None) def check(mods: list[Module], core_routes, core_settings: dict) -> None: """Проверить, что модули не делят тип команды, ключ настроек и маршрут между собой и с ядром.""" types: dict[str, str] = {} keys: dict[str, str] = {key: "ядро" for key in core_settings} routes: dict[tuple[str, str], str] = {(r[0], r[1]): "ядро" for r in core_routes} for m in mods: for ctype in sorted(m.commands.types) if m.commands else (): if ctype in types: raise RuntimeError(f"тип {ctype}: модули {types[ctype]} и {m.name}") types[ctype] = m.name for key in m.settings_defaults: if key in keys: raise RuntimeError(f"ключ настроек {key}: {keys[key]} и модуль {m.name}") keys[key] = m.name for r in m.routes: if (r[0], r[1]) in routes: raise RuntimeError(f"маршрут {r[0]} {r[1]}: {routes[(r[0], r[1])]} и модуль {m.name}") routes[(r[0], r[1])] = m.name