Build Paystack Subscriptions in FastAPI

@app.post('/api/subscribe')
async def subscribe(plan_code: str, current_user = Depends(get_current_user)):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            'https://api.paystack.co/transaction/initialize',
            json={
                'email': current_user.email,
                'plan': plan_code,
                'callback_url': BASE_URL + '/subscription/callback',
            },
            headers={
                'Authorization': 'Bearer ' + PAYSTACK_SECRET,
                'Content-Type': 'application/json',
            },
        )

    data = response.json()

    if not data.get('status'):
        raise HTTPException(status_code=400, detail=data.get('message'))

    return {
        'authorization_url': data['data']['authorization_url'],
        'reference': data['data']['reference'],
    }

async def handle_subscription_create(data: dict):
    email = data['customer']['email']
    user = await db.get_user_by_email(email)
    if not user:
        return

    await db.upsert_subscription({
        'user_id': user.id,
        'subscription_code': data['subscription_code'],
        'plan_code': data['plan']['plan_code'],
        'email_token': data.get('email_token', ''),
        'status': 'active',
        'next_payment_date': data.get('next_payment_date'),
    })

async def handle_invoice_failed(data: dict):
    sub_code = data.get('subscription', {}).get('subscription_code')
    if sub_code:
        await db.update_subscription_status(sub_code, 'past_due')

# dependencies.py
async def require_active_subscription(current_user = Depends(get_current_user)):
    subscription = await db.get_subscription(current_user.id)

    if not subscription or subscription.status != 'active':
        raise HTTPException(
            status_code=403,
            detail='An active subscription is required',
        )

    return subscription

# Usage:
@app.get('/api/premium')
async def premium_content(sub = Depends(require_active_subscription)):
    return {'content': 'Premium data', 'plan': sub.plan_code}

@app.post('/api/subscription/cancel')
async def cancel_subscription(current_user = Depends(get_current_user)):
    subscription = await db.get_subscription(current_user.id)

    if not subscription:
        raise HTTPException(status_code=404, detail='No subscription found')

    async with httpx.AsyncClient() as client:
        response = await client.post(
            'https://api.paystack.co/subscription/disable',
            json={
                'code': subscription.subscription_code,
                'token': subscription.email_token,
            },
            headers={
                'Authorization': 'Bearer ' + PAYSTACK_SECRET,
                'Content-Type': 'application/json',
            },
        )

    data = response.json()

    if data.get('status'):
        await db.update_subscription_status(subscription.subscription_code, 'cancelled')
        return {'message': 'Subscription cancelled'}

    raise HTTPException(status_code=400, detail=data.get('message'))

from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class SubscriptionStatus(BaseModel):
    has_subscription: bool
    status: Optional[str] = None
    plan_code: Optional[str] = None
    next_payment_date: Optional[datetime] = None

@app.get('/api/subscription/status', response_model=SubscriptionStatus)
async def subscription_status(current_user = Depends(get_current_user)):
    subscription = await db.get_subscription(current_user.id)

    if not subscription:
        return SubscriptionStatus(has_subscription=False)

    return SubscriptionStatus(
        has_subscription=True,
        status=subscription.status,
        plan_code=subscription.plan_code,
        next_payment_date=subscription.next_payment_date,
    )

This guide is part of the Paystack integration guides by language and framework series.

Sign up for the McTaba newsletter