feat: 实现功能模块系统,支持模块的动态注册与启用控制

- 新增 ModuleRegistry 核心,支持模块注册、可用性检查和启用状态管理
- 新增模块管理 API(管理端和公共端),提供模块状态查询和启用控制
- 将 LDAP 功能迁移到模块系统,支持通过环境变量和数据库配置控制
- 前端新增模块管理页面和 store,支持模块启用/禁用操作
- 路由守卫集成模块状态检查,未激活模块的页面自动重定向
- 面包屑导航支持模块配置页面的层级显示
- Switch 组件新增 disabled 属性支持
This commit is contained in:
fawney19
2026-01-16 16:14:12 +08:00
parent 50343d5459
commit 6c98816f9f
17 changed files with 1332 additions and 39 deletions

View File

@@ -0,0 +1,26 @@
"""
模块化系统核心
提供可扩展的功能模块管理,支持:
- 声明式模块注册
- available/enabled 双层状态控制
- 延迟导入避免重依赖加载
- 前后端状态同步
"""
from src.core.modules.base import (
ModuleCategory,
ModuleDefinition,
ModuleMetadata,
ModuleStatus,
)
from src.core.modules.registry import ModuleRegistry, get_module_registry
__all__ = [
"ModuleCategory",
"ModuleMetadata",
"ModuleDefinition",
"ModuleStatus",
"ModuleRegistry",
"get_module_registry",
]

113
src/core/modules/base.py Normal file
View File

