Phase 6 - Database-driven tiers: - Update subscription_service to query database first with legacy fallback - Add get_tier_info() db parameter and _get_tier_from_legacy() method Phase 7 - Platform health integration: - Add get_subscription_capacity() for theoretical vs actual capacity - Include subscription capacity in full health report Phase 8 - Background subscription tasks: - Add reset_period_counters() for billing period resets - Add check_trial_expirations() for trial management - Add sync_stripe_status() for Stripe synchronization - Add cleanup_stale_subscriptions() for maintenance - Add capture_capacity_snapshot() for daily metrics Phase 10 - Capacity planning & forecasting: - Add CapacitySnapshot model for historical tracking - Create capacity_forecast_service with growth trends - Add /subscription-capacity, /trends, /recommendations endpoints - Add /snapshot endpoint for manual captures Also includes billing API enhancements from phase 4: - Add upcoming-invoice, change-tier, addon purchase/cancel endpoints - Add UsageSummary schema for billing page - Enhance billing.js with addon management functions 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
301 lines
7.7 KiB
Python
301 lines
7.7 KiB
Python
# models/schema/billing.py
|
|
"""
|
|
Pydantic schemas for billing and subscription operations.
|
|
|
|
Used for both vendor billing endpoints and admin subscription management.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
|
|
# ============================================================================
|
|
# Subscription Tier Schemas
|
|
# ============================================================================
|
|
|
|
|
|
class SubscriptionTierBase(BaseModel):
|
|
"""Base schema for subscription tier."""
|
|
|
|
code: str = Field(..., min_length=1, max_length=30)
|
|
name: str = Field(..., min_length=1, max_length=100)
|
|
description: str | None = None
|
|
price_monthly_cents: int = Field(..., ge=0)
|
|
price_annual_cents: int | None = Field(None, ge=0)
|
|
orders_per_month: int | None = Field(None, ge=0)
|
|
products_limit: int | None = Field(None, ge=0)
|
|
team_members: int | None = Field(None, ge=0)
|
|
order_history_months: int | None = Field(None, ge=0)
|
|
features: list[str] = Field(default_factory=list)
|
|
stripe_product_id: str | None = None
|
|
stripe_price_monthly_id: str | None = None
|
|
stripe_price_annual_id: str | None = None
|
|
display_order: int = 0
|
|
is_active: bool = True
|
|
is_public: bool = True
|
|
|
|
|
|
class SubscriptionTierCreate(SubscriptionTierBase):
|
|
"""Schema for creating a subscription tier."""
|
|
|
|
pass
|
|
|
|
|
|
class SubscriptionTierUpdate(BaseModel):
|
|
"""Schema for updating a subscription tier."""
|
|
|
|
name: str | None = None
|
|
description: str | None = None
|
|
price_monthly_cents: int | None = Field(None, ge=0)
|
|
price_annual_cents: int | None = Field(None, ge=0)
|
|
orders_per_month: int | None = None
|
|
products_limit: int | None = None
|
|
team_members: int | None = None
|
|
order_history_months: int | None = None
|
|
features: list[str] | None = None
|
|
stripe_product_id: str | None = None
|
|
stripe_price_monthly_id: str | None = None
|
|
stripe_price_annual_id: str | None = None
|
|
display_order: int | None = None
|
|
is_active: bool | None = None
|
|
is_public: bool | None = None
|
|
|
|
|
|
class SubscriptionTierResponse(SubscriptionTierBase):
|
|
"""Schema for subscription tier response."""
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
# Computed fields for display
|
|
@property
|
|
def price_monthly_display(self) -> str:
|
|
"""Format monthly price for display."""
|
|
return f"€{self.price_monthly_cents / 100:.2f}"
|
|
|
|
@property
|
|
def price_annual_display(self) -> str | None:
|
|
"""Format annual price for display."""
|
|
if self.price_annual_cents is None:
|
|
return None
|
|
return f"€{self.price_annual_cents / 100:.2f}"
|
|
|
|
|
|
class SubscriptionTierListResponse(BaseModel):
|
|
"""Response for listing subscription tiers."""
|
|
|
|
tiers: list[SubscriptionTierResponse]
|
|
total: int
|
|
|
|
|
|
# ============================================================================
|
|
# Vendor Subscription Schemas
|
|
# ============================================================================
|
|
|
|
|
|
class VendorSubscriptionResponse(BaseModel):
|
|
"""Schema for vendor subscription response."""
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
vendor_id: int
|
|
tier: str
|
|
status: str
|
|
|
|
# Period info
|
|
period_start: datetime
|
|
period_end: datetime
|
|
is_annual: bool
|
|
trial_ends_at: datetime | None = None
|
|
|
|
# Usage
|
|
orders_this_period: int
|
|
orders_limit_reached_at: datetime | None = None
|
|
|
|
# Limits (effective)
|
|
orders_limit: int | None = None
|
|
products_limit: int | None = None
|
|
team_members_limit: int | None = None
|
|
|
|
# Custom overrides
|
|
custom_orders_limit: int | None = None
|
|
custom_products_limit: int | None = None
|
|
custom_team_limit: int | None = None
|
|
|
|
# Stripe
|
|
stripe_customer_id: str | None = None
|
|
stripe_subscription_id: str | None = None
|
|
|
|
# Cancellation
|
|
cancelled_at: datetime | None = None
|
|
cancellation_reason: str | None = None
|
|
|
|
# Timestamps
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class VendorSubscriptionWithVendor(VendorSubscriptionResponse):
|
|
"""Subscription response with vendor info."""
|
|
|
|
vendor_name: str
|
|
vendor_code: str
|
|
|
|
# Usage counts (for admin display)
|
|
products_count: int | None = None
|
|
team_count: int | None = None
|
|
|
|
|
|
class VendorSubscriptionListResponse(BaseModel):
|
|
"""Response for listing vendor subscriptions."""
|
|
|
|
subscriptions: list[VendorSubscriptionWithVendor]
|
|
total: int
|
|
page: int
|
|
per_page: int
|
|
pages: int
|
|
|
|
|
|
class VendorSubscriptionCreate(BaseModel):
|
|
"""Schema for admin creating a vendor subscription."""
|
|
|
|
tier: str = "essential"
|
|
status: str = "trial"
|
|
trial_days: int = 14
|
|
is_annual: bool = False
|
|
|
|
|
|
class VendorSubscriptionUpdate(BaseModel):
|
|
"""Schema for admin updating a vendor subscription."""
|
|
|
|
tier: str | None = None
|
|
status: str | None = None
|
|
custom_orders_limit: int | None = None
|
|
custom_products_limit: int | None = None
|
|
custom_team_limit: int | None = None
|
|
trial_ends_at: datetime | None = None
|
|
cancellation_reason: str | None = None
|
|
|
|
|
|
# ============================================================================
|
|
# Billing History Schemas
|
|
# ============================================================================
|
|
|
|
|
|
class BillingHistoryResponse(BaseModel):
|
|
"""Schema for billing history entry."""
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
vendor_id: int
|
|
stripe_invoice_id: str | None = None
|
|
invoice_number: str | None = None
|
|
invoice_date: datetime
|
|
due_date: datetime | None = None
|
|
|
|
# Amounts
|
|
subtotal_cents: int
|
|
tax_cents: int
|
|
total_cents: int
|
|
amount_paid_cents: int
|
|
currency: str = "EUR"
|
|
|
|
# Status
|
|
status: str
|
|
|
|
# URLs
|
|
invoice_pdf_url: str | None = None
|
|
hosted_invoice_url: str | None = None
|
|
|
|
# Description
|
|
description: str | None = None
|
|
|
|
# Timestamps
|
|
created_at: datetime
|
|
|
|
@property
|
|
def total_display(self) -> str:
|
|
"""Format total for display."""
|
|
return f"€{self.total_cents / 100:.2f}"
|
|
|
|
|
|
class BillingHistoryWithVendor(BillingHistoryResponse):
|
|
"""Billing history with vendor info."""
|
|
|
|
vendor_name: str
|
|
vendor_code: str
|
|
|
|
|
|
class BillingHistoryListResponse(BaseModel):
|
|
"""Response for listing billing history."""
|
|
|
|
invoices: list[BillingHistoryWithVendor]
|
|
total: int
|
|
page: int
|
|
per_page: int
|
|
pages: int
|
|
|
|
|
|
# ============================================================================
|
|
# Checkout & Portal Schemas
|
|
# ============================================================================
|
|
|
|
|
|
class CheckoutRequest(BaseModel):
|
|
"""Request for creating checkout session."""
|
|
|
|
tier_code: str
|
|
is_annual: bool = False
|
|
|
|
|
|
class CheckoutResponse(BaseModel):
|
|
"""Response with checkout session URL."""
|
|
|
|
checkout_url: str
|
|
session_id: str
|
|
|
|
|
|
class PortalSessionResponse(BaseModel):
|
|
"""Response with customer portal URL."""
|
|
|
|
portal_url: str
|
|
|
|
|
|
# ============================================================================
|
|
# Subscription Stats Schemas
|
|
# ============================================================================
|
|
|
|
|
|
class SubscriptionStatsResponse(BaseModel):
|
|
"""Subscription statistics for admin dashboard."""
|
|
|
|
total_subscriptions: int
|
|
active_count: int
|
|
trial_count: int
|
|
past_due_count: int
|
|
cancelled_count: int
|
|
expired_count: int
|
|
|
|
# By tier
|
|
tier_distribution: dict[str, int]
|
|
|
|
# Revenue
|
|
mrr_cents: int # Monthly recurring revenue
|
|
arr_cents: int # Annual recurring revenue
|
|
|
|
@property
|
|
def mrr_display(self) -> str:
|
|
"""Format MRR for display."""
|
|
return f"€{self.mrr_cents / 100:,.2f}"
|
|
|
|
@property
|
|
def arr_display(self) -> str:
|
|
"""Format ARR for display."""
|
|
return f"€{self.arr_cents / 100:,.2f}"
|