File size: 5,616 Bytes
7fa9d90 4fc97b7 7fa9d90 4b9e6f7 7fa9d90 4b9e6f7 7fa9d90 4b9e6f7 7fa9d90 4b9e6f7 7fa9d90 4b9e6f7 7fa9d90 4b9e6f7 7fa9d90 4b9e6f7 7fa9d90 4fc97b7 7fa9d90 5134951 7fa9d90 5134951 7fa9d90 5134951 7fa9d90 5134951 7fa9d90 5134951 7fa9d90 5134951 7fa9d90 5134951 7fa9d90 5134951 7fa9d90 4fc97b7 7fa9d90 | 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 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | """
Module Registry for NCAkit
Handles automatic discovery and registration of feature modules.
"""
import importlib
import pkgutil
import logging
from pathlib import Path
from typing import List, Dict, Any, Callable, Optional
from fastapi import FastAPI
logger = logging.getLogger(__name__)
class ModuleInfo:
"""Information about a registered module"""
def __init__(
self,
name: str,
prefix: str,
description: str = "",
register_fn: Callable = None
):
self.name = name
self.prefix = prefix
self.description = description
self.register_fn = register_fn
class ModuleRegistry:
"""
Centralized registry for all NCAkit modules.
Each module must have an __init__.py with:
- MODULE_NAME: str
- MODULE_PREFIX: str
- MODULE_DESCRIPTION: str (optional)
- register(app, config): function
"""
def __init__(self):
self._modules: Dict[str, ModuleInfo] = {}
self._initialized: bool = False
def discover_modules(self, modules_package: str = "modules") -> List[str]:
"""
Discover all available modules in the modules package.
Returns list of module names.
"""
discovered = []
try:
logger.info(f"Discovering modules in package: {modules_package}")
package = importlib.import_module(modules_package)
package_path = Path(package.__file__).parent
logger.info(f"Module package path: {package_path}")
for finder, name, is_pkg in pkgutil.iter_modules([str(package_path)]):
logger.info(f"Found: {name} (is_package={is_pkg})")
# Skip private/template modules
if name.startswith('_'):
logger.info(f"Skipping private module: {name}")
continue
if is_pkg:
discovered.append(name)
logger.info(f"Discovered module: {name}")
except Exception as e:
logger.error(f"Error discovering modules: {e}")
import traceback
logger.error(traceback.format_exc())
logger.info(f"Total discovered: {discovered}")
return discovered
def load_module(self, module_name: str, modules_package: str = "modules") -> Optional[ModuleInfo]:
"""Load a single module and return its info"""
logger.info(f"[STEP 4] Loading module: {module_name}")
try:
full_module_name = f"{modules_package}.{module_name}"
logger.info(f"[STEP 4.1] Importing: {full_module_name}")
module = importlib.import_module(full_module_name)
logger.info(f"[STEP 4.2] Import successful: {full_module_name}")
# Check required attributes
if not hasattr(module, 'register'):
logger.warning(f"[STEP 4.3] Module {module_name} has no register function, skipping")
return None
logger.info(f"[STEP 4.3] register() function found")
# Get module metadata
name = getattr(module, 'MODULE_NAME', module_name)
prefix = getattr(module, 'MODULE_PREFIX', f"/api/{module_name}")
description = getattr(module, 'MODULE_DESCRIPTION', "")
logger.info(f"[STEP 4.4] Module metadata: name={name}, prefix={prefix}")
info = ModuleInfo(
name=name,
prefix=prefix,
description=description,
register_fn=module.register
)
self._modules[name] = info
logger.info(f"[STEP 4.5] Loaded module: {name} (prefix: {prefix})")
return info
except Exception as e:
logger.error(f"[STEP 4.ERROR] Failed to load module {module_name}: {e}")
import traceback
logger.error(f"[STEP 4.TRACEBACK]\n{traceback.format_exc()}")
return None
def register_all(self, app: FastAPI, config: Any) -> int:
"""
Register all discovered modules with the FastAPI app.
Returns number of successfully registered modules.
"""
if self._initialized:
logger.warning("Modules already initialized")
return len(self._modules)
# Discover modules
module_names = self.discover_modules()
registered = 0
for name in module_names:
info = self.load_module(name)
if info and info.register_fn:
try:
info.register_fn(app, config)
registered += 1
logger.info(f"Registered module: {info.name}")
except Exception as e:
logger.error(f"Failed to register module {name}: {e}")
self._initialized = True
logger.info(f"Registered {registered}/{len(module_names)} modules")
return registered
def get_module(self, name: str) -> Optional[ModuleInfo]:
"""Get info about a specific module"""
return self._modules.get(name)
def list_modules(self) -> List[Dict[str, str]]:
"""List all registered modules"""
return [
{
"name": info.name,
"prefix": info.prefix,
"description": info.description
}
for info in self._modules.values()
]
# Global registry instance
registry = ModuleRegistry()
|