sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement metrics provider pattern for modular dashboard statistics

Browse Source 
This commit introduces a protocol-based metrics architecture that allows
each module to provide its own statistics for dashboards without creating
cross-module dependencies.

Key changes:
- Add MetricsProviderProtocol and MetricValue dataclass in contracts module
- Add StatsAggregatorService in core module that discovers and aggregates
  metrics from all enabled modules
- Implement metrics providers for all modules:
  - tenancy: vendor/user counts, team members, domains
  - customers: customer counts
  - cms: pages, media files
  - catalog: products
  - inventory: stock levels
  - orders: order counts, revenue
  - marketplace: import jobs, staging products
- Update dashboard routes to use StatsAggregator instead of direct imports
- Fix VendorPlatform junction table usage (Vendor.platform_id doesn't exist)
- Add comprehensive documentation for the pattern

This architecture ensures:
- Dashboards always work (aggregator in core)
- Each module owns its metrics (no cross-module coupling)
- Optional modules are truly optional (can be removed without breaking app)
- Multi-platform vendors are properly supported via VendorPlatform table

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-02-03 21:11:29 +01:00
parent a76128e016
commit a8fae0fbc7
28 changed files with 3745 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
app/modules/core/services/__init__.py
View File

@@ -28,6 +28,10 @@ from app.modules.core.services.platform_settings_service import (
PlatformSettingsService,
platform_settings_service,
)
from app.modules.core.services.stats_aggregator import (
StatsAggregatorService,
stats_aggregator,
)
from app.modules.core.services.storage_service import (
LocalStorageBackend,
R2StorageBackend,
@@ -64,4 +68,7 @@ __all__ = [
# Platform settings
"PlatformSettingsService",
"platform_settings_service",
# Stats aggregator
"StatsAggregatorService",
"stats_aggregator",
]

253
app/modules/core/services/stats_aggregator.py Normal file
View File

@@ -0,0 +1,253 @@
# app/modules/core/services/stats_aggregator.py
"""
Stats aggregator service for collecting metrics from all modules.
This service lives in core because dashboards are core functionality that should
always be available. It discovers and aggregates MetricsProviders from all enabled
modules, providing a unified interface for dashboard statistics.
Benefits:
- Dashboards always work (aggregator is in core)
- Each module owns its metrics (no cross-module coupling)
- Optional modules are truly optional (can be removed without breaking app)
- Easy to add new metrics (just implement MetricsProviderProtocol in your module)
Usage:
from app.modules.core.services.stats_aggregator import stats_aggregator
# Get vendor dashboard stats
stats = stats_aggregator.get_vendor_dashboard_stats(
db=db, vendor_id=123, platform_id=1
)
# Get admin dashboard stats
stats = stats_aggregator.get_admin_dashboard_stats(
db=db, platform_id=1
)
"""
import logging
from typing import TYPE_CHECKING, Any
from sqlalchemy.orm import Session
from app.modules.contracts.metrics import (
MetricValue,
MetricsContext,
MetricsProviderProtocol,
)
if TYPE_CHECKING:
from app.modules.base import ModuleDefinition
logger = logging.getLogger(__name__)
class StatsAggregatorService:
"""
Aggregates metrics from all module providers.
This service discovers MetricsProviders from enabled modules and provides
a unified interface for dashboard statistics. It handles graceful degradation
when modules are disabled or providers fail.
"""
def _get_enabled_providers(
self, db: Session, platform_id: int
) -> list[tuple["ModuleDefinition", MetricsProviderProtocol]]:
"""
Get metrics providers from enabled modules.
Args:
db: Database session
platform_id: Platform ID to check module enablement
Returns:
List of (module, provider) tuples for enabled modules with providers
"""
from app.modules.registry import MODULES
from app.modules.service import module_service
providers: list[tuple[ModuleDefinition, MetricsProviderProtocol]] = []
for module in MODULES.values():
# Skip modules without metrics providers
if not module.has_metrics_provider():
continue
# Core modules are always enabled, check others
if not module.is_core:
try:
if not module_service.is_module_enabled(db, platform_id, module.code):
continue
except Exception as e:
logger.warning(
f"Failed to check if module {module.code} is enabled: {e}"
)
continue
# Get the provider instance
try:
provider = module.get_metrics_provider_instance()
if provider is not None:
providers.append((module, provider))
except Exception as e:
logger.warning(
f"Failed to get metrics provider for module {module.code}: {e}"
)
return providers
def get_vendor_dashboard_stats(
self,
db: Session,
vendor_id: int,
platform_id: int,
context: MetricsContext | None = None,
) -> dict[str, list[MetricValue]]:
"""
Get all metrics for a vendor, grouped by category.
Called by the vendor dashboard to display vendor-scoped statistics.
Args:
db: Database session
vendor_id: ID of the vendor to get metrics for
platform_id: Platform ID (for module enablement check)
context: Optional filtering/scoping context
Returns:
Dict mapping category name to list of MetricValue objects
"""
providers = self._get_enabled_providers(db, platform_id)
result: dict[str, list[MetricValue]] = {}
for module, provider in providers:
try:
metrics = provider.get_vendor_metrics(db, vendor_id, context)
if metrics:
result[provider.metrics_category] = metrics
except Exception as e:
logger.warning(
f"Failed to get vendor metrics from module {module.code}: {e}"
)
# Continue with other providers - graceful degradation
return result
def get_admin_dashboard_stats(
self,
db: Session,
platform_id: int,
context: MetricsContext | None = None,
) -> dict[str, list[MetricValue]]:
"""
Get all metrics for a platform, grouped by category.
Called by the admin dashboard to display platform-wide statistics.
Args:
db: Database session
platform_id: ID of the platform to get metrics for
context: Optional filtering/scoping context
Returns:
Dict mapping category name to list of MetricValue objects
"""
providers = self._get_enabled_providers(db, platform_id)
result: dict[str, list[MetricValue]] = {}
for module, provider in providers:
try:
metrics = provider.get_platform_metrics(db, platform_id, context)
if metrics:
result[provider.metrics_category] = metrics
except Exception as e:
logger.warning(
f"Failed to get platform metrics from module {module.code}: {e}"
)
# Continue with other providers - graceful degradation
return result
def get_vendor_stats_flat(
self,
db: Session,
vendor_id: int,
platform_id: int,
context: MetricsContext | None = None,
) -> dict[str, Any]:
"""
Get vendor metrics as a flat dictionary.
This is a convenience method that flattens the category-grouped metrics
into a single dictionary with metric keys as keys. Useful for backward
compatibility with existing dashboard code.
Args:
db: Database session
vendor_id: ID of the vendor to get metrics for
platform_id: Platform ID (for module enablement check)
context: Optional filtering/scoping context
Returns:
Flat dict mapping metric keys to values
"""
categorized = self.get_vendor_dashboard_stats(db, vendor_id, platform_id, context)
return self._flatten_metrics(categorized)
def get_admin_stats_flat(
self,
db: Session,
platform_id: int,
context: MetricsContext | None = None,
) -> dict[str, Any]:
"""
Get platform metrics as a flat dictionary.
This is a convenience method that flattens the category-grouped metrics
into a single dictionary with metric keys as keys. Useful for backward
compatibility with existing dashboard code.
Args:
db: Database session
platform_id: Platform ID
context: Optional filtering/scoping context
Returns:
Flat dict mapping metric keys to values
"""
categorized = self.get_admin_dashboard_stats(db, platform_id, context)
return self._flatten_metrics(categorized)
def _flatten_metrics(
self, categorized: dict[str, list[MetricValue]]
) -> dict[str, Any]:
"""Flatten categorized metrics into a single dictionary."""
flat: dict[str, Any] = {}
for metrics in categorized.values():
for metric in metrics:
flat[metric.key] = metric.value
return flat
def get_available_categories(
self, db: Session, platform_id: int
) -> list[str]:
"""
Get list of available metric categories for a platform.
Args:
db: Database session
platform_id: Platform ID (for module enablement check)
Returns:
List of category names from enabled providers
"""
providers = self._get_enabled_providers(db, platform_id)
return [provider.metrics_category for _, provider in providers]
# Singleton instance
stats_aggregator = StatsAggregatorService()
__all__ = ["StatsAggregatorService", "stats_aggregator"]