# app/modules/service.py """ Module service for platform module operations. Provides methods to check module enablement, get enabled modules, and filter menu items based on module configuration. Module configuration is stored in Platform.settings["enabled_modules"]. If not configured, all modules are enabled (backwards compatibility). """ import logging from functools import lru_cache from sqlalchemy.orm import Session from app.modules.base import ModuleDefinition from app.modules.registry import ( MODULES, get_core_module_codes, get_menu_item_module, get_module, ) from models.database.admin_menu_config import FrontendType from models.database.platform import Platform logger = logging.getLogger(__name__) class ModuleService: """ Service for platform module operations. Handles module enablement checking, module listing, and menu item filtering based on enabled modules. Module configuration is stored in Platform.settings["enabled_modules"]: - If key exists: Only listed modules (plus core) are enabled - If key missing: All modules are enabled (backwards compatibility) Example Platform.settings: { "enabled_modules": ["core", "billing", "inventory", "orders"], "module_config": { "billing": {"stripe_mode": "live"}, "inventory": {"low_stock_threshold": 10} } } """ # ========================================================================= # Module Enablement # ========================================================================= def get_platform_modules( self, db: Session, platform_id: int, ) -> list[ModuleDefinition]: """ Get all enabled modules for a platform. Args: db: Database session platform_id: Platform ID Returns: List of enabled ModuleDefinition objects (always includes core) """ enabled_codes = self._get_enabled_module_codes(db, platform_id) return [MODULES[code] for code in enabled_codes if code in MODULES] def get_enabled_module_codes( self, db: Session, platform_id: int, ) -> set[str]: """ Get set of enabled module codes for a platform. Args: db: Database session platform_id: Platform ID Returns: Set of enabled module codes (always includes core modules) """ return self._get_enabled_module_codes(db, platform_id) def is_module_enabled( self, db: Session, platform_id: int, module_code: str, ) -> bool: """ Check if a specific module is enabled for a platform. Core modules are always enabled. Args: db: Database session platform_id: Platform ID module_code: Module code to check Returns: True if module is enabled """ # Core modules are always enabled if module_code in get_core_module_codes(): return True enabled_codes = self._get_enabled_module_codes(db, platform_id) return module_code in enabled_codes def _get_enabled_module_codes( self, db: Session, platform_id: int, ) -> set[str]: """ Get enabled module codes from platform settings. Internal method that reads Platform.settings["enabled_modules"]. If not configured, returns all module codes (backwards compatibility). Always includes core modules. Args: db: Database session platform_id: Platform ID Returns: Set of enabled module codes """ platform = db.query(Platform).filter(Platform.id == platform_id).first() if not platform: logger.warning(f"Platform {platform_id} not found, returning all modules") return set(MODULES.keys()) settings = platform.settings or {} enabled_modules = settings.get("enabled_modules") # If not configured, enable all modules (backwards compatibility) if enabled_modules is None: return set(MODULES.keys()) # Always include core modules core_codes = get_core_module_codes() enabled_set = set(enabled_modules) | core_codes # Resolve dependencies - add required modules enabled_set = self._resolve_dependencies(enabled_set) return enabled_set def _resolve_dependencies(self, enabled_codes: set[str]) -> set[str]: """ Resolve module dependencies by adding required modules. If module A requires module B, and A is enabled, B must also be enabled. Args: enabled_codes: Set of explicitly enabled module codes Returns: Set of enabled module codes including dependencies """ resolved = set(enabled_codes) changed = True while changed: changed = False for code in list(resolved): module = get_module(code) if module: for required in module.requires: if required not in resolved: resolved.add(required) changed = True logger.debug( f"Module '{code}' requires '{required}', auto-enabling" ) return resolved # ========================================================================= # Menu Item Filtering # ========================================================================= def get_module_menu_items( self, db: Session, platform_id: int, frontend_type: FrontendType, ) -> set[str]: """ Get all menu item IDs available for enabled modules. Args: db: Database session platform_id: Platform ID frontend_type: Which frontend (admin or vendor) Returns: Set of menu item IDs from enabled modules """ enabled_modules = self.get_platform_modules(db, platform_id) menu_items = set() for module in enabled_modules: menu_items.update(module.get_menu_items(frontend_type)) return menu_items def is_menu_item_module_enabled( self, db: Session, platform_id: int, menu_item_id: str, frontend_type: FrontendType, ) -> bool: """ Check if the module providing a menu item is enabled. Args: db: Database session platform_id: Platform ID menu_item_id: Menu item ID frontend_type: Which frontend (admin or vendor) Returns: True if the module providing this menu item is enabled, or if menu item is not associated with any module. """ module_code = get_menu_item_module(menu_item_id, frontend_type) # If menu item isn't associated with any module, allow it if module_code is None: return True return self.is_module_enabled(db, platform_id, module_code) def filter_menu_items_by_modules( self, db: Session, platform_id: int, menu_item_ids: set[str], frontend_type: FrontendType, ) -> set[str]: """ Filter menu items to only those from enabled modules. Args: db: Database session platform_id: Platform ID menu_item_ids: Set of menu item IDs to filter frontend_type: Which frontend (admin or vendor) Returns: Filtered set of menu item IDs """ available_items = self.get_module_menu_items(db, platform_id, frontend_type) # Items that are in available_items, OR items not associated with any module filtered = set() for item_id in menu_item_ids: module_code = get_menu_item_module(item_id, frontend_type) if module_code is None or item_id in available_items: filtered.add(item_id) return filtered # ========================================================================= # Module Configuration # ========================================================================= def get_module_config( self, db: Session, platform_id: int, module_code: str, ) -> dict: """ Get module-specific configuration for a platform. Args: db: Database session platform_id: Platform ID module_code: Module code Returns: Module configuration dict (empty if not configured) """ platform = db.query(Platform).filter(Platform.id == platform_id).first() if not platform: return {} settings = platform.settings or {} module_configs = settings.get("module_config", {}) return module_configs.get(module_code, {}) def set_enabled_modules( self, db: Session, platform_id: int, module_codes: list[str], ) -> bool: """ Set the enabled modules for a platform. Core modules are automatically included. Dependencies are automatically resolved. Args: db: Database session platform_id: Platform ID module_codes: List of module codes to enable Returns: True if successful, False if platform not found """ platform = db.query(Platform).filter(Platform.id == platform_id).first() if not platform: logger.error(f"Platform {platform_id} not found") return False # Validate module codes valid_codes = set(MODULES.keys()) invalid = [code for code in module_codes if code not in valid_codes] if invalid: logger.warning(f"Invalid module codes ignored: {invalid}") module_codes = [code for code in module_codes if code in valid_codes] # Always include core modules core_codes = get_core_module_codes() enabled_set = set(module_codes) | core_codes # Resolve dependencies enabled_set = self._resolve_dependencies(enabled_set) # Update platform settings settings = platform.settings or {} settings["enabled_modules"] = list(enabled_set) platform.settings = settings logger.info( f"Updated enabled modules for platform {platform_id}: {sorted(enabled_set)}" ) return True def enable_module( self, db: Session, platform_id: int, module_code: str, ) -> bool: """ Enable a single module for a platform. Also enables required dependencies. Args: db: Database session platform_id: Platform ID module_code: Module code to enable Returns: True if successful """ if module_code not in MODULES: logger.error(f"Unknown module: {module_code}") return False platform = db.query(Platform).filter(Platform.id == platform_id).first() if not platform: logger.error(f"Platform {platform_id} not found") return False settings = platform.settings or {} enabled = set(settings.get("enabled_modules", list(MODULES.keys()))) enabled.add(module_code) # Resolve dependencies enabled = self._resolve_dependencies(enabled) settings["enabled_modules"] = list(enabled) platform.settings = settings logger.info(f"Enabled module '{module_code}' for platform {platform_id}") return True def disable_module( self, db: Session, platform_id: int, module_code: str, ) -> bool: """ Disable a single module for a platform. Core modules cannot be disabled. Also disables modules that depend on this one. Args: db: Database session platform_id: Platform ID module_code: Module code to disable Returns: True if successful, False if core or not found """ if module_code not in MODULES: logger.error(f"Unknown module: {module_code}") return False module = MODULES[module_code] if module.is_core: logger.warning(f"Cannot disable core module: {module_code}") return False platform = db.query(Platform).filter(Platform.id == platform_id).first() if not platform: logger.error(f"Platform {platform_id} not found") return False settings = platform.settings or {} enabled = set(settings.get("enabled_modules", list(MODULES.keys()))) # Remove this module enabled.discard(module_code) # Remove modules that depend on this one dependents = self._get_dependent_modules(module_code) for dependent in dependents: if dependent in enabled: enabled.discard(dependent) logger.info( f"Also disabled '{dependent}' (depends on '{module_code}')" ) settings["enabled_modules"] = list(enabled) platform.settings = settings logger.info(f"Disabled module '{module_code}' for platform {platform_id}") return True def _get_dependent_modules(self, module_code: str) -> set[str]: """ Get modules that depend on a given module. Args: module_code: Module code to find dependents for Returns: Set of module codes that require the given module """ dependents = set() for code, module in MODULES.items(): if module_code in module.requires: dependents.add(code) # Recursively find dependents of dependents dependents.update(self._get_dependent_modules(code)) return dependents # ========================================================================= # Platform Code Helpers # ========================================================================= def get_platform_modules_by_code( self, db: Session, platform_code: str, ) -> list[ModuleDefinition]: """ Get enabled modules for a platform by code. Args: db: Database session platform_code: Platform code (e.g., "oms", "loyalty") Returns: List of enabled ModuleDefinition objects """ platform = db.query(Platform).filter(Platform.code == platform_code).first() if not platform: logger.warning(f"Platform '{platform_code}' not found, returning all modules") return list(MODULES.values()) return self.get_platform_modules(db, platform.id) def is_module_enabled_by_code( self, db: Session, platform_code: str, module_code: str, ) -> bool: """ Check if a module is enabled for a platform by code. Args: db: Database session platform_code: Platform code (e.g., "oms", "loyalty") module_code: Module code to check Returns: True if module is enabled """ platform = db.query(Platform).filter(Platform.code == platform_code).first() if not platform: logger.warning(f"Platform '{platform_code}' not found, assuming enabled") return True return self.is_module_enabled(db, platform.id, module_code) # Singleton instance module_service = ModuleService() __all__ = [ "ModuleService", "module_service", ]