McTaba Labs logo
Bonaventure OgetoBy Bonaventure Ogeto|

Build Paystack Subscriptions in FastAPI

To build Paystack subscriptions in FastAPI, create a plan via the API, initialize a transaction with the plan code using an async endpoint, handle subscription webhooks to track status, and use FastAPI dependencies to gate premium endpoints behind active subscription checks.

Subscribe Endpoint

@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'],
    }

Subscription Webhook Handlers

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')

Subscription Access Dependency

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

Cancel Subscription

@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'))

Subscription Status Endpoint

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,
    )

Ship Payments Faster

Key Takeaways

  • ✓Paystack handles all recurring billing. You create plans, subscribe customers, and Paystack charges them automatically.
  • ✓Use async httpx calls for all Paystack API interactions in FastAPI.
  • ✓Webhook events drive subscription state. Handle subscription.create, charge.success, and invoice.payment_failed.
  • ✓FastAPI dependencies work well for subscription access control. Create a dependency that checks the user has an active subscription.
  • ✓Store subscription_code and email_token from the webhook. You need both for managing subscriptions.
  • ✓Pydantic models give you type-safe subscription responses for your frontend.

Frequently Asked Questions

Can I offer free trials with Paystack and FastAPI?
Paystack does not have native free trials. Implement trials in your FastAPI app: grant access for the trial period, then initialize the subscription payment when the trial ends.
How do I handle plan upgrades?
Cancel the current subscription and create a new one on the desired plan. Handle prorating in your application logic.
What is the email_token?
The email_token is sent in the subscription.create webhook. You need it alongside the subscription_code to cancel a subscription via the Paystack API. Store it when you receive the webhook.

Ready to build real-world apps?

Join the McTaba Labs full-stack marathon (4 months full-time · 6 months part-time). Learn M-Pesa, USSD, and WhatsApp engineering while shipping 8 production apps.

Apply to the McTaba Marathon