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

feat: implement DashboardWidgetProvider pattern for modular dashboard widgets

Browse Source 
Add protocol-based widget system following the MetricsProvider pattern:

- Create DashboardWidgetProviderProtocol in contracts/widgets.py
- Add WidgetAggregatorService in core to discover and aggregate widgets
- Implement MarketplaceWidgetProvider for recent_imports widget
- Implement TenancyWidgetProvider for recent_vendors widget
- Update admin dashboard to use widget_aggregator
- Add widget_provider field to ModuleDefinition

Architecture documentation:
- Add widget-provider-pattern.md with implementation guide
- Add cross-module-import-rules.md enforcing core/optional separation
- Update module-system.md with widget_provider and import rules

This enables modules to provide rich dashboard widgets without core modules
importing from optional modules, maintaining true module independence.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-02-04 19:01:23 +01:00
parent a8fae0fbc7
commit 3e38db79aa
13 changed files with 1906 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
app/modules/base.py
View File

@@ -45,6 +45,7 @@ if TYPE_CHECKING:
from pydantic import BaseModel
from app.modules.contracts.metrics import MetricsProviderProtocol
from app.modules.contracts.widgets import DashboardWidgetProviderProtocol
from app.modules.enums import FrontendType
@@ -413,6 +414,26 @@ class ModuleDefinition:
# 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
# =========================================================================
# Menu Item Methods (Legacy - uses menu_items dict of IDs)
# =========================================================================
@@ -799,6 +820,28 @@ class ModuleDefinition:
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()
# =========================================================================
# Magic Methods
# =========================================================================

28
app/modules/contracts/__init__.py
View File

