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

fix(billing): resolve 3 IMPORT-001 architecture violations in billing module

Browse Source 
Replace direct imports from optional modules (catalog, orders, analytics)
with provider pattern calls (stats_aggregator, feature_aggregator) and
move usage_service from analytics to billing where it belongs.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-02-08 15:34:29 +01:00
parent 82585b1363
commit 55751d95b9
5 changed files with 43 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/modules/billing/routes/api/store_usage.py
View File

@@ -18,7 +18,7 @@ from sqlalchemy.orm import Session
from app.api.deps import get_current_store_api, require_module_access
from app.core.database import get_db
from app.modules.analytics.services.usage_service import usage_service
from app.modules.billing.services.usage_service import usage_service
from app.modules.enums import FrontendType
from models.schema.auth import UserContext

16
app/modules/billing/services/__init__.py
View File

@@ -41,6 +41,15 @@ from app.modules.billing.services.platform_pricing_service import (
PlatformPricingService,
platform_pricing_service,
)
from app.modules.billing.services.usage_service import (
UsageService,
usage_service,
UsageData,
UsageMetricData,
TierInfoData,
UpgradeTierData,
LimitCheckData,
)
__all__ = [
"SubscriptionService",
@@ -63,4 +72,11 @@ __all__ = [
"capacity_forecast_service",
"PlatformPricingService",
"platform_pricing_service",
"UsageService",
"usage_service",
"UsageData",
"UsageMetricData",
"TierInfoData",
"UpgradeTierData",
"LimitCheckData",
]

29
app/modules/billing/services/capacity_forecast_service.py
View File

@@ -16,13 +16,14 @@ from decimal import Decimal
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.modules.catalog.models import Product
from app.modules.billing.models import (
CapacitySnapshot,
MerchantSubscription,
SubscriptionStatus,
)
from app.modules.tenancy.models import Store, StoreUser
from app.modules.contracts.metrics import MetricsContext
from app.modules.core.services.stats_aggregator import stats_aggregator
from app.modules.tenancy.models import Platform, Store, StoreUser
logger = logging.getLogger(__name__)
@@ -87,8 +88,17 @@ class CapacityForecastService:
or 0
)
# Resource metrics
total_products = db.query(func.count(Product.id)).scalar() or 0
# Resource metrics via provider pattern (avoids direct catalog/orders imports)
start_of_month = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
platform = db.query(Platform).first()
platform_id = platform.id if platform else 1
stats = stats_aggregator.get_admin_stats_flat(
db, platform_id,
context=MetricsContext(date_from=start_of_month),
)
total_products = stats.get("catalog.total_products", 0)
total_team = (
db.query(func.count(StoreUser.id))
.filter(StoreUser.is_active == True) # noqa: E712
@@ -96,15 +106,8 @@ class CapacityForecastService:
or 0
)
# Orders this month
start_of_month = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
from app.modules.orders.models import Order
total_orders = (
db.query(func.count(Order.id))
.filter(Order.created_at >= start_of_month)
.scalar() or 0
)
# Orders this month (from stats aggregator)
total_orders = stats.get("orders.in_period", 0)
# Storage metrics
try:

375
app/modules/billing/services/usage_service.py Normal file
View File