@@ -0,0 +1,113 @@
"""
模块基础定义
包含模块元数据、定义和状态的数据结构
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import TYPE_CHECKING, Any, Awaitable, Callable, List, Optional
if TYPE_CHECKING:
from fastapi import APIRouter
class ModuleCategory(str, Enum):
"""模块分类"""
AUTH = "auth" # 认证相关
MONITORING = "monitoring" # 监控相关
SECURITY = "security" # 安全相关
INTEGRATION = "integration" # 第三方集成
class ModuleHealth(str, Enum):
"""模块健康状态"""
HEALTHY = "healthy"
DEGRADED = "degraded"
UNHEALTHY = "unhealthy"
UNKNOWN = "unknown"
@dataclass
class ModuleMetadata:
"""
模块元数据 - 纯数据描述,无重依赖
用于声明式定义模块的基本信息和配置
"""
# 基本信息
name: str # 唯一标识: ldap, audit_log
display_name: str # 显示名称: "LDAP 认证"
description: str # 模块描述
# 分类
category: ModuleCategory
# 可用性控制(部署级)
env_key: str # 环境变量名: LDAP_AVAILABLE
default_available: bool = False # 默认是否可用
required_packages: List[str] = field(default_factory=list) # 依赖的 Python 包
dependencies: List[str] = field(default_factory=list) # 依赖的其他模块
# 路由配置 - 模块自定义前缀
api_prefix: Optional[str] = None # 如 "/api/admin/ldap"
# 前端配置
admin_route: Optional[str] = None # 管理页面路由: "/admin/ldap"
admin_menu_icon: Optional[str] = None # 菜单图标
admin_menu_group: Optional[str] = None # 菜单分组: "system", "security"
admin_menu_order: int = 100 # 菜单排序(越小越靠前)
@dataclass
class ModuleDefinition:
"""
完整模块定义
包含元数据和生命周期钩子,钩子函数内部延迟导入重依赖
"""
metadata: ModuleMetadata
# 工厂函数 - 内部再 import 重依赖
router_factory: Optional[Callable[[], "APIRouter"]] = None
service_factory: Optional[Callable[[], Any]] = None
# 生命周期钩子
on_startup: Optional[Callable[[], Awaitable[None]]] = None
on_shutdown: Optional[Callable[[], Awaitable[None]]] = None
health_check: Optional[Callable[[], Awaitable[ModuleHealth]]] = None
# 自定义依赖检测(可选,用于检测 ldap3 等库是否安装)
check_dependencies: Optional[Callable[[], bool]] = None
@dataclass
class ModuleStatus:
"""
模块运行状态
用于 API 返回,供前端使用
"""
name: str
available: bool # 部署级可用(环境变量 + 依赖库)
enabled: bool # 运行级启用(数据库配置)
active: bool # 最终激活状态 (available && enabled && dependencies_ok)
# 显示信息
display_name: str
description: str
category: ModuleCategory
# 前端配置
admin_route: Optional[str]
admin_menu_icon: Optional[str]
admin_menu_group: Optional[str]
admin_menu_order: int
# 健康状态
health: ModuleHealth = ModuleHealth.UNKNOWN

View File

@@ -0,0 +1,288 @@
"""
模块注册中心
负责模块的注册、状态管理和生命周期控制
"""
import importlib.util
import os
from typing import TYPE_CHECKING, Dict, List, Optional, Set
from src.core.logger import logger
from src.core.modules.base import (
ModuleCategory,
ModuleDefinition,
ModuleHealth,
ModuleStatus,
)
if TYPE_CHECKING:
from sqlalchemy.orm import Session
class ModuleRegistry:
"""
模块注册中心 - 单例模式
职责:
- 注册模块定义(仅元数据,不加载重依赖)
- 检查模块可用性(环境变量 + 依赖库)
- 管理模块启用状态(数据库配置)
- 提供模块状态查询
"""
_instance: Optional["ModuleRegistry"] = None
def __init__(self):
self._modules: Dict[str, ModuleDefinition] = {}
self._initialized: Set[str] = set()
@classmethod
def get_instance(cls) -> "ModuleRegistry":
"""获取单例实例"""
if cls._instance is None:
cls._instance = cls()
return cls._instance
@classmethod
def reset_instance(cls) -> None:
"""重置单例(仅用于测试)"""
cls._instance = None
def register(self, module: ModuleDefinition) -> None:
"""
注册模块
仅注册元数据,不加载重依赖
"""
name = module.metadata.name
if name in self._modules:
logger.warning(f"Module [{name}] already registered, skipping")
return
self._modules[name] = module
logger.debug(f"Module [{name}] registered")
def get_module(self, name: str) -> Optional[ModuleDefinition]:
"""获取模块定义"""
return self._modules.get(name)
def get_all_modules(self) -> List[ModuleDefinition]:
"""获取所有已注册模块"""
return list(self._modules.values())
# ========== 可用性检查(部署级)==========
def is_available(self, name: str) -> bool:
"""
检查模块是否部署可用
检查顺序:
1. 模块是否已注册
2. 环境变量是否启用
3. 依赖的 Python 包是否安装
4. 自定义依赖检测(如果有)
"""
if name not in self._modules:
return False
module = self._modules[name]
meta = module.metadata
# 1. 检查环境变量
env_value = os.getenv(meta.env_key)
if env_value is not None:
if env_value.lower() not in ("true", "1", "yes"):
return False
elif not meta.default_available:
return False
# 2. 检查依赖的 Python 包
for pkg in meta.required_packages:
if importlib.util.find_spec(pkg) is None:
logger.debug(f"Module [{name}] unavailable: package '{pkg}' not installed")
return False
# 3. 自定义依赖检测
if module.check_dependencies:
try:
if not module.check_dependencies():
logger.debug(f"Module [{name}] unavailable: custom dependency check failed")
return False
except Exception as e:
logger.warning(f"Module [{name}] dependency check error: {e}")
return False
return True
def get_available_modules(self) -> List[ModuleDefinition]:
"""获取所有部署可用的模块"""
return [m for m in self._modules.values() if self.is_available(m.metadata.name)]
# ========== 启用状态检查(运行级)==========
def is_enabled(self, name: str, db: "Session") -> bool:
"""
检查模块是否运行启用(数据库配置)
Args:
name: 模块名称
db: 数据库会话
"""
from src.services.system.config import SystemConfigService
config_key = f"module.{name}.enabled"
value = SystemConfigService.get_config(db, config_key, default=False)
return bool(value)
def set_enabled(self, name: str, enabled: bool, db: "Session") -> None:
"""
设置模块启用状态
Args:
name: 模块名称
enabled: 是否启用
db: 数据库会话
"""
from src.services.system.config import SystemConfigService
if name not in self._modules:
raise ValueError(f"Module [{name}] not registered")
config_key = f"module.{name}.enabled"
module = self._modules[name]
description = f"模块 [{module.metadata.display_name}] 启用状态"
SystemConfigService.set_config(db, config_key, enabled, description)
# ========== 激活状态检查 ==========
def is_active(self, name: str, db: "Session") -> bool:
"""
检查模块是否最终激活
激活条件available && enabled && 依赖模块都激活
"""
if not self.is_available(name):
return False
if not self.is_enabled(name, db):
return False
# 检查依赖模块
module = self._modules[name]
for dep in module.metadata.dependencies:
if not self.is_active(dep, db):
return False
return True
# ========== 状态查询 ==========
def get_module_status(
self, name: str, db: "Session", health: Optional[ModuleHealth] = None
) -> Optional[ModuleStatus]:
"""
获取单个模块状态
Args:
name: 模块名称
db: 数据库会话
health: 预先获取的健康状态(可选,用于异步场景)
"""
if name not in self._modules:
return None
module = self._modules[name]
meta = module.metadata
available = self.is_available(name)
return ModuleStatus(
name=name,
available=available,
enabled=self.is_enabled(name, db) if available else False,
active=self.is_active(name, db) if available else False,
display_name=meta.display_name,
description=meta.description,
category=meta.category,
admin_route=meta.admin_route if available else None,
admin_menu_icon=meta.admin_menu_icon,
admin_menu_group=meta.admin_menu_group,
admin_menu_order=meta.admin_menu_order,
health=health if health else ModuleHealth.UNKNOWN,
)
async def check_health(self, name: str) -> ModuleHealth:
"""
执行模块健康检查
Args:
name: 模块名称
Returns:
健康状态
"""
if name not in self._modules:
return ModuleHealth.UNKNOWN
module = self._modules[name]
if not module.health_check:
return ModuleHealth.UNKNOWN
try:
return await module.health_check()
except Exception as e:
logger.warning(f"Module [{name}] health check failed: {e}")
return ModuleHealth.UNHEALTHY
async def get_module_status_async(
self, name: str, db: "Session"
) -> Optional[ModuleStatus]:
"""异步获取模块状态(包含健康检查)"""
if name not in self._modules:
return None
health = await self.check_health(name) if self.is_available(name) else ModuleHealth.UNKNOWN
return self.get_module_status(name, db, health=health)
async def get_all_status_async(self, db: "Session") -> Dict[str, ModuleStatus]:
"""异步获取所有模块状态(包含健康检查)"""
result = {}
for name in self._modules:
status = await self.get_module_status_async(name, db)
if status:
result[name] = status
return result
def get_all_status(self, db: "Session") -> Dict[str, ModuleStatus]:
"""获取所有模块状态(同步版本,不含健康检查)"""
result = {}
for name in self._modules:
status = self.get_module_status(name, db)
if status:
result[name] = status
return result
def get_available_status(self, db: "Session") -> Dict[str, ModuleStatus]:
"""获取所有可用模块的状态"""
result = {}
for name, module in self._modules.items():
if self.is_available(name):
status = self.get_module_status(name, db)
if status:
result[name] = status
return result
def get_auth_modules_status(self, db: "Session") -> List[ModuleStatus]:
"""获取认证模块状态(供登录页使用)"""
result = []
for name, module in self._modules.items():
if module.metadata.category == ModuleCategory.AUTH:
if self.is_available(name):
status = self.get_module_status(name, db)
if status:
result.append(status)
return result
def get_module_registry() -> ModuleRegistry:
"""获取模块注册中心实例"""
return ModuleRegistry.get_instance()