145 lines
5.1 KiB
Python
145 lines
5.1 KiB
Python
# 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.
|
|
"""
|
|
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)
|
|
|
|
if hasattr(module, "router"):
|
|
app.include_router(module.router)
|
|
logger.info(f"Loaded router: {featureDir}")
|
|
results[featureDir] = {"status": "loaded", "module": modulePath}
|
|
else:
|
|
logger.warning(f"No 'router' 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)}
|
|
|
|
# Register features in RBAC catalog and sync template roles to database
|
|
from modules.security.rbacCatalog import getCatalogService
|
|
catalogService = getCatalogService()
|
|
registrationResults = registerAllFeaturesInCatalog(catalogService)
|
|
|
|
for featureName, success in registrationResults.items():
|
|
if featureName in results:
|
|
results[featureName]["rbac_registered"] = success
|
|
|
|
return results
|
|
|
|
|
|
def loadFeatureMainModules() -> Dict[str, Any]:
|
|
"""
|
|
Dynamically load main modules from all discovered feature containers.
|
|
"""
|
|
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
|
|
|
|
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}")
|
|
|
|
return mainModules
|
|
|
|
|
|
def registerAllFeaturesInCatalog(catalogService) -> Dict[str, bool]:
|
|
"""
|
|
Register all features' RBAC objects in the catalog.
|
|
Also registers system-level RBAC objects.
|
|
"""
|
|
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 feature modules
|
|
mainModules = loadFeatureMainModules()
|
|
|
|
for featureName, module in mainModules.items():
|
|
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
|