Michael Chihlas
2f8ec3775e
Authed users can now request a Stripe-hosted Customer Portal URL for card
updates and cancellation via GET /api/v1/billing/portal-session. The path is
already in both _SUBSCRIPTION_GUARD_ALLOWLIST and _EMAIL_VERIFICATION_ALLOWLIST
so canceled or unverified-past-grace users can still update billing.
- Returns 503 with {"error": "stripe_not_configured"} when STRIPE_SECRET_KEY unset.
- Returns 400 with {"error": "no_stripe_customer"} when account has no
stripe_customer_id (must complete checkout first).
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
77 lines
2.9 KiB
Python
77 lines
2.9 KiB
Python
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.deps import get_current_active_user
|
|
from app.core.admin_database import get_admin_db
|
|
from app.core.config import settings
|
|
from app.models.account import Account
|
|
from app.models.user import User
|
|
from app.schemas.billing import (
|
|
BillingPortalSessionResponse,
|
|
BillingStateResponse,
|
|
CheckoutSessionCreate,
|
|
CheckoutSessionResponse,
|
|
)
|
|
from app.services.billing import BillingService
|
|
|
|
router = APIRouter(prefix="/billing", tags=["billing"])
|
|
|
|
|
|
@router.post("/checkout-session", response_model=CheckoutSessionResponse)
|
|
async def create_checkout_session(
|
|
payload: CheckoutSessionCreate,
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
db: Annotated[AsyncSession, Depends(get_admin_db)],
|
|
) -> CheckoutSessionResponse:
|
|
account = (await db.execute(
|
|
select(Account).where(Account.id == current_user.account_id)
|
|
)).scalar_one()
|
|
url = await BillingService.create_checkout_session(
|
|
db=db,
|
|
account=account,
|
|
plan=payload.plan,
|
|
seats=payload.seats,
|
|
billing_interval=payload.billing_interval,
|
|
success_url=f"{settings.FRONTEND_URL}/account/billing?success=1",
|
|
cancel_url=f"{settings.FRONTEND_URL}/account/billing/select-plan",
|
|
)
|
|
return CheckoutSessionResponse(url=url)
|
|
|
|
|
|
@router.get("/state", response_model=BillingStateResponse)
|
|
async def get_billing_state(
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
db: Annotated[AsyncSession, Depends(get_admin_db)],
|
|
) -> BillingStateResponse:
|
|
account = (await db.execute(
|
|
select(Account).where(Account.id == current_user.account_id)
|
|
)).scalar_one()
|
|
state = await BillingService.get_billing_state(db, account)
|
|
return BillingStateResponse(**state)
|
|
|
|
|
|
@router.get("/portal-session", response_model=BillingPortalSessionResponse)
|
|
async def get_billing_portal_session(
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
db: Annotated[AsyncSession, Depends(get_admin_db)],
|
|
) -> BillingPortalSessionResponse:
|
|
"""Return a Stripe-hosted Customer Portal URL for the account so the user
|
|
can update card / cancel. Allowlisted from the subscription + email-verify
|
|
guards (a canceled or unverified-past-grace user must still be able to
|
|
update billing)."""
|
|
if not settings.stripe_enabled:
|
|
raise HTTPException(status_code=503, detail={"error": "stripe_not_configured"})
|
|
|
|
account = (await db.execute(
|
|
select(Account).where(Account.id == current_user.account_id)
|
|
)).scalar_one()
|
|
|
|
try:
|
|
url = await BillingService.open_customer_portal(account)
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail={"error": "no_stripe_customer"})
|
|
return BillingPortalSessionResponse(url=url)
|