# app/modules/base.py """ Base module definition class. A Module is a self-contained unit of functionality that can be enabled/disabled per platform. Each module contains: - Features: Granular capabilities for tier-based access control - Menu items: Sidebar entries per frontend type - Routes: API and page routes (future: dynamically registered) - Services: Business logic (optional, for self-contained modules) - Models: Database models (optional, for self-contained modules) - Schemas: Pydantic schemas (optional, for self-contained modules) - Templates: Jinja2 templates (optional, for self-contained modules) - Migrations: Database migrations (optional, for self-contained modules) Module Classification: - Core modules: Always enabled, cannot be disabled (core, tenancy, cms, customers) - Optional modules: Can be enabled/disabled per platform - Internal modules: Admin-only tools, not customer-facing (dev-tools, monitoring) Self-Contained Module Structure: app/modules// ├── __init__.py ├── definition.py # ModuleDefinition instance ├── config.py # Module configuration schema (optional) ├── exceptions.py # Module-specific exceptions (optional) ├── routes/ # FastAPI routers ├── services/ # Business logic ├── tasks/ # Celery background tasks (optional) │ └── __init__.py # Task module discovery marker ├── models/ # SQLAlchemy models ├── schemas/ # Pydantic schemas ├── migrations/ # Alembic migrations for this module │ └── versions/ # Migration scripts ├── templates/ # Jinja2 templates (namespaced) └── locales/ # Translation files """ from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Any, Callable if TYPE_CHECKING: from fastapi import APIRouter from pydantic import BaseModel from app.modules.enums import FrontendType # ============================================================================= # Menu Item Definitions # ============================================================================= @dataclass class MenuItemDefinition: """ Definition of a single menu item within a section. Attributes: id: Unique identifier (e.g., "catalog.products", "orders.list") label_key: i18n key for the menu item label icon: Lucide icon name (e.g., "box", "shopping-cart") route: URL path (can include placeholders like {vendor_code}) order: Sort order within section (lower = higher priority) is_mandatory: If True, cannot be hidden by user preferences requires_permission: Permission code required to see this item badge_source: Key for dynamic badge count (e.g., "pending_orders_count") is_super_admin_only: Only visible to super admins (admin frontend only) Example: MenuItemDefinition( id="catalog.products", label_key="catalog.menu.products", icon="box", route="/admin/catalog/products", order=10, is_mandatory=True ) """ id: str label_key: str icon: str route: str order: int = 100 is_mandatory: bool = False requires_permission: str | None = None badge_source: str | None = None is_super_admin_only: bool = False @dataclass class MenuSectionDefinition: """ Definition of a menu section containing related menu items. Sections group related menu items together in the sidebar. A section can be collapsed/expanded by the user. Attributes: id: Unique section identifier (e.g., "catalog", "orders") label_key: i18n key for section header (None for headerless sections) icon: Lucide icon name for section (optional) order: Sort order among sections (lower = higher priority) items: List of menu items in this section is_super_admin_only: Only visible to super admins is_collapsible: Whether section can be collapsed Example: MenuSectionDefinition( id="catalog", label_key="catalog.menu.section", icon="package", order=20, items=[ MenuItemDefinition( id="catalog.products", label_key="catalog.menu.products", icon="box", route="/admin/catalog/products", order=10 ), ] ) """ id: str label_key: str | None icon: str | None = None order: int = 100 items: list[MenuItemDefinition] = field(default_factory=list) is_super_admin_only: bool = False is_collapsible: bool = True # ============================================================================= # Permission Definitions # ============================================================================= @dataclass class PermissionDefinition: """ Definition of a permission that a module exposes. Permissions are granular capabilities that can be assigned to roles/users. Each module defines its own permissions, which are then discovered and aggregated by the tenancy module for role assignment. Attributes: id: Unique identifier in format "resource.action" (e.g., "products.view") label_key: i18n key for the permission label description_key: i18n key for permission description category: Grouping category for UI organization (e.g., "products", "orders") is_owner_only: If True, only vendor owners can have this permission Example: PermissionDefinition( id="products.view", label_key="catalog.permissions.products_view", description_key="catalog.permissions.products_view_desc", category="products", ) """ id: str label_key: str description_key: str = "" category: str = "general" is_owner_only: bool = False # ============================================================================= # Scheduled Task Definitions # ============================================================================= @dataclass class ScheduledTask: """ Definition of a Celery Beat scheduled task. Used in ModuleDefinition to declare scheduled tasks that should be registered with Celery Beat when the module is loaded. Attributes: name: Unique name for the schedule entry (e.g., "billing.reset_counters") task: Full Python path to the task (e.g., "app.modules.billing.tasks.subscription.reset_period_counters") schedule: Cron expression string or crontab dict - String format: "minute hour day_of_month month day_of_week" (e.g., "5 0 * * *") - Dict format: {"minute": 5, "hour": 0} for crontab kwargs args: Positional arguments to pass to the task kwargs: Keyword arguments to pass to the task options: Celery task options (e.g., {"queue": "scheduled"}) Example: ScheduledTask( name="billing.reset_period_counters", task="app.modules.billing.tasks.subscription.reset_period_counters", schedule="5 0 * * *", # Daily at 00:05 options={"queue": "scheduled"}, ) """ name: str task: str schedule: str | dict[str, Any] args: tuple = () kwargs: dict[str, Any] = field(default_factory=dict) options: dict[str, Any] = field(default_factory=dict) @dataclass class ModuleDefinition: """ Definition of a platform module. A module groups related functionality that can be enabled/disabled per platform. Core modules cannot be disabled and are always available. Self-contained modules include their own services, models, schemas, templates, and migrations. The path attributes describe where these components are located. Attributes: # Identity code: Unique identifier (e.g., "billing", "marketplace") name: Display name (e.g., "Billing & Subscriptions") description: Description of what this module provides version: Semantic version of the module (e.g., "1.0.0") # Dependencies requires: List of module codes this module depends on # Components features: List of feature codes this module provides menu_items: Dict mapping FrontendType to list of menu item IDs permissions: List of permission codes this module defines # Classification is_core: Core modules cannot be disabled is_internal: Internal modules are admin-only (not customer-facing) # Configuration config_schema: Pydantic model class for module configuration default_config: Default configuration values # Routes admin_router: FastAPI router for admin routes vendor_router: FastAPI router for vendor routes # Lifecycle hooks on_enable: Called when module is enabled for a platform on_disable: Called when module is disabled for a platform on_startup: Called when application starts (for enabled modules) health_check: Called to check module health status # Self-contained module paths (optional) is_self_contained: Whether module uses self-contained structure services_path: Path to services subpackage models_path: Path to models subpackage schemas_path: Path to schemas subpackage templates_path: Path to templates directory (relative to module) exceptions_path: Path to exceptions module locales_path: Path to locales directory (relative to module) migrations_path: Path to migrations directory (relative to module) Example (traditional thin wrapper): billing_module = ModuleDefinition( code="billing", name="Billing & Subscriptions", description="Subscription tiers, billing history, and payment processing", features=["subscription_management", "billing_history", "stripe_integration"], menu_items={ FrontendType.ADMIN: ["subscription-tiers", "subscriptions", "billing-history"], FrontendType.VENDOR: ["billing"], }, ) Example (self-contained module with configuration): from pydantic import BaseModel, Field class CMSConfig(BaseModel): max_pages: int = Field(default=100, ge=1) enable_seo: bool = True cms_module = ModuleDefinition( code="cms", name="Content Management", version="1.0.0", features=["cms_basic", "cms_custom_pages"], config_schema=CMSConfig, default_config={"max_pages": 100, "enable_seo": True}, is_self_contained=True, services_path="app.modules.cms.services", models_path="app.modules.cms.models", migrations_path="migrations", health_check=lambda: {"status": "healthy"}, ) """ # ========================================================================= # Identity # ========================================================================= code: str name: str description: str = "" version: str = "1.0.0" # ========================================================================= # Dependencies # ========================================================================= requires: list[str] = field(default_factory=list) # ========================================================================= # Components # ========================================================================= features: list[str] = field(default_factory=list) menu_items: dict[FrontendType, list[str]] = field(default_factory=dict) permissions: list[PermissionDefinition] = field(default_factory=list) # ========================================================================= # Menu Definitions (Module-Driven Menus) # ========================================================================= # NEW: Full menu definitions per frontend type. When set, these take # precedence over menu_items for menu rendering. This enables modules # to fully define their own menu structure with icons, routes, and labels. menus: dict[FrontendType, list[MenuSectionDefinition]] = field(default_factory=dict) # ========================================================================= # Classification # ========================================================================= is_core: bool = False is_internal: bool = False # ========================================================================= # Configuration # ========================================================================= config_schema: "type[BaseModel] | None" = None default_config: dict[str, Any] = field(default_factory=dict) # ========================================================================= # Routes (registered dynamically) # ========================================================================= admin_router: "APIRouter | None" = None vendor_router: "APIRouter | None" = None # ========================================================================= # Lifecycle Hooks # ========================================================================= on_enable: Callable[[int], None] | None = None # Called with platform_id on_disable: Callable[[int], None] | None = None # Called with platform_id on_startup: Callable[[], None] | None = None # Called on app startup health_check: Callable[[], dict[str, Any]] | None = None # Returns health status # ========================================================================= # Self-Contained Module Paths (optional) # ========================================================================= is_self_contained: bool = False services_path: str | None = None models_path: str | None = None schemas_path: str | None = None templates_path: str | None = None # Relative to module directory exceptions_path: str | None = None locales_path: str | None = None # Relative to module directory migrations_path: str | None = None # Relative to module directory, e.g., "migrations" # ========================================================================= # Celery Tasks (optional) # ========================================================================= tasks_path: str | None = None # Python import path, e.g., "app.modules.billing.tasks" scheduled_tasks: list[ScheduledTask] = field(default_factory=list) # ========================================================================= # Context Providers (Module-Driven Page Context) # ========================================================================= # Callables that provide context data for page templates per frontend type. # Each provider receives (request, db, platform) and returns a dict. # This enables modules to contribute context only when enabled for a platform. # # Example: # def get_billing_platform_context(request, db, platform): # from app.modules.billing.models import TIER_LIMITS # return {"tiers": get_tiers_data()} # # billing_module = ModuleDefinition( # code="billing", # context_providers={ # FrontendType.PLATFORM: get_billing_platform_context, # }, # ) context_providers: dict[FrontendType, Callable[..., dict[str, Any]]] = field(default_factory=dict) # ========================================================================= # Menu Item Methods (Legacy - uses menu_items dict of IDs) # ========================================================================= def get_menu_items(self, frontend_type: FrontendType) -> list[str]: """Get menu item IDs for a specific frontend type (legacy).""" return self.menu_items.get(frontend_type, []) def get_all_menu_items(self) -> set[str]: """Get all menu item IDs across all frontend types (legacy).""" all_items = set() for items in self.menu_items.values(): all_items.update(items) return all_items def has_menu_item(self, menu_item_id: str) -> bool: """Check if this module provides a specific menu item.""" return menu_item_id in self.get_all_menu_items() # ========================================================================= # Menu Definition Methods (New - uses menus dict of full definitions) # ========================================================================= def get_menu_sections(self, frontend_type: FrontendType) -> list[MenuSectionDefinition]: """ Get menu section definitions for a specific frontend type. Args: frontend_type: The frontend type to get menus for Returns: List of MenuSectionDefinition objects, sorted by order """ sections = self.menus.get(frontend_type, []) return sorted(sections, key=lambda s: s.order) def get_all_menu_definitions(self) -> dict[FrontendType, list[MenuSectionDefinition]]: """ Get all menu definitions for all frontend types. Returns: Dict mapping FrontendType to list of MenuSectionDefinition """ return self.menus def has_menus_for_frontend(self, frontend_type: FrontendType) -> bool: """Check if this module has menu definitions for a frontend type.""" return frontend_type in self.menus and len(self.menus[frontend_type]) > 0 def get_mandatory_menu_item_ids(self, frontend_type: FrontendType) -> set[str]: """ Get IDs of all mandatory menu items for a frontend type. Returns: Set of menu item IDs that are marked as is_mandatory=True """ mandatory_ids = set() for section in self.menus.get(frontend_type, []): for item in section.items: if item.is_mandatory: mandatory_ids.add(item.id) return mandatory_ids # ========================================================================= # Feature Methods # ========================================================================= def has_feature(self, feature_code: str) -> bool: """Check if this module provides a specific feature.""" return feature_code in self.features def has_permission(self, permission_id: str) -> bool: """Check if this module defines a specific permission.""" return any(p.id == permission_id for p in self.permissions) def get_permission_ids(self) -> set[str]: """Get all permission IDs defined by this module.""" return {p.id for p in self.permissions} # ========================================================================= # Dependency Methods # ========================================================================= def check_dependencies(self, enabled_modules: set[str]) -> list[str]: """ Check if all required modules are enabled. Args: enabled_modules: Set of enabled module codes Returns: List of missing required module codes """ return [req for req in self.requires if req not in enabled_modules] # ========================================================================= # Configuration Methods # ========================================================================= def validate_config(self, config: dict[str, Any]) -> dict[str, Any]: """ Validate configuration against the schema. Args: config: Configuration dict to validate Returns: Validated configuration dict Raises: ValidationError: If configuration is invalid """ if self.config_schema is None: return config # Merge with defaults merged = {**self.default_config, **config} # Validate using Pydantic model validated = self.config_schema(**merged) return validated.model_dump() def get_default_config(self) -> dict[str, Any]: """Get the default configuration for this module.""" if self.config_schema is None: return self.default_config.copy() # Use Pydantic model defaults return self.config_schema().model_dump() # ========================================================================= # Lifecycle Methods # ========================================================================= def run_on_enable(self, platform_id: int) -> None: """Run the on_enable hook if defined.""" if self.on_enable: self.on_enable(platform_id) def run_on_disable(self, platform_id: int) -> None: """Run the on_disable hook if defined.""" if self.on_disable: self.on_disable(platform_id) def run_on_startup(self) -> None: """Run the on_startup hook if defined.""" if self.on_startup: self.on_startup() def get_health_status(self) -> dict[str, Any]: """ Get the health status of this module. Returns: Dict with at least a 'status' key ('healthy', 'degraded', 'unhealthy') """ if self.health_check: return self.health_check() return {"status": "healthy", "message": "No health check defined"} # ========================================================================= # Self-Contained Module Methods # ========================================================================= def get_module_dir(self) -> Path: """ Get the filesystem path to this module's directory. Returns: Path to app/modules// """ # Handle module codes with hyphens (e.g., dev-tools -> dev_tools) dir_name = self.code.replace("-", "_") return Path(__file__).parent / dir_name def get_templates_dir(self) -> Path | None: """ Get the filesystem path to this module's templates directory. Returns: Path to templates directory, or None if not self-contained """ if not self.is_self_contained or not self.templates_path: return None return self.get_module_dir() / self.templates_path def get_locales_dir(self) -> Path | None: """ Get the filesystem path to this module's locales directory. Returns: Path to locales directory, or None if not configured """ if not self.is_self_contained or not self.locales_path: return None return self.get_module_dir() / self.locales_path def get_migrations_dir(self) -> Path | None: """ Get the filesystem path to this module's migrations directory. Returns: Path to migrations directory, or None if not configured """ if not self.migrations_path: return None return self.get_module_dir() / self.migrations_path def get_import_path(self, component: str) -> str | None: """ Get the Python import path for a module component. Args: component: One of "services", "models", "schemas", "exceptions", "tasks" Returns: Import path string, or None if not configured """ paths = { "services": self.services_path, "models": self.models_path, "schemas": self.schemas_path, "exceptions": self.exceptions_path, "tasks": self.tasks_path, } return paths.get(component) def get_tasks_module(self) -> str | None: """ Get the Python import path for this module's tasks. Returns the explicitly configured tasks_path, or infers it from the module code if the module is self-contained. Returns: Import path string (e.g., "app.modules.billing.tasks"), or None """ if self.tasks_path: return self.tasks_path if self.is_self_contained: dir_name = self.code.replace("-", "_") return f"app.modules.{dir_name}.tasks" return None def get_tasks_dir(self) -> Path | None: """ Get the filesystem path to this module's tasks directory. Returns: Path to tasks directory, or None if not configured """ tasks_module = self.get_tasks_module() if not tasks_module: return None return self.get_module_dir() / "tasks" def has_tasks(self) -> bool: """ Check if this module has a tasks directory. Returns: True if tasks directory exists and contains __init__.py """ tasks_dir = self.get_tasks_dir() if not tasks_dir: return False return tasks_dir.exists() and (tasks_dir / "__init__.py").exists() def validate_structure(self) -> list[str]: """ Validate that self-contained module has expected directory structure. Returns: List of missing or invalid paths (empty if valid) """ if not self.is_self_contained: return [] issues = [] module_dir = self.get_module_dir() if not module_dir.exists(): issues.append(f"Module directory not found: {module_dir}") return issues # Check optional directories expected_dirs = [] if self.services_path: expected_dirs.append("services") if self.models_path: expected_dirs.append("models") if self.schemas_path: expected_dirs.append("schemas") if self.templates_path: expected_dirs.append(self.templates_path) if self.locales_path: expected_dirs.append(self.locales_path) if self.migrations_path: expected_dirs.append(self.migrations_path) for dir_name in expected_dirs: dir_path = module_dir / dir_name if not dir_path.exists(): issues.append(f"Missing directory: {dir_path}") return issues # ========================================================================= # Classification Methods # ========================================================================= def get_tier(self) -> str: """ Get the tier classification of this module. Returns: 'core', 'internal', or 'optional' """ if self.is_core: return "core" elif self.is_internal: return "internal" else: return "optional" # ========================================================================= # Context Provider Methods # ========================================================================= def has_context_provider(self, frontend_type: FrontendType) -> bool: """Check if this module has a context provider for a frontend type.""" return frontend_type in self.context_providers def get_context_contribution( self, frontend_type: FrontendType, request: Any, db: Any, platform: Any, ) -> dict[str, Any]: """ Get context contribution from this module for a frontend type. Args: frontend_type: The frontend type (PLATFORM, ADMIN, VENDOR, STOREFRONT) request: FastAPI Request object db: Database session platform: Platform object (may be None for some contexts) Returns: Dict of context variables, or empty dict if no provider Note: Exceptions are caught and logged by the caller. This method may raise if the provider fails. """ provider = self.context_providers.get(frontend_type) if provider is None: return {} return provider(request, db, platform) def get_supported_frontend_types(self) -> list[FrontendType]: """Get list of frontend types this module provides context for.""" return list(self.context_providers.keys()) # ========================================================================= # Magic Methods # ========================================================================= def __hash__(self) -> int: return hash(self.code) def __eq__(self, other: object) -> bool: if isinstance(other, ModuleDefinition): return self.code == other.code return False def __repr__(self) -> str: tier = self.get_tier() sc = ", self_contained" if self.is_self_contained else "" return f""