@@ -0,0 +1,375 @@
# app/modules/billing/services/usage_service.py
"""
Usage and limits service.
Provides methods for:
- Getting current usage vs limits
- Calculating upgrade recommendations
- Checking limits before actions
Uses the feature provider system for usage counting
and feature_service for limit resolution.
"""
import logging
from dataclasses import dataclass
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.modules.billing.models import MerchantSubscription, SubscriptionTier
from app.modules.billing.services.feature_aggregator import feature_aggregator
from app.modules.tenancy.models import StoreUser
logger = logging.getLogger(__name__)
@dataclass
class UsageMetricData:
"""Usage metric data."""
name: str
current: int
limit: int | None
percentage: float
is_unlimited: bool
is_at_limit: bool
is_approaching_limit: bool
@dataclass
class TierInfoData:
"""Tier information."""
code: str
name: str
price_monthly_cents: int
is_highest_tier: bool
@dataclass
class UpgradeTierData:
"""Upgrade tier information."""
code: str
name: str
price_monthly_cents: int
price_increase_cents: int
benefits: list[str]
@dataclass
class UsageData:
"""Full usage data."""
tier: TierInfoData
usage: list[UsageMetricData]
has_limits_approaching: bool
has_limits_reached: bool
upgrade_available: bool
upgrade_tier: UpgradeTierData | None
upgrade_reasons: list[str]
@dataclass
class LimitCheckData:
"""Limit check result."""
limit_type: str
can_proceed: bool
current: int
limit: int | None
percentage: float
message: str | None
upgrade_tier_code: str | None
upgrade_tier_name: str | None
class UsageService:
"""Service for usage and limits management."""
def _resolve_store_to_subscription(
self, db: Session, store_id: int
) -> MerchantSubscription | None:
"""Resolve store_id to MerchantSubscription."""
from app.modules.billing.services.subscription_service import subscription_service
return subscription_service.get_subscription_for_store(db, store_id)
def get_store_usage(self, db: Session, store_id: int) -> UsageData:
"""
Get comprehensive usage data for a store.
Returns current usage, limits, and upgrade recommendations.
"""
subscription = self._resolve_store_to_subscription(db, store_id)
# Get current tier
tier = subscription.tier if subscription else None
# Calculate usage metrics
usage_metrics = self._calculate_usage_metrics(db, store_id, subscription)
# Check for approaching/reached limits
has_limits_approaching = any(m.is_approaching_limit for m in usage_metrics)
has_limits_reached = any(m.is_at_limit for m in usage_metrics)
# Get upgrade info
next_tier = self._get_next_tier(db, tier)
is_highest_tier = next_tier is None
# Build upgrade info
upgrade_tier_info = None
upgrade_reasons = []
if next_tier:
upgrade_tier_info = self._build_upgrade_tier_info(tier, next_tier)
upgrade_reasons = self._build_upgrade_reasons(
usage_metrics, has_limits_reached, has_limits_approaching
)
tier_code = tier.code if tier else "unknown"
tier_name = tier.name if tier else "Unknown"
tier_price = tier.price_monthly_cents if tier else 0
return UsageData(
tier=TierInfoData(
code=tier_code,
name=tier_name,
price_monthly_cents=tier_price,
is_highest_tier=is_highest_tier,
),
usage=usage_metrics,
has_limits_approaching=has_limits_approaching,
has_limits_reached=has_limits_reached,
upgrade_available=not is_highest_tier,
upgrade_tier=upgrade_tier_info,
upgrade_reasons=upgrade_reasons,
)
def check_limit(
self, db: Session, store_id: int, limit_type: str
) -> LimitCheckData:
"""
Check a specific limit before performing an action.
Args:
db: Database session
store_id: Store ID
limit_type: Feature code (e.g., "orders_per_month", "products_limit", "team_members")
"""
from app.modules.billing.services.feature_service import feature_service
# Map legacy limit_type names to feature codes
feature_code_map = {
"orders": "orders_per_month",
"products": "products_limit",
"team_members": "team_members",
}
feature_code = feature_code_map.get(limit_type, limit_type)
can_proceed, message = feature_service.check_resource_limit(
db, feature_code, store_id=store_id
)
# Get current usage for response
current = 0
limit = None
if feature_code == "products_limit":
current = self._get_product_count(db, store_id)
elif feature_code == "team_members":
current = self._get_team_member_count(db, store_id)
# Get effective limit
subscription = self._resolve_store_to_subscription(db, store_id)
if subscription and subscription.tier:
limit = subscription.tier.get_limit_for_feature(feature_code)
is_unlimited = limit is None
percentage = 0 if is_unlimited else (current / limit * 100 if limit and limit > 0 else 100)
# Get upgrade info if at limit
upgrade_tier_code = None
upgrade_tier_name = None
if not can_proceed and subscription and subscription.tier:
next_tier = self._get_next_tier(db, subscription.tier)
if next_tier:
upgrade_tier_code = next_tier.code
upgrade_tier_name = next_tier.name
return LimitCheckData(
limit_type=limit_type,
can_proceed=can_proceed,
current=current,
limit=None if is_unlimited else limit,
percentage=percentage,
message=message,
upgrade_tier_code=upgrade_tier_code,
upgrade_tier_name=upgrade_tier_name,
)
# =========================================================================
# Private Helper Methods
# =========================================================================
def _get_product_count(self, db: Session, store_id: int) -> int:
"""Get product count for store via feature aggregator."""
usage = feature_aggregator.get_store_usage(db, store_id)
products = usage.get("products_limit")
return products.current_count if products else 0
def _get_team_member_count(self, db: Session, store_id: int) -> int:
"""Get active team member count for store."""
return (
db.query(func.count(StoreUser.id))
.filter(StoreUser.store_id == store_id, StoreUser.is_active == True) # noqa: E712
.scalar()
or 0
)
def _calculate_usage_metrics(
self, db: Session, store_id: int, subscription: MerchantSubscription | None
) -> list[UsageMetricData]:
"""Calculate all usage metrics for a store using TierFeatureLimit."""
metrics = []
tier = subscription.tier if subscription else None
# Define the quantitative features to track
feature_configs = [
("orders_per_month", "orders", lambda: self._get_orders_this_period(db, store_id, subscription)),
("products_limit", "products", lambda: self._get_product_count(db, store_id)),
("team_members", "team_members", lambda: self._get_team_member_count(db, store_id)),
]
for feature_code, display_name, count_fn in feature_configs:
current = count_fn()
limit = tier.get_limit_for_feature(feature_code) if tier else 0
is_unlimited = limit is None
percentage = (
0
if is_unlimited
else (current / limit * 100 if limit and limit > 0 else 100)
)
metrics.append(
UsageMetricData(
name=display_name,
current=current,
limit=None if is_unlimited else limit,
percentage=percentage,
is_unlimited=is_unlimited,
is_at_limit=not is_unlimited and limit is not None and current >= limit,
is_approaching_limit=not is_unlimited and percentage >= 80,
)
)
return metrics
def _get_orders_this_period(
self, db: Session, store_id: int, subscription: MerchantSubscription | None
) -> int:
"""Get order count for the current billing period via feature aggregator."""
usage = feature_aggregator.get_store_usage(db, store_id)
orders = usage.get("orders_per_month")
return orders.current_count if orders else 0
def _get_next_tier(
self, db: Session, current_tier: SubscriptionTier | None
) -> SubscriptionTier | None:
"""Get next tier for upgrade."""
current_tier_order = current_tier.display_order if current_tier else 0
return (
db.query(SubscriptionTier)
.filter(
SubscriptionTier.is_active == True, # noqa: E712
SubscriptionTier.display_order > current_tier_order,
)
.order_by(SubscriptionTier.display_order)
.first()
)
def _build_upgrade_tier_info(
self, current_tier: SubscriptionTier | None, next_tier: SubscriptionTier
) -> UpgradeTierData:
"""Build upgrade tier information with benefits."""
benefits = []
current_features = current_tier.get_feature_codes() if current_tier else set()
next_features = next_tier.get_feature_codes()
new_features = next_features - current_features
# Numeric limit improvements
limit_features = [
("orders_per_month", "orders/month"),
("products_limit", "products"),
("team_members", "team members"),
]
for feature_code, label in limit_features:
next_limit = next_tier.get_limit_for_feature(feature_code)
current_limit = current_tier.get_limit_for_feature(feature_code) if current_tier else 0
if next_limit is None and (current_limit is not None and current_limit != 0):
benefits.append(f"Unlimited {label}")
elif next_limit is not None and (current_limit is None or next_limit > (current_limit or 0)):
benefits.append(f"{next_limit:,} {label}")
# Binary feature benefits
feature_names = {
"analytics_dashboard": "Advanced Analytics",
"api_access": "API Access",
"automation_rules": "Automation Rules",
"team_roles": "Team Roles & Permissions",
"custom_domain": "Custom Domain",
"webhooks": "Webhooks",
"accounting_export": "Accounting Export",
}
for feature in list(new_features)[:3]:
if feature in feature_names:
benefits.append(feature_names[feature])
current_price = current_tier.price_monthly_cents if current_tier else 0
return UpgradeTierData(
code=next_tier.code,
name=next_tier.name,
price_monthly_cents=next_tier.price_monthly_cents,
price_increase_cents=next_tier.price_monthly_cents - current_price,
benefits=benefits,
)
def _build_upgrade_reasons(
self,
usage_metrics: list[UsageMetricData],
has_limits_reached: bool,
has_limits_approaching: bool,
) -> list[str]:
"""Build upgrade reasons based on usage."""
reasons = []
if has_limits_reached:
for m in usage_metrics:
if m.is_at_limit:
reasons.append(f"You've reached your {m.name.replace('_', ' ')} limit")
elif has_limits_approaching:
for m in usage_metrics:
if m.is_approaching_limit:
reasons.append(
f"You're approaching your {m.name.replace('_', ' ')} limit ({int(m.percentage)}%)"
)
return reasons
# Singleton instance
usage_service = UsageService()
__all__ = [
"usage_service",
"UsageService",
"UsageData",
"UsageMetricData",
"TierInfoData",
"UpgradeTierData",
"LimitCheckData",
]