feat: implement tiered subscription plans with usage-based pricing and dynamic plan management

Add multi-tier subscription system (Starter, Professional, Enterprise) with per-tier storage and bandwidth pricing, replace static free-tier limits with nullable usage-based fields in SubscriptionPlan model, introduce subscribe/unsubscribe API endpoints with plan validation, update invoice generator to calculate costs based on user's active plan tier, and refactor pricing page to load plans dynamically from backend API instead of hardcoded HTML.
This commit is contained in:
2026-01-22 15:40:16 +00:00
parent a34668941f
commit 619e320780
10 changed files with 445 additions and 148 deletions
+14
View File
@@ -234,3 +234,17 @@ async def get_new_recovery_codes(
await current_user.save()
return recovery_codes
@router.get("/me")
async def get_current_user_info(current_user: User = Depends(get_current_user)):
"""Get current user information"""
return {
"id": current_user.id,
"username": current_user.username,
"email": current_user.email,
"is_active": current_user.is_active,
"is_verified": current_user.is_verified,
"is_2fa_enabled": current_user.is_2fa_enabled,
"created_at": current_user.created_at.isoformat() if current_user.created_at else None,
}
+139
View File
@@ -408,6 +408,145 @@ async def list_plans():
]
class SubscribeRequest(BaseModel):
plan_name: str
class UnsubscribeRequest(BaseModel):
cancel_immediately: bool = False
@router.post("/subscribe")
async def subscribe_to_plan(
request: SubscribeRequest,
current_user: User = Depends(get_current_user),
):
try:
# Find the plan
plan = await SubscriptionPlan.get_or_none(
name=request.plan_name,
is_active=True
)
if not plan:
raise HTTPException(
status_code=404,
detail=f"Plan '{request.plan_name}' not found"
)
# Check if user already has a subscription
existing_subscription = await UserSubscription.get_or_none(
user=current_user
)
if existing_subscription:
# Update existing subscription
existing_subscription.plan = plan
existing_subscription.billing_type = "subscription"
await existing_subscription.save()
return {
"message": f"Successfully updated to {plan.display_name} plan",
"plan": plan.display_name,
"billing_type": "subscription"
}
else:
# Create new subscription
subscription = await UserSubscription.create(
user=current_user,
plan=plan,
billing_type="subscription",
status="active"
)
return {
"message": f"Successfully subscribed to {plan.display_name} plan",
"plan": plan.display_name,
"billing_type": "subscription",
"subscription_id": subscription.id
}
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Failed to subscribe to plan: {str(e)}"
)
@router.post("/unsubscribe")
async def unsubscribe_from_plan(
request: UnsubscribeRequest,
current_user: User = Depends(get_current_user),
):
try:
subscription = await UserSubscription.get_or_none(
user=current_user
)
if not subscription:
raise HTTPException(
status_code=404,
detail="No active subscription found"
)
if request.cancel_immediately:
# Cancel subscription immediately
await subscription.delete()
return {"message": "Subscription cancelled immediately"}
else:
# Mark for cancellation at end of billing period
subscription.status = "cancelled"
await subscription.save()
return {"message": "Subscription will be cancelled at end of billing period"}
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Failed to unsubscribe: {str(e)}"
)
@router.get("/subscription")
async def get_subscription(
current_user: User = Depends(get_current_user),
) -> SubscriptionResponse:
try:
subscription = await UserSubscription.get_or_none(
user=current_user
).prefetch_related("plan")
if not subscription:
# Return default starter subscription
default_plan = await SubscriptionPlan.get_or_none(
name="starter",
is_active=True
)
return SubscriptionResponse(
id=0,
billing_type="pay_as_you_go",
plan_name=default_plan.display_name if default_plan else "Starter",
status="active",
current_period_start=None,
current_period_end=None
)
return SubscriptionResponse(
id=subscription.id,
billing_type=subscription.billing_type,
plan_name=subscription.plan.display_name if subscription.plan else None,
status=subscription.status,
current_period_start=subscription.current_period_start,
current_period_end=subscription.current_period_end
)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Failed to fetch subscription: {str(e)}"
)
@router.get("/stripe-key")
async def get_stripe_key():
from ..settings import settings