@@ -32,6 +32,17 @@ Metrics Provider Pattern:
def get_vendor_metrics(self, db, vendor_id, context=None) -> list[MetricValue]:
return [MetricValue(key="orders.total", value=42, label="Total", category="orders")]
Widget Provider Pattern:
from app.modules.contracts.widgets import DashboardWidgetProviderProtocol, DashboardWidget
class OrderWidgetProvider:
@property
def widgets_category(self) -> str:
return "orders"
def get_vendor_widgets(self, db, vendor_id, context=None) -> list[DashboardWidget]:
return [DashboardWidget(key="orders.recent", widget_type="list", ...)]
"""
from app.modules.contracts.base import ServiceProtocol
@@ -41,6 +52,15 @@ from app.modules.contracts.metrics import (
MetricsContext,
MetricsProviderProtocol,
)
from app.modules.contracts.widgets import (
BreakdownWidget,
DashboardWidget,
DashboardWidgetProviderProtocol,
ListWidget,
WidgetBreakdownItem,
WidgetContext,
WidgetListItem,
)
__all__ = [
# Base protocols
@@ -51,4 +71,12 @@ __all__ = [
"MetricValue",
"MetricsContext",
"MetricsProviderProtocol",
# Widget protocols
"WidgetContext",
"WidgetListItem",
"WidgetBreakdownItem",
"ListWidget",
"BreakdownWidget",
"DashboardWidget",
"DashboardWidgetProviderProtocol",
]

348
app/modules/contracts/widgets.py Normal file
View File

@@ -0,0 +1,348 @@
# app/modules/contracts/widgets.py
"""
Dashboard widget provider protocol for cross-module widget aggregation.
This module defines the protocol that modules implement to expose dashboard widgets.
The core module's WidgetAggregator discovers and aggregates all providers.
Benefits:
- Each module owns its widgets (no cross-module coupling)
- Dashboards always work (aggregator is in core)
- Optional modules are truly optional (can be removed without breaking app)
- Easy to add new widgets (just implement protocol in your module)
Widget Types:
- ListWidget: Displays a list of items (recent imports, recent vendors, etc.)
- BreakdownWidget: Displays grouped statistics (by category, by status, etc.)
Usage:
# 1. Implement the protocol in your module
class OrderWidgetProvider:
@property
def widgets_category(self) -> str:
return "orders"
def get_vendor_widgets(self, db, vendor_id, context=None) -> list[DashboardWidget]:
return [
DashboardWidget(
key="orders.recent",
widget_type="list",
title="Recent Orders",
category="orders",
data=ListWidget(items=[...]),
)
]
# 2. Register in module definition
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,
# ...
)
# 3. Widgets appear automatically in dashboards when module is enabled
"""
from dataclasses import dataclass, field
from datetime import datetime
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
if TYPE_CHECKING:
from sqlalchemy.orm import Session
# =============================================================================
# Context
# =============================================================================
@dataclass
class WidgetContext:
"""
Context for widget data collection.
Provides filtering and scoping options for widget providers.
Attributes:
date_from: Start of date range filter
date_to: End of date range filter
limit: Maximum number of items for list widgets
include_details: Whether to include extra details (may be expensive)
"""
date_from: datetime | None = None
date_to: datetime | None = None
limit: int = 5
include_details: bool = False
# =============================================================================
# Widget Item Types
# =============================================================================
@dataclass
class WidgetListItem:
"""
Single item in a list widget (recent vendors, orders, imports).
Attributes:
id: Unique identifier for the item
title: Primary display text
subtitle: Secondary display text (optional)
status: Status indicator ("success", "warning", "error", "neutral")
timestamp: When the item was created/updated
url: Link to item details (optional)
metadata: Additional data for rendering
"""
id: int | str
title: str
subtitle: str | None = None
status: str | None = None # "success", "warning", "error", "neutral"
timestamp: datetime | None = None
url: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class WidgetBreakdownItem:
"""
Single row in a breakdown widget (stats by marketplace, category).
Attributes:
label: Display label for the row
value: Primary numeric value
secondary_value: Optional secondary value (e.g., comparison)
percentage: Percentage representation
icon: Lucide icon name (optional)
"""
label: str
value: int | float
secondary_value: int | float | None = None
percentage: float | None = None
icon: str | None = None
# =============================================================================
# Widget Containers
# =============================================================================
@dataclass
class ListWidget:
"""
Widget containing a list of items.
Used for: recent imports, recent vendors, recent orders, etc.
Attributes:
items: List of WidgetListItem objects
total_count: Total number of items (for "X more" display)
view_all_url: Link to view all items
"""
items: list[WidgetListItem]
total_count: int | None = None
view_all_url: str | None = None
@dataclass
class BreakdownWidget:
"""
Widget containing grouped statistics.
Used for: status breakdown, category distribution, etc.
Attributes:
items: List of WidgetBreakdownItem objects
total: Sum of all values
"""
items: list[WidgetBreakdownItem]
total: int | float | None = None
# Type alias for widget data
WidgetData = ListWidget | BreakdownWidget
# =============================================================================
# Main Widget Envelope
# =============================================================================
@dataclass
class DashboardWidget:
"""
Standard widget envelope with metadata.
This is the main unit of data returned by widget providers.
It contains both the widget data and metadata for display.
Attributes:
key: Unique identifier for this widget (e.g., "marketplace.recent_imports")
Format: "{category}.{widget_name}" for consistency
widget_type: Type of widget ("list" or "breakdown")
title: Human-readable title for display
category: Grouping category (should match widgets_category of provider)
data: The actual widget data (ListWidget or BreakdownWidget)
icon: Optional Lucide icon name for UI display
description: Optional longer description
order: Display order (lower = higher priority)
Example:
DashboardWidget(
key="marketplace.recent_imports",
widget_type="list",
title="Recent Imports",
category="marketplace",
data=ListWidget(
items=[
WidgetListItem(id=1, title="Import #1", status="success")
],
),
icon="download",
order=10,
)
"""
key: str
widget_type: str # "list" or "breakdown"
title: str
category: str
data: WidgetData
icon: str | None = None
description: str | None = None
order: int = 100
# =============================================================================
# Protocol Interface
# =============================================================================
@runtime_checkable
class DashboardWidgetProviderProtocol(Protocol):
"""
Protocol for modules that provide dashboard widgets.
Each module implements this to expose its own widgets.
The core module's WidgetAggregator discovers and aggregates all providers.
Implementation Notes:
- Providers should be stateless (all data via db session)
- Return empty list if no widgets available (don't raise)
- Use consistent key format: "{category}.{widget_name}"
- Include icon hints for UI rendering
- Be mindful of query performance (use efficient queries)
- Translations are handled by the module itself
Example Implementation:
class MarketplaceWidgetProvider:
@property
def widgets_category(self) -> str:
return "marketplace"
def get_vendor_widgets(
self, db: Session, vendor_id: int, context: WidgetContext | None = None
) -> list[DashboardWidget]:
from app.modules.marketplace.models import MarketplaceImportJob
limit = context.limit if context else 5
jobs = db.query(MarketplaceImportJob).filter(...).limit(limit).all()
return [
DashboardWidget(
key="marketplace.recent_imports",
widget_type="list",
title="Recent Imports",
category="marketplace",
data=ListWidget(items=[...]),
icon="download"
)
]
def get_platform_widgets(
self, db: Session, platform_id: int, context: WidgetContext | None = None
) -> list[DashboardWidget]:
# Aggregate across all vendors in platform
...
"""
@property
def widgets_category(self) -> str:
"""
Category name for this provider's widgets.
Should be a short, lowercase identifier matching the module's domain.
Examples: "orders", "inventory", "customers", "marketplace"
Returns:
Category string used for grouping widgets
"""
...
def get_vendor_widgets(
self,
db: "Session",
vendor_id: int,
context: WidgetContext | None = None,
) -> list[DashboardWidget]:
"""
Get widgets for a specific vendor dashboard.
Called by the vendor dashboard to display vendor-scoped widgets.
Should only include data belonging to the specified vendor.
Args:
db: Database session for queries
vendor_id: ID of the vendor to get widgets for
context: Optional filtering/scoping context
Returns:
List of DashboardWidget objects for this vendor
"""
...
def get_platform_widgets(
self,
db: "Session",
platform_id: int,
context: WidgetContext | None = None,
) -> list[DashboardWidget]:
"""
Get widgets aggregated for a platform.
Called by the admin dashboard to display platform-wide widgets.
Should aggregate data across all vendors in the platform.
Args:
db: Database session for queries
platform_id: ID of the platform to get widgets for
context: Optional filtering/scoping context
Returns:
List of DashboardWidget objects aggregated for the platform
"""
...
__all__ = [
# Context
"WidgetContext",
# Item types
"WidgetListItem",
"WidgetBreakdownItem",
# Containers
"ListWidget",
"BreakdownWidget",
"WidgetData",
# Main envelope
"DashboardWidget",
# Protocol
"DashboardWidgetProviderProtocol",
]

45
app/modules/core/routes/api/admin_dashboard.py
View File

@@ -5,17 +5,22 @@ Admin dashboard and statistics endpoints.
This module uses the StatsAggregator service from core to collect metrics from all
enabled modules. Each module provides its own metrics via the MetricsProvider protocol.
Dashboard widgets are collected via the WidgetAggregator service, which discovers
DashboardWidgetProvider implementations from all enabled modules.
For backward compatibility, this also falls back to the analytics stats_service
for comprehensive statistics that haven't been migrated to the provider pattern yet.
"""
import logging
from typing import Any
from fastapi import APIRouter, Depends, Request
from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_api
from app.core.database import get_db
from app.modules.contracts.widgets import ListWidget
from app.modules.core.schemas.dashboard import (
AdminDashboardResponse,
ImportStatsResponse,
@@ -28,7 +33,7 @@ from app.modules.core.schemas.dashboard import (
VendorStatsResponse,
)
from app.modules.core.services.stats_aggregator import stats_aggregator
from app.modules.tenancy.services.admin_service import admin_service
from app.modules.core.services.widget_aggregator import widget_aggregator
from models.schema.auth import UserContext
admin_dashboard_router = APIRouter(prefix="/dashboard")
@@ -77,6 +82,31 @@ def _extract_metric_value(
return default
def _widget_list_item_to_dict(item) -> dict[str, Any]:
"""Convert a WidgetListItem to a dictionary for API response."""
return {
"id": item.id,
"title": item.title,
"subtitle": item.subtitle,
"status": item.status,
"timestamp": item.timestamp,
"url": item.url,
**item.metadata,
}
def _extract_widget_items(
widgets: dict[str, list], category: str, key: str
) -> list[dict[str, Any]]:
"""Extract items from a list widget for backward compatibility."""
if category not in widgets:
return []
for widget in widgets[category]:
if widget.key == key and isinstance(widget.data, ListWidget):
return [_widget_list_item_to_dict(item) for item in widget.data.items]
return []
@admin_dashboard_router.get("", response_model=AdminDashboardResponse)
def get_admin_dashboard(
request: Request,
@@ -89,6 +119,9 @@ def get_admin_dashboard(
# Get aggregated metrics from all enabled modules
metrics = stats_aggregator.get_admin_dashboard_stats(db=db, platform_id=platform_id)
# Get aggregated widgets from all enabled modules
widgets = widget_aggregator.get_admin_dashboard_widgets(db=db, platform_id=platform_id)
# Extract user stats from tenancy module
total_users = _extract_metric_value(metrics, "tenancy", "tenancy.total_users", 0)
active_users = _extract_metric_value(metrics, "tenancy", "tenancy.active_users", 0)
@@ -110,6 +143,12 @@ def get_admin_dashboard(
metrics, "tenancy", "tenancy.inactive_vendors", 0
)
# Extract recent_vendors from tenancy widget (backward compatibility)
recent_vendors = _extract_widget_items(widgets, "tenancy", "tenancy.recent_vendors")
# Extract recent_imports from marketplace widget (backward compatibility)
recent_imports = _extract_widget_items(widgets, "marketplace", "marketplace.recent_imports")
return AdminDashboardResponse(
platform={
"name": "Multi-Tenant Ecommerce Platform",
@@ -128,8 +167,8 @@ def get_admin_dashboard(
pending=int(pending_vendors),
inactive=int(inactive_vendors),
),
recent_vendors=admin_service.get_recent_vendors(db, limit=5),
recent_imports=admin_service.get_recent_import_jobs(db, limit=10),
recent_vendors=recent_vendors,
recent_imports=recent_imports,
)

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

@@ -39,6 +39,10 @@ from app.modules.core.services.storage_service import (
get_storage_backend,
reset_storage_backend,
)
from app.modules.core.services.widget_aggregator import (
WidgetAggregatorService,
widget_aggregator,
)
__all__ = [
# Auth
@@ -71,4 +75,7 @@ __all__ = [
# Stats aggregator
"StatsAggregatorService",
"stats_aggregator",
# Widget aggregator
"WidgetAggregatorService",
"widget_aggregator",
]

256
app/modules/core/services/widget_aggregator.py Normal file
View File

@@ -0,0 +1,256 @@
# app/modules/core/services/widget_aggregator.py
"""
Widget aggregator service for collecting dashboard widgets from all modules.
This service lives in core because dashboards are core functionality that should
always be available. It discovers and aggregates DashboardWidgetProviders from all
enabled modules, providing a unified interface for dashboard widgets.
Benefits:
- Dashboards always work (aggregator is in core)
- Each module owns its widgets (no cross-module coupling)
- Optional modules are truly optional (can be removed without breaking app)
- Easy to add new widgets (just implement DashboardWidgetProviderProtocol in your module)
Usage:
from app.modules.core.services.widget_aggregator import widget_aggregator
# Get vendor dashboard widgets
widgets = widget_aggregator.get_vendor_dashboard_widgets(
db=db, vendor_id=123, platform_id=1
)
# Get admin dashboard widgets
widgets = widget_aggregator.get_admin_dashboard_widgets(
db=db, platform_id=1
)
"""
import logging
from typing import TYPE_CHECKING
from sqlalchemy.orm import Session
from app.modules.contracts.widgets import (
DashboardWidget,
DashboardWidgetProviderProtocol,
WidgetContext,
)
if TYPE_CHECKING:
from app.modules.base import ModuleDefinition
logger = logging.getLogger(__name__)
class WidgetAggregatorService:
"""
Aggregates widgets from all module providers.
This service discovers DashboardWidgetProviders from enabled modules and provides
a unified interface for dashboard widgets. It handles graceful degradation
when modules are disabled or providers fail.
"""
def _get_enabled_providers(
self, db: Session, platform_id: int
) -> list[tuple["ModuleDefinition", DashboardWidgetProviderProtocol]]:
"""
Get widget 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, DashboardWidgetProviderProtocol]] = []
for module in MODULES.values():
# Skip modules without widget providers
if not module.has_widget_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_widget_provider_instance()
if provider is not None:
providers.append((module, provider))
except Exception as e:
logger.warning(
f"Failed to get widget provider for module {module.code}: {e}"
)
return providers
def get_vendor_dashboard_widgets(
self,
db: Session,
vendor_id: int,
platform_id: int,
context: WidgetContext | None = None,
) -> dict[str, list[DashboardWidget]]:
"""
Get all widgets for a vendor, grouped by category.
Called by the vendor dashboard to display vendor-scoped widgets.
Args:
db: Database session
vendor_id: ID of the vendor to get widgets for
platform_id: Platform ID (for module enablement check)
context: Optional filtering/scoping context
Returns:
Dict mapping category name to list of DashboardWidget objects
"""
providers = self._get_enabled_providers(db, platform_id)
result: dict[str, list[DashboardWidget]] = {}
for module, provider in providers:
try:
widgets = provider.get_vendor_widgets(db, vendor_id, context)
if widgets:
result[provider.widgets_category] = widgets
except Exception as e:
logger.warning(
f"Failed to get vendor widgets from module {module.code}: {e}"
)
# Continue with other providers - graceful degradation
return result
def get_admin_dashboard_widgets(
self,
db: Session,
platform_id: int,
context: WidgetContext | None = None,
) -> dict[str, list[DashboardWidget]]:
"""
Get all widgets for a platform, grouped by category.
Called by the admin dashboard to display platform-wide widgets.
Args:
db: Database session
platform_id: ID of the platform to get widgets for
context: Optional filtering/scoping context
Returns:
Dict mapping category name to list of DashboardWidget objects
"""
providers = self._get_enabled_providers(db, platform_id)
result: dict[str, list[DashboardWidget]] = {}
for module, provider in providers:
try:
widgets = provider.get_platform_widgets(db, platform_id, context)
if widgets:
result[provider.widgets_category] = widgets
except Exception as e:
logger.warning(
f"Failed to get platform widgets from module {module.code}: {e}"
)
# Continue with other providers - graceful degradation
return result
def get_widgets_flat(
self,
db: Session,
platform_id: int,
vendor_id: int | None = None,
context: WidgetContext | None = None,
) -> list[DashboardWidget]:
"""
Get widgets as a flat list sorted by order.
Convenience method that returns all widgets in a single list,
sorted by their order field.
Args:
db: Database session
platform_id: Platform ID
vendor_id: If provided, get vendor widgets; otherwise platform widgets
context: Optional filtering/scoping context
Returns:
Flat list of DashboardWidget objects sorted by order
"""
if vendor_id is not None:
categorized = self.get_vendor_dashboard_widgets(
db, vendor_id, platform_id, context
)
else:
categorized = self.get_admin_dashboard_widgets(db, platform_id, context)
# Flatten and sort by order
all_widgets: list[DashboardWidget] = []
for widgets in categorized.values():
all_widgets.extend(widgets)
return sorted(all_widgets, key=lambda w: w.order)
def get_widget_by_key(
self,
db: Session,
platform_id: int,
key: str,
vendor_id: int | None = None,
context: WidgetContext | None = None,
) -> DashboardWidget | None:
"""
Get a specific widget by its key.
Args:
db: Database session
platform_id: Platform ID
key: Widget key (e.g., "marketplace.recent_imports")
vendor_id: If provided, get vendor widget; otherwise platform widget
context: Optional filtering/scoping context
Returns:
The DashboardWidget with the specified key, or None if not found
"""
widgets = self.get_widgets_flat(db, platform_id, vendor_id, context)
for widget in widgets:
if widget.key == key:
return widget
return None
def get_available_categories(
self, db: Session, platform_id: int
) -> list[str]:
"""
Get list of available widget 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.widgets_category for _, provider in providers]
# Singleton instance
widget_aggregator = WidgetAggregatorService()
__all__ = ["WidgetAggregatorService", "widget_aggregator"]

9
app/modules/marketplace/definition.py
View File

@@ -33,6 +33,13 @@ def _get_metrics_provider():
return marketplace_metrics_provider
def _get_widget_provider():
"""Lazy import of widget provider to avoid circular imports."""
from app.modules.marketplace.services.marketplace_widgets import marketplace_widget_provider
return marketplace_widget_provider
# Marketplace module definition
marketplace_module = ModuleDefinition(
code="marketplace",
@@ -155,6 +162,8 @@ marketplace_module = ModuleDefinition(
],
# Metrics provider for dashboard statistics
metrics_provider=_get_metrics_provider,
# Widget provider for dashboard widgets
widget_provider=_get_widget_provider,
)

211
app/modules/marketplace/services/marketplace_widgets.py Normal file
View File

@@ -0,0 +1,211 @@
# app/modules/marketplace/services/marketplace_widgets.py
"""
Marketplace dashboard widget provider.
Provides widgets for marketplace-related data on vendor and admin dashboards.
Implements the DashboardWidgetProviderProtocol.
Widgets provided:
- recent_imports: List of recent import jobs with status
"""
import logging
from sqlalchemy.orm import Session
from app.modules.contracts.widgets import (
DashboardWidget,
DashboardWidgetProviderProtocol,
ListWidget,
WidgetContext,
WidgetListItem,
)
logger = logging.getLogger(__name__)
class MarketplaceWidgetProvider:
"""
Widget provider for marketplace module.
Provides dashboard widgets for import jobs and other marketplace data.
"""
@property
def widgets_category(self) -> str:
return "marketplace"
def _map_status_to_display(self, status: str) -> str:
"""Map job status to widget status indicator."""
status_map = {
"pending": "neutral",
"processing": "warning",
"completed": "success",
"completed_with_errors": "warning",
"failed": "error",
}
return status_map.get(status, "neutral")
def get_vendor_widgets(
self,
db: Session,
vendor_id: int,
context: WidgetContext | None = None,
) -> list[DashboardWidget]:
"""
Get marketplace widgets for a vendor dashboard.
Args:
db: Database session
vendor_id: ID of the vendor
context: Optional filtering/scoping context
Returns:
List of DashboardWidget objects for the vendor
"""
from app.modules.marketplace.models import MarketplaceImportJob
limit = context.limit if context else 5
# Get recent imports for this vendor
jobs = (
db.query(MarketplaceImportJob)
.filter(MarketplaceImportJob.vendor_id == vendor_id)
.order_by(MarketplaceImportJob.created_at.desc())
.limit(limit)
.all()
)
items = [
WidgetListItem(
id=job.id,
title=f"Import #{job.id}",
subtitle=f"{job.marketplace} - {job.language.upper()}",
status=self._map_status_to_display(job.status),
timestamp=job.created_at,
url=f"/vendor/marketplace/imports/{job.id}",
metadata={
"total_processed": job.total_processed or 0,
"imported_count": job.imported_count or 0,
"error_count": job.error_count or 0,
"status": job.status,
},
)
for job in jobs
]
# Get total count for "view all" indicator
total_count = (
db.query(MarketplaceImportJob)
.filter(MarketplaceImportJob.vendor_id == vendor_id)
.count()
)
return [
DashboardWidget(
key="marketplace.recent_imports",
widget_type="list",
title="Recent Imports",
category="marketplace",
data=ListWidget(
items=items,
total_count=total_count,
view_all_url="/vendor/marketplace/imports",
),
icon="download",
description="Latest product import jobs",
order=20,
)
]
def get_platform_widgets(
self,
db: Session,
platform_id: int,
context: WidgetContext | None = None,
) -> list[DashboardWidget]:
"""
Get marketplace widgets for the admin/platform dashboard.
Args:
db: Database session
platform_id: ID of the platform
context: Optional filtering/scoping context
Returns:
List of DashboardWidget objects for the platform
"""
from sqlalchemy.orm import joinedload
from app.modules.marketplace.models import MarketplaceImportJob
from app.modules.tenancy.models import Vendor, VendorPlatform
limit = context.limit if context else 5
# Get vendor IDs for this platform
vendor_ids_subquery = (
db.query(VendorPlatform.vendor_id)
.filter(VendorPlatform.platform_id == platform_id)
.subquery()
)
# Get recent imports across all vendors in the platform
jobs = (
db.query(MarketplaceImportJob)
.options(joinedload(MarketplaceImportJob.vendor))
.filter(MarketplaceImportJob.vendor_id.in_(vendor_ids_subquery))
.order_by(MarketplaceImportJob.created_at.desc())
.limit(limit)
.all()
)
items = [
WidgetListItem(
id=job.id,
title=f"Import #{job.id}",
subtitle=job.vendor.name if job.vendor else "Unknown Vendor",
status=self._map_status_to_display(job.status),
timestamp=job.created_at,
url=f"/admin/marketplace/imports/{job.id}",
metadata={
"vendor_id": job.vendor_id,
"vendor_code": job.vendor.vendor_code if job.vendor else None,
"marketplace": job.marketplace,
"total_processed": job.total_processed or 0,
"imported_count": job.imported_count or 0,
"error_count": job.error_count or 0,
"status": job.status,
},
)
for job in jobs
]
# Get total count for platform
total_count = (
db.query(MarketplaceImportJob)
.filter(MarketplaceImportJob.vendor_id.in_(vendor_ids_subquery))
.count()
)
return [
DashboardWidget(
key="marketplace.recent_imports",
widget_type="list",
title="Recent Imports",
category="marketplace",
data=ListWidget(
items=items,
total_count=total_count,
view_all_url="/admin/marketplace/letzshop",
),
icon="download",
description="Latest product import jobs across all vendors",
order=20,
)
]
# Singleton instance
marketplace_widget_provider = MarketplaceWidgetProvider()
__all__ = ["MarketplaceWidgetProvider", "marketplace_widget_provider"]

9
app/modules/tenancy/definition.py
View File

@@ -22,6 +22,13 @@ def _get_metrics_provider():
return tenancy_metrics_provider
def _get_widget_provider():
"""Lazy import of widget provider to avoid circular imports."""
from app.modules.tenancy.services.tenancy_widgets import tenancy_widget_provider
return tenancy_widget_provider
tenancy_module = ModuleDefinition(
code="tenancy",
name="Tenancy Management",
@@ -161,6 +168,8 @@ tenancy_module = ModuleDefinition(
exceptions_path="app.modules.tenancy.exceptions",
# Metrics provider for dashboard statistics
metrics_provider=_get_metrics_provider,
# Widget provider for dashboard widgets
widget_provider=_get_widget_provider,
)
__all__ = ["tenancy_module"]

156
app/modules/tenancy/services/tenancy_widgets.py Normal file
View File

@@ -0,0 +1,156 @@
# app/modules/tenancy/services/tenancy_widgets.py
"""
Tenancy dashboard widget provider.
Provides widgets for tenancy-related data on admin dashboards.
Implements the DashboardWidgetProviderProtocol.
Widgets provided:
- recent_vendors: List of recently created vendors with status
"""
import logging
from sqlalchemy.orm import Session
from app.modules.contracts.widgets import (
DashboardWidget,
DashboardWidgetProviderProtocol,
ListWidget,
WidgetContext,
WidgetListItem,
)
logger = logging.getLogger(__name__)
class TenancyWidgetProvider:
"""
Widget provider for tenancy module.
Provides dashboard widgets for vendors, users, and other tenancy data.
"""
@property
def widgets_category(self) -> str:
return "tenancy"
def _get_vendor_status(self, vendor) -> str:
"""Determine widget status indicator for a vendor."""
if not vendor.is_active:
return "neutral"
if not vendor.is_verified:
return "warning"
return "success"
def get_vendor_widgets(
self,
db: Session,
vendor_id: int,
context: WidgetContext | None = None,
) -> list[DashboardWidget]:
"""
Get tenancy widgets for a vendor dashboard.
Tenancy module doesn't provide vendor-scoped widgets
(vendors don't see other vendors).
Args:
db: Database session
vendor_id: ID of the vendor
context: Optional filtering/scoping context
Returns:
Empty list (no vendor-scoped tenancy widgets)
"""
# Tenancy widgets are platform/admin-only
return []
def get_platform_widgets(
self,
db: Session,
platform_id: int,
context: WidgetContext | None = None,
) -> list[DashboardWidget]:
"""
Get tenancy widgets for the admin/platform dashboard.
Args:
db: Database session
platform_id: ID of the platform
context: Optional filtering/scoping context
Returns:
List of DashboardWidget objects for the platform
"""
from sqlalchemy.orm import joinedload
from app.modules.tenancy.models import Vendor, VendorPlatform
limit = context.limit if context else 5
# Get vendor IDs for this platform
vendor_ids_subquery = (
db.query(VendorPlatform.vendor_id)
.filter(VendorPlatform.platform_id == platform_id)
.subquery()
)
# Get recent vendors for this platform
vendors = (
db.query(Vendor)
.options(joinedload(Vendor.company))
.filter(Vendor.id.in_(vendor_ids_subquery))
.order_by(Vendor.created_at.desc())
.limit(limit)
.all()
)
items = [
WidgetListItem(
id=vendor.id,
title=vendor.name,
subtitle=vendor.vendor_code,
status=self._get_vendor_status(vendor),
timestamp=vendor.created_at,
url=f"/admin/vendors/{vendor.id}",
metadata={
"vendor_code": vendor.vendor_code,
"subdomain": vendor.subdomain,
"is_active": vendor.is_active,
"is_verified": vendor.is_verified,
"company_name": vendor.company.name if vendor.company else None,
},
)
for vendor in vendors
]
# Get total vendor count for platform
total_count = (
db.query(Vendor)
.filter(Vendor.id.in_(vendor_ids_subquery))
.count()
)
return [
DashboardWidget(
key="tenancy.recent_vendors",
widget_type="list",
title="Recent Vendors",
category="tenancy",
data=ListWidget(
items=items,
total_count=total_count,
view_all_url="/admin/vendors",
),
icon="shopping-bag",
description="Recently created vendor accounts",
order=10,
)
]
# Singleton instance
tenancy_widget_provider = TenancyWidgetProvider()
__all__ = ["TenancyWidgetProvider", "tenancy_widget_provider"]