Clean up accumulated backward-compat shims, deprecated wrappers, unused aliases, and legacy code across the codebase. Since the platform is not live yet, this establishes a clean baseline. Changes: - Delete deprecated middleware/context.py (RequestContext, get_request_context) - Remove unused factory get_store_email_settings_service() - Remove deprecated pagination_full macro, /admin/platform-homepage route - Remove ConversationResponse, InvoiceSettings* unprefixed aliases - Simplify celery_config.py (remove empty LEGACY_TASK_MODULES) - Standardize billing exceptions: *Error aliases → *Exception names - Consolidate duplicate TierNotFoundError/FeatureNotFoundError classes - Remove deprecated is_admin_request() from Store/PlatformContextManager - Remove is_platform_default field, MediaUploadResponse legacy flat fields - Remove MediaItemResponse.url alias, update JS to use file_url - Update all affected tests and documentation Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
270 lines
8.4 KiB
Python
270 lines
8.4 KiB
Python
# app/modules/billing/exceptions.py
|
|
"""
|
|
Billing module exceptions.
|
|
|
|
This module provides exception classes for billing operations including:
|
|
- Subscription management
|
|
- Payment processing (Stripe)
|
|
- Feature management
|
|
- Tier management
|
|
"""
|
|
|
|
from app.exceptions.base import (
|
|
BusinessLogicException,
|
|
ResourceNotFoundException,
|
|
ServiceUnavailableException,
|
|
ValidationException,
|
|
)
|
|
|
|
__all__ = [
|
|
# Base billing exception
|
|
"BillingException",
|
|
# Subscription exceptions
|
|
"SubscriptionNotFoundException",
|
|
"NoActiveSubscriptionException",
|
|
"SubscriptionNotCancelledException",
|
|
"SubscriptionAlreadyCancelledException",
|
|
# Tier exceptions
|
|
"TierNotFoundException",
|
|
"TierLimitExceededException",
|
|
# Payment exceptions
|
|
"PaymentSystemNotConfiguredException",
|
|
"StripeNotConfiguredException",
|
|
"StripePriceNotConfiguredException",
|
|
"PaymentFailedException",
|
|
# Webhook exceptions
|
|
"InvalidWebhookSignatureException",
|
|
"WebhookMissingSignatureException",
|
|
"WebhookVerificationException",
|
|
# Feature exceptions
|
|
"FeatureNotFoundException",
|
|
"FeatureNotAvailableException",
|
|
"InvalidFeatureCodesError",
|
|
]
|
|
|
|
|
|
# =============================================================================
|
|
# Base Billing Exception
|
|
# =============================================================================
|
|
|
|
|
|
class BillingException(BusinessLogicException):
|
|
"""Base exception for billing module errors."""
|
|
|
|
def __init__(self, message: str, error_code: str = "BILLING_ERROR", details: dict | None = None):
|
|
super().__init__(message=message, error_code=error_code, details=details)
|
|
|
|
|
|
# =============================================================================
|
|
# Subscription Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class SubscriptionNotFoundException(ResourceNotFoundException):
|
|
"""Raised when a subscription is not found."""
|
|
|
|
def __init__(self, store_id: int):
|
|
super().__init__(
|
|
resource_type="Subscription",
|
|
identifier=str(store_id),
|
|
error_code="SUBSCRIPTION_NOT_FOUND",
|
|
)
|
|
|
|
|
|
class NoActiveSubscriptionException(BusinessLogicException):
|
|
"""Raised when no active subscription exists for an operation that requires one."""
|
|
|
|
def __init__(self, message: str = "No active subscription found"):
|
|
super().__init__(
|
|
message=message,
|
|
error_code="NO_ACTIVE_SUBSCRIPTION",
|
|
)
|
|
|
|
|
|
class SubscriptionNotCancelledException(BusinessLogicException):
|
|
"""Raised when trying to reactivate a subscription that is not cancelled."""
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
message="Subscription is not cancelled",
|
|
error_code="SUBSCRIPTION_NOT_CANCELLED",
|
|
)
|
|
|
|
|
|
class SubscriptionAlreadyCancelledException(BusinessLogicException):
|
|
"""Raised when trying to cancel an already cancelled subscription."""
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
message="Subscription is already cancelled",
|
|
error_code="SUBSCRIPTION_ALREADY_CANCELLED",
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Tier Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class TierNotFoundException(ResourceNotFoundException):
|
|
"""Raised when a subscription tier is not found."""
|
|
|
|
def __init__(self, tier_code: str):
|
|
super().__init__(
|
|
resource_type="SubscriptionTier",
|
|
identifier=tier_code,
|
|
message=f"Subscription tier '{tier_code}' not found",
|
|
error_code="TIER_NOT_FOUND",
|
|
)
|
|
self.tier_code = tier_code
|
|
|
|
|
|
class TierLimitExceededException(BillingException):
|
|
"""Raised when a tier limit is exceeded."""
|
|
|
|
def __init__(self, message: str, limit_type: str, current: int, limit: int):
|
|
super().__init__(
|
|
message=message,
|
|
error_code="TIER_LIMIT_EXCEEDED",
|
|
details={
|
|
"limit_type": limit_type,
|
|
"current": current,
|
|
"limit": limit,
|
|
},
|
|
)
|
|
self.limit_type = limit_type
|
|
self.current = current
|
|
self.limit = limit
|
|
|
|
|
|
# =============================================================================
|
|
# Payment Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class PaymentSystemNotConfiguredException(ServiceUnavailableException):
|
|
"""Raised when the payment system (Stripe) is not configured."""
|
|
|
|
def __init__(self):
|
|
super().__init__(message="Payment system not configured")
|
|
|
|
|
|
class StripeNotConfiguredException(BillingException):
|
|
"""Raised when Stripe is not configured."""
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
message="Stripe is not configured",
|
|
error_code="STRIPE_NOT_CONFIGURED",
|
|
)
|
|
|
|
|
|
class StripePriceNotConfiguredException(BusinessLogicException):
|
|
"""Raised when Stripe price is not configured for a tier."""
|
|
|
|
def __init__(self, tier_code: str):
|
|
super().__init__(
|
|
message=f"Stripe price not configured for tier '{tier_code}'",
|
|
error_code="STRIPE_PRICE_NOT_CONFIGURED",
|
|
details={"tier_code": tier_code},
|
|
)
|
|
self.tier_code = tier_code
|
|
|
|
|
|
class PaymentFailedException(BillingException):
|
|
"""Raised when a payment fails."""
|
|
|
|
def __init__(self, message: str, stripe_error: str | None = None):
|
|
details = {}
|
|
if stripe_error:
|
|
details["stripe_error"] = stripe_error
|
|
super().__init__(
|
|
message=message,
|
|
error_code="PAYMENT_FAILED",
|
|
details=details if details else None,
|
|
)
|
|
self.stripe_error = stripe_error
|
|
|
|
|
|
# =============================================================================
|
|
# Webhook Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class InvalidWebhookSignatureException(BusinessLogicException):
|
|
"""Raised when Stripe webhook signature verification fails."""
|
|
|
|
def __init__(self, message: str = "Invalid webhook signature"):
|
|
super().__init__(
|
|
message=message,
|
|
error_code="INVALID_WEBHOOK_SIGNATURE",
|
|
)
|
|
|
|
|
|
class WebhookMissingSignatureException(BusinessLogicException):
|
|
"""Raised when Stripe webhook is missing the signature header."""
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
message="Missing Stripe-Signature header",
|
|
error_code="WEBHOOK_MISSING_SIGNATURE",
|
|
)
|
|
|
|
|
|
class WebhookVerificationException(BillingException):
|
|
"""Raised when webhook signature verification fails."""
|
|
|
|
def __init__(self, message: str = "Invalid webhook signature"):
|
|
super().__init__(
|
|
message=message,
|
|
error_code="WEBHOOK_VERIFICATION_FAILED",
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Feature Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class FeatureNotFoundException(ResourceNotFoundException):
|
|
"""Raised when a feature is not found."""
|
|
|
|
def __init__(self, feature_code: str):
|
|
super().__init__(
|
|
resource_type="Feature",
|
|
identifier=feature_code,
|
|
message=f"Feature '{feature_code}' not found",
|
|
)
|
|
self.feature_code = feature_code
|
|
|
|
|
|
class FeatureNotAvailableException(BillingException):
|
|
"""Raised when a feature is not available in current tier."""
|
|
|
|
def __init__(self, feature: str, current_tier: str, required_tier: str):
|
|
message = f"Feature '{feature}' requires {required_tier} tier (current: {current_tier})"
|
|
super().__init__(
|
|
message=message,
|
|
error_code="FEATURE_NOT_AVAILABLE",
|
|
details={
|
|
"feature": feature,
|
|
"current_tier": current_tier,
|
|
"required_tier": required_tier,
|
|
},
|
|
)
|
|
self.feature = feature
|
|
self.current_tier = current_tier
|
|
self.required_tier = required_tier
|
|
|
|
|
|
class InvalidFeatureCodesError(ValidationException):
|
|
"""Invalid feature codes provided."""
|
|
|
|
def __init__(self, invalid_codes: set[str]):
|
|
codes_str = ", ".join(sorted(invalid_codes))
|
|
super().__init__(
|
|
message=f"Invalid feature codes: {codes_str}",
|
|
details={"invalid_codes": list(invalid_codes)},
|
|
)
|
|
self.invalid_codes = invalid_codes
|