You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from .base import ModuleMenuSeed, ModulePermission
|
|
|
|
|
|
ModulePhase = str
|
|
|
|
|
|
@dataclass
|
|
class ModuleRegistry:
|
|
"""Registry for in-process modules."""
|
|
|
|
modules: list[Any] = field(default_factory=list)
|
|
permissions: dict[str, ModulePermission] = field(default_factory=dict)
|
|
menu_seeds: dict[str, ModuleMenuSeed] = field(default_factory=dict)
|
|
|
|
def register(self, module: Any) -> None:
|
|
name = getattr(module, "name", None)
|
|
if not name:
|
|
raise ValueError("module must define a non-empty name")
|
|
if self.get(name) is not None:
|
|
raise ValueError(f"module already registered: {name}")
|
|
self.modules.append(module)
|
|
|
|
def extend(self, modules: Iterable[Any] | None) -> None:
|
|
for module in modules or []:
|
|
self.register(module)
|
|
|
|
def get(self, name: str) -> Any | None:
|
|
for module in self.modules:
|
|
if getattr(module, "name", None) == name:
|
|
return module
|
|
return None
|
|
|
|
def run_phase(self, phase: ModulePhase, app) -> None:
|
|
for module in self.modules:
|
|
callback = getattr(module, phase, None)
|
|
if callback is not None:
|
|
callback(app)
|
|
|
|
def register_permission(self, permission: ModulePermission) -> ModulePermission:
|
|
if not permission.code:
|
|
raise ValueError("permission code is required")
|
|
if permission.code in self.permissions:
|
|
raise ValueError(f"permission already registered: {permission.code}")
|
|
self.permissions[permission.code] = permission
|
|
return permission
|
|
|
|
def register_menu_seed(self, menu_seed: ModuleMenuSeed) -> ModuleMenuSeed:
|
|
if not menu_seed.id:
|
|
raise ValueError("menu seed id is required")
|
|
if not menu_seed.name:
|
|
raise ValueError("menu seed name is required")
|
|
if menu_seed.id in self.menu_seeds:
|
|
raise ValueError(f"menu seed already registered: {menu_seed.id}")
|
|
self.menu_seeds[menu_seed.id] = menu_seed
|
|
return menu_seed
|
|
|
|
def list_permissions(self) -> list[ModulePermission]:
|
|
return list(self.permissions.values())
|
|
|
|
def list_menu_seeds(self) -> list[ModuleMenuSeed]:
|
|
return sorted(
|
|
self.menu_seeds.values(),
|
|
key=lambda menu_seed: (menu_seed.sort, menu_seed.name),
|
|
)
|
|
|
|
|
|
def get_module_registry(app) -> ModuleRegistry:
|
|
registry = getattr(app.state, "iti_modules", None)
|
|
if registry is None:
|
|
registry = ModuleRegistry()
|
|
app.state.iti_modules = registry
|
|
return registry
|
|
|
|
|
|
def init_modules(app, modules: Iterable[Any] | None = None) -> ModuleRegistry:
|
|
registry = get_module_registry(app)
|
|
registry.extend(modules)
|
|
registry.run_phase("init_app", app)
|
|
registry.run_phase("register_tasks", app)
|
|
return registry
|