Files
orion/app/modules/base.py
Samir Boulahtit ef9ea29643 feat: module-driven onboarding system + simplified 3-step signup
Add OnboardingProviderProtocol so modules declare their own post-signup
onboarding steps. The core OnboardingAggregator discovers enabled
providers and exposes a dashboard API (GET /dashboard/onboarding).
A session-scoped banner on the store dashboard shows a checklist that
guides merchants through setup without blocking signup.

Signup is simplified from 4 steps to 3 (Plan → Account → Payment):
store creation is merged into account creation, store language is
captured from the user's browsing language, and platform-specific
template branching is removed.

Includes 47 unit and integration tests covering all new providers,
the aggregator, the API endpoint, and the signup service changes.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-28 23:39:42 +01:00

1016 lines
39 KiB
Python

# 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/<code>/
├── __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 collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from fastapi import APIRouter
from pydantic import BaseModel
from app.modules.contracts.audit import AuditProviderProtocol
from app.modules.contracts.cms import MediaUsageProviderProtocol
from app.modules.contracts.features import FeatureProviderProtocol
from app.modules.contracts.metrics import MetricsProviderProtocol
from app.modules.contracts.onboarding import OnboardingProviderProtocol
from app.modules.contracts.widgets import DashboardWidgetProviderProtocol
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 {store_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 store 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
store_router: FastAPI router for store 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.STORE: ["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
store_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)
# =========================================================================
# Metrics Provider (Module-Driven Statistics)
# =========================================================================
# Callable that returns a MetricsProviderProtocol implementation.
# Use a callable (factory function) to enable lazy loading and avoid
# circular imports. Each module can provide its own metrics for dashboards.
#
# Example:
# def _get_metrics_provider():
# from app.modules.orders.services.order_metrics import order_metrics_provider
# return order_metrics_provider
#
# orders_module = ModuleDefinition(
# code="orders",
# metrics_provider=_get_metrics_provider,
# )
#
# The provider will be discovered by core's StatsAggregator service.
metrics_provider: "Callable[[], MetricsProviderProtocol] | None" = None
# =========================================================================
# Widget Provider (Module-Driven Dashboard Widgets)
# =========================================================================
# Callable that returns a DashboardWidgetProviderProtocol implementation.
# Use a callable (factory function) to enable lazy loading and avoid
# circular imports. Each module can provide its own widgets for dashboards.
#
# Example:
# def _get_widget_provider():
# from app.modules.orders.services.order_widgets import order_widget_provider
# return order_widget_provider
#
# orders_module = ModuleDefinition(
# code="orders",
# widget_provider=_get_widget_provider,
# )
#
# The provider will be discovered by core's WidgetAggregator service.
widget_provider: "Callable[[], DashboardWidgetProviderProtocol] | None" = None
# =========================================================================
# Audit Provider (Module-Driven Audit Logging)
# =========================================================================
# Callable that returns an AuditProviderProtocol implementation.
# Use a callable (factory function) to enable lazy loading and avoid
# circular imports. Modules can provide audit logging backends.
#
# Example:
# def _get_audit_provider():
# from app.modules.monitoring.services.audit_provider import audit_provider
# return audit_provider
#
# monitoring_module = ModuleDefinition(
# code="monitoring",
# audit_provider=_get_audit_provider,
# )
#
# The provider will be discovered by core's AuditAggregator service.
audit_provider: "Callable[[], AuditProviderProtocol] | None" = None
# =========================================================================
# Feature Provider (Module-Driven Billable Features)
# =========================================================================
# Callable that returns a FeatureProviderProtocol implementation.
# Use a callable (factory function) to enable lazy loading and avoid
# circular imports. Each module can declare its billable features
# and provide usage tracking for limit enforcement.
#
# Example:
# def _get_feature_provider():
# from app.modules.catalog.services.catalog_features import catalog_feature_provider
# return catalog_feature_provider
#
# catalog_module = ModuleDefinition(
# code="catalog",
# feature_provider=_get_feature_provider,
# )
#
# The provider will be discovered by billing's FeatureAggregator service.
feature_provider: "Callable[[], FeatureProviderProtocol] | None" = None
# =========================================================================
# Media Usage Provider (Module-Driven Media Usage Tracking)
# =========================================================================
# Callable that returns a MediaUsageProviderProtocol implementation.
# Modules that use media files (catalog, etc.) can register a provider
# to report where media is being used.
media_usage_provider: "Callable[[], MediaUsageProviderProtocol] | None" = None
# =========================================================================
# Onboarding Provider (Module-Driven Post-Signup Onboarding)
# =========================================================================
# Callable that returns an OnboardingProviderProtocol implementation.
# Modules declare onboarding steps (what needs to be configured after signup)
# and provide completion checks. The core module's OnboardingAggregator
# discovers and aggregates all providers into a dashboard checklist banner.
#
# Example:
# def _get_onboarding_provider():
# from app.modules.marketplace.services.marketplace_onboarding import (
# marketplace_onboarding_provider,
# )
# return marketplace_onboarding_provider
#
# marketplace_module = ModuleDefinition(
# code="marketplace",
# onboarding_provider=_get_onboarding_provider,
# )
#
# The provider will be discovered by core's OnboardingAggregator service.
onboarding_provider: "Callable[[], OnboardingProviderProtocol] | None" = None
# =========================================================================
# 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/<code>/
"""
# 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"
if self.is_internal:
return "internal"
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, STORE, 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())
# =========================================================================
# Metrics Provider Methods
# =========================================================================
def has_metrics_provider(self) -> bool:
"""Check if this module has a metrics provider."""
return self.metrics_provider is not None
def get_metrics_provider_instance(self) -> "MetricsProviderProtocol | None":
"""
Get the metrics provider instance for this module.
Calls the metrics_provider factory function to get the provider.
Returns None if no provider is configured.
Returns:
MetricsProviderProtocol instance, or None
"""
if self.metrics_provider is None:
return None
return self.metrics_provider()
# =========================================================================
# Widget Provider Methods
# =========================================================================
def has_widget_provider(self) -> bool:
"""Check if this module has a widget provider."""
return self.widget_provider is not None
def get_widget_provider_instance(self) -> "DashboardWidgetProviderProtocol | None":
"""
Get the widget provider instance for this module.
Calls the widget_provider factory function to get the provider.
Returns None if no provider is configured.
Returns:
DashboardWidgetProviderProtocol instance, or None
"""
if self.widget_provider is None:
return None
return self.widget_provider()
# =========================================================================
# Audit Provider Methods
# =========================================================================
def has_audit_provider(self) -> bool:
"""Check if this module has an audit provider."""
return self.audit_provider is not None
def get_audit_provider_instance(self) -> "AuditProviderProtocol | None":
"""
Get the audit provider instance for this module.
Calls the audit_provider factory function to get the provider.
Returns None if no provider is configured.
Returns:
AuditProviderProtocol instance, or None
"""
if self.audit_provider is None:
return None
return self.audit_provider()
# =========================================================================
# Feature Provider Methods
# =========================================================================
def has_feature_provider(self) -> bool:
"""Check if this module has a feature provider."""
return self.feature_provider is not None
def get_feature_provider_instance(self) -> "FeatureProviderProtocol | None":
"""
Get the feature provider instance for this module.
Calls the feature_provider factory function to get the provider.
Returns None if no provider is configured.
Returns:
FeatureProviderProtocol instance, or None
"""
if self.feature_provider is None:
return None
return self.feature_provider()
# =========================================================================
# Media Usage Provider Methods
# =========================================================================
def has_media_usage_provider(self) -> bool:
"""Check if this module has a media usage provider."""
return self.media_usage_provider is not None
def get_media_usage_provider_instance(self) -> "MediaUsageProviderProtocol | None":
"""Get the media usage provider instance for this module.
Returns:
MediaUsageProviderProtocol instance, or None
"""
if self.media_usage_provider is None:
return None
return self.media_usage_provider()
# =========================================================================
# Onboarding Provider Methods
# =========================================================================
def has_onboarding_provider(self) -> bool:
"""Check if this module has an onboarding provider."""
return self.onboarding_provider is not None
def get_onboarding_provider_instance(self) -> "OnboardingProviderProtocol | None":
"""Get the onboarding provider instance for this module.
Returns:
OnboardingProviderProtocol instance, or None
"""
if self.onboarding_provider is None:
return None
return self.onboarding_provider()
# =========================================================================
# 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"<Module({self.code}, tier={tier}{sc})>"