Files
orion/models/schema/billing.py
Samir Boulahtit c6e7f4087f feat: complete subscription billing system phases 6-10
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>
2025-12-26 20:51:13 +01:00

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}"