# Copyright (c) 2025 Patrick Motsch # All rights reserved. """ Feature Registry for Plug&Play Feature Container Loading. Dynamically discovers and loads feature containers from the features directory. Note: This module is in modules/system/ but manages modules/features/. """ import os import glob import importlib import logging from typing import List, Dict, Any from fastapi import FastAPI logger = logging.getLogger(__name__) # Path to the features directory (relative to this file's location) FEATURES_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "features") def discoverFeatureContainers() -> List[str]: """ Discover all feature container directories by filename pattern. A valid feature container has a routeFeature*.py file. """ containers = [] pattern = os.path.join(FEATURES_DIR, "*", "routeFeature*.py") for filepath in glob.glob(pattern): featureDir = os.path.basename(os.path.dirname(filepath)) if featureDir not in containers and not featureDir.startswith("_"): containers.append(featureDir) return sorted(containers) def loadFeatureRouters(app: FastAPI) -> Dict[str, Any]: """ Dynamically load and register routers from all discovered feature containers. Also registers feature template roles and AccessRules in the database. Searches for: - 'router' (main feature router) - 'templateRouter' (additional router for templates) - Any other *Router named exports """ results = {} pattern = os.path.join(FEATURES_DIR, "*", "routeFeature*.py") for filepath in glob.glob(pattern): featureDir = os.path.basename(os.path.dirname(filepath)) routerFile = os.path.basename(filepath)[:-3] # Remove .py if featureDir.startswith("_"): continue try: modulePath = f"modules.features.{featureDir}.{routerFile}" module = importlib.import_module(modulePath) loadedRouters = [] # Load main router if hasattr(module, "router"): app.include_router(module.router) loadedRouters.append("router") # Load additional named routers (e.g., templateRouter) for attrName in dir(module): if attrName.endswith("Router") and attrName != "APIRouter": routerObj = getattr(module, attrName) if hasattr(routerObj, "routes"): # Check if it's a FastAPI router app.include_router(routerObj) loadedRouters.append(attrName) if loadedRouters: logger.info(f"Loaded routers from {featureDir}: {loadedRouters}") results[featureDir] = {"status": "loaded", "module": modulePath, "routers": loadedRouters} else: logger.warning(f"No routers found in {modulePath}") results[featureDir] = {"status": "no_router_object"} except Exception as e: logger.error(f"Failed to load router from {featureDir}: {e}") results[featureDir] = {"status": "error", "error": str(e)} return results _cachedMainModules = None def loadFeatureMainModules() -> Dict[str, Any]: """ Dynamically load main modules from all discovered feature containers. Results are cached after the first call. """ global _cachedMainModules if _cachedMainModules is not None: return _cachedMainModules mainModules = {} pattern = os.path.join(FEATURES_DIR, "*", "main*.py") for filepath in glob.glob(pattern): filename = os.path.basename(filepath) if filename == "__init__.py": continue featureDir = os.path.basename(os.path.dirname(filepath)) if featureDir.startswith("_"): continue # Skip if this feature already has a main module loaded (avoid duplicates) if featureDir in mainModules: continue mainFile = filename[:-3] # Remove .py try: modulePath = f"modules.features.{featureDir}.{mainFile}" module = importlib.import_module(modulePath) mainModules[featureDir] = module logger.debug(f"Loaded main module: {featureDir}") except Exception as e: logger.error(f"Failed to load main module from {featureDir}: {e}") _cachedMainModules = mainModules return mainModules def registerAllFeaturesInCatalog(catalogService) -> Dict[str, bool]: """ Register all features' RBAC objects in the catalog. Also registers system-level RBAC objects and feature definitions. """ results = {} # Register system-level RBAC objects first try: from modules.system.mainSystem import registerFeature as registerSystemFeature success = registerSystemFeature(catalogService) results["system"] = success if success: logger.info("Registered RBAC objects: system") except ImportError as e: logger.warning(f"System module not found, skipping system RBAC registration: {e}") except Exception as e: logger.error(f"Error registering system RBAC objects: {e}") results["system"] = False # Register service center RBAC objects (service.web, service.extraction, etc.) try: from modules.serviceCenter import registerServiceObjects success = registerServiceObjects(catalogService) results["servicecenter"] = success except ImportError as e: logger.warning(f"Service center not found, skipping service RBAC registration: {e}") except Exception as e: logger.error(f"Error registering service RBAC objects: {e}") results["servicecenter"] = False # Register feature modules mainModules = loadFeatureMainModules() for featureName, module in mainModules.items(): # Register feature definition in catalog (for /api/features/ endpoint) if hasattr(module, "getFeatureDefinition"): try: featureDef = module.getFeatureDefinition() catalogService.registerFeatureDefinition( featureCode=featureDef.get("code", featureName), label=featureDef.get("label", {"en": featureName, "de": featureName}), icon=featureDef.get("icon", "mdi-puzzle"), instantiable=featureDef.get("instantiable", True), enabled=featureDef.get("enabled", True), ) logger.info(f"Registered feature definition: {featureDef.get('code', featureName)}") except Exception as e: logger.error(f"Error registering feature definition for {featureName}: {e}") # Register RBAC objects (UI, RESOURCE, DATA) if hasattr(module, "registerFeature"): try: success = module.registerFeature(catalogService) results[featureName] = success if success: logger.info(f"Registered RBAC objects: {featureName}") except Exception as e: logger.error(f"Error registering {featureName}: {e}") results[featureName] = False return results