Samir Boulahtit
86e85a98b8
Some checks failed
Enforce MOD-025/MOD-026 rules: zero top-level cross-module model imports remain in any service file. All 66 files migrated using deferred import patterns (method-body, _get_model() helpers, instance-cached self._Model) and new cross-module service methods in tenancy. Documentation updated with Pattern 6 (deferred imports), migration plan marked complete, and violations status reflects 84→0 service-layer violations. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
373 lines
12 KiB
Python
373 lines
12 KiB
Python
# 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.orm import Session
|
|
|
|
from app.modules.billing.models import MerchantSubscription, SubscriptionTier
|
|
from app.modules.billing.services.feature_aggregator import feature_aggregator
|
|
|
|
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."""
|
|
from app.modules.tenancy.services.team_service import team_service
|
|
|
|
return team_service.get_active_team_member_count(db, store_id)
|
|
|
|
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",
|
|
]
|