Samir Boulahtit
3c7e4458af
Some checks failed
Add name_translations JSON column to SubscriptionTier for multi-language tier names. Pre-resolve tier names and build dynamic feature lists from module providers in route handlers. Fix Jinja2 macro scoping by importing pricing partial with context. Backfill content_translations for all 43 content pages across 4 platforms (en/fr/de/lb). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
363 lines
12 KiB
Python
363 lines
12 KiB
Python
# app/modules/billing/models/subscription.py
|
|
"""
|
|
Subscription database models for tier-based access control.
|
|
|
|
Provides models for:
|
|
- SubscriptionTier: Database-driven tier definitions with Stripe integration
|
|
- AddOnProduct: Purchasable add-ons (domains, SSL, email packages)
|
|
- StoreAddOn: Add-ons purchased by each store
|
|
- StripeWebhookEvent: Idempotency tracking for webhook processing
|
|
- BillingHistory: Invoice and payment history
|
|
|
|
Merchant-level subscriptions are in merchant_subscription.py.
|
|
Feature limits per tier are in tier_feature_limit.py.
|
|
"""
|
|
|
|
import enum
|
|
|
|
from sqlalchemy import (
|
|
Boolean,
|
|
Column,
|
|
DateTime,
|
|
ForeignKey,
|
|
Index,
|
|
Integer,
|
|
String,
|
|
Text,
|
|
)
|
|
from sqlalchemy.dialects.sqlite import JSON
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from app.core.database import Base
|
|
from models.database.base import TimestampMixin
|
|
|
|
|
|
class TierCode(str, enum.Enum):
|
|
"""Subscription tier codes."""
|
|
|
|
ESSENTIAL = "essential"
|
|
PROFESSIONAL = "professional"
|
|
BUSINESS = "business"
|
|
ENTERPRISE = "enterprise"
|
|
|
|
|
|
class SubscriptionStatus(str, enum.Enum):
|
|
"""Subscription status."""
|
|
|
|
TRIAL = "trial" # Free trial period
|
|
ACTIVE = "active" # Paid and active
|
|
PAST_DUE = "past_due" # Payment failed, grace period
|
|
CANCELLED = "cancelled" # Cancelled, access until period end
|
|
EXPIRED = "expired" # No longer active
|
|
|
|
|
|
class AddOnCategory(str, enum.Enum):
|
|
"""Add-on product categories."""
|
|
|
|
DOMAIN = "domain"
|
|
SSL = "ssl"
|
|
EMAIL = "email"
|
|
STORAGE = "storage"
|
|
|
|
|
|
class BillingPeriod(str, enum.Enum):
|
|
"""Billing period for add-ons."""
|
|
|
|
MONTHLY = "monthly"
|
|
ANNUAL = "annual"
|
|
ONE_TIME = "one_time"
|
|
|
|
|
|
# ============================================================================
|
|
# SubscriptionTier - Database-driven tier definitions
|
|
# ============================================================================
|
|
|
|
|
|
class SubscriptionTier(Base, TimestampMixin):
|
|
"""
|
|
Database-driven tier definitions with Stripe integration.
|
|
|
|
Feature limits are now stored in the TierFeatureLimit table
|
|
(one row per feature per tier) instead of hardcoded columns.
|
|
|
|
Can be:
|
|
- Global tier (platform_id=NULL): Available to all platforms
|
|
- Platform-specific tier (platform_id set): Only for that platform
|
|
"""
|
|
|
|
__tablename__ = "subscription_tiers"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
|
|
# Platform association (NULL = global tier available to all platforms)
|
|
platform_id = Column(
|
|
Integer,
|
|
ForeignKey("platforms.id", ondelete="CASCADE"),
|
|
nullable=True,
|
|
index=True,
|
|
comment="Platform this tier belongs to (NULL = global tier)",
|
|
)
|
|
|
|
code = Column(String(30), nullable=False, index=True)
|
|
name = Column(String(100), nullable=False)
|
|
name_translations = Column(
|
|
JSON,
|
|
nullable=True,
|
|
default=None,
|
|
comment="Language-keyed name dict for multi-language support",
|
|
)
|
|
description = Column(Text, nullable=True)
|
|
|
|
# Pricing (in cents for precision)
|
|
price_monthly_cents = Column(Integer, nullable=False)
|
|
price_annual_cents = Column(Integer, nullable=True) # Null for enterprise/custom
|
|
|
|
# Stripe Product/Price IDs
|
|
stripe_product_id = Column(String(100), nullable=True)
|
|
stripe_price_monthly_id = Column(String(100), nullable=True)
|
|
stripe_price_annual_id = Column(String(100), nullable=True)
|
|
|
|
# Display and visibility
|
|
display_order = Column(Integer, default=0)
|
|
is_active = Column(Boolean, default=True, nullable=False)
|
|
is_public = Column(Boolean, default=True, nullable=False) # False for enterprise
|
|
|
|
# Relationship to Platform
|
|
platform = relationship(
|
|
"Platform",
|
|
back_populates="subscription_tiers",
|
|
foreign_keys=[platform_id],
|
|
)
|
|
|
|
# Feature limits (one row per feature)
|
|
feature_limits = relationship(
|
|
"TierFeatureLimit",
|
|
back_populates="tier",
|
|
cascade="all, delete-orphan",
|
|
lazy="selectin",
|
|
)
|
|
|
|
__table_args__ = (
|
|
Index("idx_tier_platform_active", "platform_id", "is_active"),
|
|
)
|
|
|
|
def __repr__(self):
|
|
platform_info = f", platform_id={self.platform_id}" if self.platform_id else ""
|
|
return f"<SubscriptionTier(code='{self.code}', name='{self.name}'{platform_info})>"
|
|
|
|
def get_feature_codes(self) -> set[str]:
|
|
"""Get all feature codes enabled for this tier."""
|
|
return {fl.feature_code for fl in (self.feature_limits or [])}
|
|
|
|
def get_limit_for_feature(self, feature_code: str) -> int | None:
|
|
"""Get the limit value for a specific feature (None = unlimited)."""
|
|
for fl in (self.feature_limits or []):
|
|
if fl.feature_code == feature_code:
|
|
return fl.limit_value
|
|
return None
|
|
|
|
def has_feature(self, feature_code: str) -> bool:
|
|
"""Check if this tier includes a specific feature."""
|
|
return feature_code in self.get_feature_codes()
|
|
|
|
def get_translated_name(self, lang: str, default_lang: str = "fr") -> str:
|
|
"""Get name in the given language, falling back to default_lang then self.name."""
|
|
if self.name_translations:
|
|
return (
|
|
self.name_translations.get(lang)
|
|
or self.name_translations.get(default_lang)
|
|
or self.name
|
|
)
|
|
return self.name
|
|
|
|
|
|
# ============================================================================
|
|
# AddOnProduct - Purchasable add-ons
|
|
# ============================================================================
|
|
|
|
|
|
class AddOnProduct(Base, TimestampMixin):
|
|
"""
|
|
Purchasable add-on products (domains, SSL, email packages).
|
|
|
|
These are separate from subscription tiers and can be added to any tier.
|
|
"""
|
|
|
|
__tablename__ = "addon_products"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
code = Column(String(50), unique=True, nullable=False, index=True)
|
|
name = Column(String(100), nullable=False)
|
|
description = Column(Text, nullable=True)
|
|
category = Column(String(50), nullable=False, index=True)
|
|
|
|
# Pricing
|
|
price_cents = Column(Integer, nullable=False)
|
|
billing_period = Column(
|
|
String(20), default=BillingPeriod.MONTHLY.value, nullable=False
|
|
)
|
|
|
|
# For tiered add-ons (e.g., email_5, email_10)
|
|
quantity_unit = Column(String(50), nullable=True) # emails, GB, etc.
|
|
quantity_value = Column(Integer, nullable=True) # 5, 10, 50, etc.
|
|
|
|
# Stripe
|
|
stripe_product_id = Column(String(100), nullable=True)
|
|
stripe_price_id = Column(String(100), nullable=True)
|
|
|
|
# Display
|
|
display_order = Column(Integer, default=0)
|
|
is_active = Column(Boolean, default=True, nullable=False)
|
|
|
|
def __repr__(self):
|
|
return f"<AddOnProduct(code='{self.code}', name='{self.name}')>"
|
|
|
|
|
|
# ============================================================================
|
|
# StoreAddOn - Add-ons purchased by store
|
|
# ============================================================================
|
|
|
|
|
|
class StoreAddOn(Base, TimestampMixin):
|
|
"""
|
|
Add-ons purchased by a store.
|
|
|
|
Tracks active add-on subscriptions and their billing status.
|
|
"""
|
|
|
|
__tablename__ = "store_addons"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
store_id = Column(Integer, ForeignKey("stores.id"), nullable=False, index=True)
|
|
addon_product_id = Column(
|
|
Integer, ForeignKey("addon_products.id"), nullable=False, index=True
|
|
)
|
|
|
|
# Status
|
|
status = Column(String(20), default="active", nullable=False, index=True)
|
|
|
|
# For domains: store the actual domain name
|
|
domain_name = Column(String(255), nullable=True, index=True)
|
|
|
|
# Quantity (for tiered add-ons like email packages)
|
|
quantity = Column(Integer, default=1, nullable=False)
|
|
|
|
# Stripe billing
|
|
stripe_subscription_item_id = Column(String(100), nullable=True)
|
|
|
|
# Period tracking
|
|
period_start = Column(DateTime(timezone=True), nullable=True)
|
|
period_end = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Cancellation
|
|
cancelled_at = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Relationships
|
|
store = relationship("Store", back_populates="addons")
|
|
addon_product = relationship("AddOnProduct")
|
|
|
|
__table_args__ = (
|
|
Index("idx_store_addon_status", "store_id", "status"),
|
|
Index("idx_store_addon_product", "store_id", "addon_product_id"),
|
|
)
|
|
|
|
def __repr__(self):
|
|
return f"<StoreAddOn(store_id={self.store_id}, addon={self.addon_product_id}, status='{self.status}')>"
|
|
|
|
|
|
# ============================================================================
|
|
# StripeWebhookEvent - Webhook idempotency tracking
|
|
# ============================================================================
|
|
|
|
|
|
class StripeWebhookEvent(Base, TimestampMixin):
|
|
"""
|
|
Log of processed Stripe webhook events for idempotency.
|
|
|
|
Prevents duplicate processing of the same event.
|
|
"""
|
|
|
|
__tablename__ = "stripe_webhook_events"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
event_id = Column(String(100), unique=True, nullable=False, index=True)
|
|
event_type = Column(String(100), nullable=False, index=True)
|
|
|
|
# Processing status
|
|
status = Column(String(20), default="pending", nullable=False, index=True)
|
|
processed_at = Column(DateTime(timezone=True), nullable=True)
|
|
error_message = Column(Text, nullable=True)
|
|
|
|
# Raw event data (encrypted for security)
|
|
payload_encrypted = Column(Text, nullable=True)
|
|
|
|
# Related entities (for quick lookup)
|
|
store_id = Column(Integer, ForeignKey("stores.id"), nullable=True, index=True)
|
|
merchant_subscription_id = Column(
|
|
Integer, ForeignKey("merchant_subscriptions.id"), nullable=True, index=True
|
|
)
|
|
|
|
__table_args__ = (Index("idx_webhook_event_type_status", "event_type", "status"),)
|
|
|
|
def __repr__(self):
|
|
return f"<StripeWebhookEvent(event_id='{self.event_id}', type='{self.event_type}', status='{self.status}')>"
|
|
|
|
|
|
# ============================================================================
|
|
# BillingHistory - Invoice and payment history
|
|
# ============================================================================
|
|
|
|
|
|
class BillingHistory(Base, TimestampMixin):
|
|
"""
|
|
Invoice and payment history for merchants.
|
|
|
|
Stores Stripe invoice data for display and reporting.
|
|
"""
|
|
|
|
__tablename__ = "billing_history"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
store_id = Column(Integer, ForeignKey("stores.id"), nullable=True, index=True)
|
|
|
|
# Merchant association (billing is now merchant-level)
|
|
merchant_id = Column(Integer, ForeignKey("merchants.id"), nullable=True, index=True)
|
|
|
|
# Stripe references
|
|
stripe_invoice_id = Column(String(100), unique=True, nullable=True, index=True)
|
|
stripe_payment_intent_id = Column(String(100), nullable=True)
|
|
|
|
# Invoice details
|
|
invoice_number = Column(String(50), nullable=True)
|
|
invoice_date = Column(DateTime(timezone=True), nullable=False)
|
|
due_date = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Amounts (in cents for precision)
|
|
subtotal_cents = Column(Integer, nullable=False)
|
|
tax_cents = Column(Integer, default=0, nullable=False)
|
|
total_cents = Column(Integer, nullable=False)
|
|
amount_paid_cents = Column(Integer, default=0, nullable=False)
|
|
currency = Column(String(3), default="EUR", nullable=False)
|
|
|
|
# Status
|
|
status = Column(String(20), nullable=False, index=True)
|
|
|
|
# PDF URLs
|
|
invoice_pdf_url = Column(String(500), nullable=True)
|
|
hosted_invoice_url = Column(String(500), nullable=True)
|
|
|
|
# Description and line items
|
|
description = Column(Text, nullable=True)
|
|
line_items = Column(JSON, nullable=True)
|
|
|
|
# Relationships
|
|
store = relationship("Store", back_populates="billing_history")
|
|
|
|
__table_args__ = (
|
|
Index("idx_billing_store_date", "store_id", "invoice_date"),
|
|
Index("idx_billing_status", "store_id", "status"),
|
|
)
|
|
|
|
def __repr__(self):
|
|
return f"<BillingHistory(store_id={self.store_id}, invoice='{self.invoice_number}', status='{self.status}')>"
|