Samir Boulahtit
6232bb47f6
Phase 1 OMS implementation: Invoicing: - Add Invoice and VendorInvoiceSettings database models - Full EU VAT support (27 countries, OSS, B2B reverse charge) - Invoice PDF generation with WeasyPrint + Jinja2 templates - Vendor invoice API endpoints for settings, creation, PDF download Subscription Tiers: - Add VendorSubscription model with 4 tiers (Essential/Professional/Business/Enterprise) - Tier limit enforcement for orders, products, team members - Feature gating based on subscription tier - Automatic trial subscription creation for new vendors - Integrate limit checks into order creation (direct and Letzshop sync) Marketing: - Update pricing documentation with 4-tier structure - Revise back-office positioning strategy - Update homepage with Veeqo-inspired Letzshop-focused messaging 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
355 lines
11 KiB
Python
355 lines
11 KiB
Python
# models/database/subscription.py
|
|
"""
|
|
Subscription database models for tier-based access control.
|
|
|
|
Provides models for:
|
|
- SubscriptionTier: Tier definitions with limits and features
|
|
- VendorSubscription: Per-vendor subscription tracking
|
|
|
|
Tier Structure:
|
|
- Essential (€49/mo): 100 orders/mo, 200 products, 1 user, LU invoicing
|
|
- Professional (€99/mo): 500 orders/mo, unlimited products, 3 users, EU VAT
|
|
- Business (€199/mo): 2000 orders/mo, unlimited products, 10 users, analytics, API
|
|
- Enterprise (€399+/mo): Unlimited, white-label, custom integrations
|
|
"""
|
|
|
|
import enum
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import (
|
|
Boolean,
|
|
Column,
|
|
DateTime,
|
|
ForeignKey,
|
|
Index,
|
|
Integer,
|
|
Numeric,
|
|
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
|
|
|
|
|
|
# Tier limit definitions (hardcoded for now, could be moved to DB)
|
|
TIER_LIMITS = {
|
|
TierCode.ESSENTIAL: {
|
|
"name": "Essential",
|
|
"price_monthly_cents": 4900, # €49
|
|
"price_annual_cents": 49000, # €490 (2 months free)
|
|
"orders_per_month": 100,
|
|
"products_limit": 200,
|
|
"team_members": 1,
|
|
"order_history_months": 6,
|
|
"features": [
|
|
"letzshop_sync",
|
|
"inventory_basic",
|
|
"invoice_lu",
|
|
"customer_view",
|
|
],
|
|
},
|
|
TierCode.PROFESSIONAL: {
|
|
"name": "Professional",
|
|
"price_monthly_cents": 9900, # €99
|
|
"price_annual_cents": 99000, # €990
|
|
"orders_per_month": 500,
|
|
"products_limit": None, # Unlimited
|
|
"team_members": 3,
|
|
"order_history_months": 24,
|
|
"features": [
|
|
"letzshop_sync",
|
|
"inventory_locations",
|
|
"inventory_purchase_orders",
|
|
"invoice_lu",
|
|
"invoice_eu_vat",
|
|
"customer_view",
|
|
"customer_export",
|
|
],
|
|
},
|
|
TierCode.BUSINESS: {
|
|
"name": "Business",
|
|
"price_monthly_cents": 19900, # €199
|
|
"price_annual_cents": 199000, # €1990
|
|
"orders_per_month": 2000,
|
|
"products_limit": None, # Unlimited
|
|
"team_members": 10,
|
|
"order_history_months": None, # Unlimited
|
|
"features": [
|
|
"letzshop_sync",
|
|
"inventory_locations",
|
|
"inventory_purchase_orders",
|
|
"invoice_lu",
|
|
"invoice_eu_vat",
|
|
"invoice_bulk",
|
|
"customer_view",
|
|
"customer_export",
|
|
"analytics_dashboard",
|
|
"accounting_export",
|
|
"api_access",
|
|
"automation_rules",
|
|
"team_roles",
|
|
],
|
|
},
|
|
TierCode.ENTERPRISE: {
|
|
"name": "Enterprise",
|
|
"price_monthly_cents": 39900, # €399 starting
|
|
"price_annual_cents": None, # Custom
|
|
"orders_per_month": None, # Unlimited
|
|
"products_limit": None, # Unlimited
|
|
"team_members": None, # Unlimited
|
|
"order_history_months": None, # Unlimited
|
|
"features": [
|
|
"letzshop_sync",
|
|
"inventory_locations",
|
|
"inventory_purchase_orders",
|
|
"invoice_lu",
|
|
"invoice_eu_vat",
|
|
"invoice_bulk",
|
|
"customer_view",
|
|
"customer_export",
|
|
"analytics_dashboard",
|
|
"accounting_export",
|
|
"api_access",
|
|
"automation_rules",
|
|
"team_roles",
|
|
"white_label",
|
|
"multi_vendor",
|
|
"custom_integrations",
|
|
"sla_guarantee",
|
|
"dedicated_support",
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
class VendorSubscription(Base, TimestampMixin):
|
|
"""
|
|
Per-vendor subscription tracking.
|
|
|
|
Tracks the vendor's subscription tier, billing period,
|
|
and usage counters for limit enforcement.
|
|
"""
|
|
|
|
__tablename__ = "vendor_subscriptions"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
vendor_id = Column(
|
|
Integer, ForeignKey("vendors.id"), unique=True, nullable=False, index=True
|
|
)
|
|
|
|
# Tier
|
|
tier = Column(
|
|
String(20), default=TierCode.ESSENTIAL.value, nullable=False, index=True
|
|
)
|
|
|
|
# Status
|
|
status = Column(
|
|
String(20), default=SubscriptionStatus.TRIAL.value, nullable=False, index=True
|
|
)
|
|
|
|
# Billing period
|
|
period_start = Column(DateTime(timezone=True), nullable=False)
|
|
period_end = Column(DateTime(timezone=True), nullable=False)
|
|
is_annual = Column(Boolean, default=False, nullable=False)
|
|
|
|
# Trial info
|
|
trial_ends_at = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Usage counters (reset each billing period)
|
|
orders_this_period = Column(Integer, default=0, nullable=False)
|
|
orders_limit_reached_at = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Overrides (for custom enterprise deals)
|
|
custom_orders_limit = Column(Integer, nullable=True) # Override tier limit
|
|
custom_products_limit = Column(Integer, nullable=True)
|
|
custom_team_limit = Column(Integer, nullable=True)
|
|
|
|
# Payment info (for future Stripe integration)
|
|
stripe_customer_id = Column(String(100), nullable=True, index=True)
|
|
stripe_subscription_id = Column(String(100), nullable=True, index=True)
|
|
|
|
# Cancellation
|
|
cancelled_at = Column(DateTime(timezone=True), nullable=True)
|
|
cancellation_reason = Column(Text, nullable=True)
|
|
|
|
# Relationships
|
|
vendor = relationship("Vendor", back_populates="subscription")
|
|
|
|
__table_args__ = (
|
|
Index("idx_subscription_vendor_status", "vendor_id", "status"),
|
|
Index("idx_subscription_period", "period_start", "period_end"),
|
|
)
|
|
|
|
def __repr__(self):
|
|
return f"<VendorSubscription(vendor_id={self.vendor_id}, tier='{self.tier}', status='{self.status}')>"
|
|
|
|
# =========================================================================
|
|
# Tier Limit Properties
|
|
# =========================================================================
|
|
|
|
@property
|
|
def tier_limits(self) -> dict:
|
|
"""Get the limit definitions for current tier."""
|
|
return TIER_LIMITS.get(TierCode(self.tier), TIER_LIMITS[TierCode.ESSENTIAL])
|
|
|
|
@property
|
|
def orders_limit(self) -> int | None:
|
|
"""Get effective orders limit (custom or tier default)."""
|
|
if self.custom_orders_limit is not None:
|
|
return self.custom_orders_limit
|
|
return self.tier_limits.get("orders_per_month")
|
|
|
|
@property
|
|
def products_limit(self) -> int | None:
|
|
"""Get effective products limit (custom or tier default)."""
|
|
if self.custom_products_limit is not None:
|
|
return self.custom_products_limit
|
|
return self.tier_limits.get("products_limit")
|
|
|
|
@property
|
|
def team_members_limit(self) -> int | None:
|
|
"""Get effective team members limit (custom or tier default)."""
|
|
if self.custom_team_limit is not None:
|
|
return self.custom_team_limit
|
|
return self.tier_limits.get("team_members")
|
|
|
|
@property
|
|
def features(self) -> list[str]:
|
|
"""Get list of enabled features for current tier."""
|
|
return self.tier_limits.get("features", [])
|
|
|
|
# =========================================================================
|
|
# Status Checks
|
|
# =========================================================================
|
|
|
|
@property
|
|
def is_active(self) -> bool:
|
|
"""Check if subscription allows access."""
|
|
return self.status in [
|
|
SubscriptionStatus.TRIAL.value,
|
|
SubscriptionStatus.ACTIVE.value,
|
|
SubscriptionStatus.PAST_DUE.value, # Grace period
|
|
SubscriptionStatus.CANCELLED.value, # Until period end
|
|
]
|
|
|
|
@property
|
|
def is_trial(self) -> bool:
|
|
"""Check if currently in trial."""
|
|
return self.status == SubscriptionStatus.TRIAL.value
|
|
|
|
@property
|
|
def trial_days_remaining(self) -> int | None:
|
|
"""Get remaining trial days."""
|
|
if not self.is_trial or not self.trial_ends_at:
|
|
return None
|
|
remaining = (self.trial_ends_at - datetime.now(UTC)).days
|
|
return max(0, remaining)
|
|
|
|
# =========================================================================
|
|
# Limit Checks
|
|
# =========================================================================
|
|
|
|
def can_create_order(self) -> tuple[bool, str | None]:
|
|
"""
|
|
Check if vendor can create/import another order.
|
|
|
|
Returns: (can_create, error_message)
|
|
"""
|
|
if not self.is_active:
|
|
return False, "Subscription is not active"
|
|
|
|
limit = self.orders_limit
|
|
if limit is None: # Unlimited
|
|
return True, None
|
|
|
|
if self.orders_this_period >= limit:
|
|
return False, f"Monthly order limit reached ({limit} orders). Upgrade to continue."
|
|
|
|
return True, None
|
|
|
|
def can_add_product(self, current_count: int) -> tuple[bool, str | None]:
|
|
"""
|
|
Check if vendor can add another product.
|
|
|
|
Args:
|
|
current_count: Current number of products
|
|
|
|
Returns: (can_add, error_message)
|
|
"""
|
|
if not self.is_active:
|
|
return False, "Subscription is not active"
|
|
|
|
limit = self.products_limit
|
|
if limit is None: # Unlimited
|
|
return True, None
|
|
|
|
if current_count >= limit:
|
|
return False, f"Product limit reached ({limit} products). Upgrade to add more."
|
|
|
|
return True, None
|
|
|
|
def can_add_team_member(self, current_count: int) -> tuple[bool, str | None]:
|
|
"""
|
|
Check if vendor can add another team member.
|
|
|
|
Args:
|
|
current_count: Current number of team members
|
|
|
|
Returns: (can_add, error_message)
|
|
"""
|
|
if not self.is_active:
|
|
return False, "Subscription is not active"
|
|
|
|
limit = self.team_members_limit
|
|
if limit is None: # Unlimited
|
|
return True, None
|
|
|
|
if current_count >= limit:
|
|
return False, f"Team member limit reached ({limit} members). Upgrade to add more."
|
|
|
|
return True, None
|
|
|
|
def has_feature(self, feature: str) -> bool:
|
|
"""Check if a feature is enabled for current tier."""
|
|
return feature in self.features
|
|
|
|
# =========================================================================
|
|
# Usage Tracking
|
|
# =========================================================================
|
|
|
|
def increment_order_count(self) -> None:
|
|
"""Increment the order counter for this period."""
|
|
self.orders_this_period += 1
|
|
|
|
# Track when limit was first reached
|
|
limit = self.orders_limit
|
|
if limit and self.orders_this_period >= limit and not self.orders_limit_reached_at:
|
|
self.orders_limit_reached_at = datetime.now(UTC)
|
|
|
|
def reset_period_counters(self) -> None:
|
|
"""Reset counters for new billing period."""
|
|
self.orders_this_period = 0
|
|
self.orders_limit_reached_at = None
|