subs-webhook 0.4.0

A lightweight, asynchronous FastAPI extension designed to handle SaaS subscriptions, API key management, and plan-based access control. It integrates directly with Pabbly (and is extensible for other providers) to sync subscription statuses and automates access control using Redis caching for high performance.

🚀 Subscription Management Webhook Handler

A lightweight, asynchronous FastAPI extension designed to handle SaaS subscriptions, API key management, and plan-based access control. It integrates directly with Pabbly (and is extensible for other providers) to sync subscription statuses and automates access control using Redis caching for high performance.

---

✨ Features

• Plan-Based Access Control: Restrict API routes based on the user's active subscription plan.
• High Performance: Uses Redis to cache authentication results, minimizing database hits on every request.
• Async & Non-blocking: Built on SQLAlchemy (Async) and aiosqlite.
• Webhook Integration: Built-in handler for Pabbly webhooks (customer_create, subscription_create, invoice_paid, subscription_cancel) to auto-sync user status.
• Granular Feature Flags: Inject plan-specific features (e.g., rate limits, usage quotas) directly into the request.state.
• Automatic DB Management: Handles SQLite WAL mode and table creation automatically.

---

🛠 Prerequisites

• Python 3.9+
• Redis (Required for caching auth tokens)
• FastAPI application

---

🚀 Quick Start

1. Initialize the System

Ensure package is installed:

pip install subs-webhook

In your main main.py file, initialize the subscription system. This sets up the SQLite database, Redis connection, and webhook routes.

from fastapi import FastAPI
from pathlib import Path
from subs_webhook import init_subs

app = FastAPI()

# Configuration
SUBS_DB_PATH = Path("./subscriptions.db")
REDIS_URL = "redis://localhost:6379/0"

# Initialize Subscriptions
# This will:
# 1. Create the SQLite DB if missing.
# 2. Register webhook routes at /api/v1/subscription-webhook/...
init_subs(
    app,
    sqlite_path=SUBS_DB_PATH,
    redis_url=REDIS_URL,
    prefix="/api/v1"
)

2. Define Your Plans

Create a configuration dictionary (or load it from a JSON file). This maps plan names to allowed routes and specific feature flags.

plans_config.json

{
  "basic_tier": {
    "features": { "rate_limit": 100, "access_level": "standard" },
    "routes": ["/api/v1/dashboard", "/api/v1/reports/summary"]
  },
  "pro_tier": {
    "features": { "rate_limit": 5000, "access_level": "premium" },
    "routes": [
      "/api/v1/dashboard",
      "/api/v1/reports/summary",
      "/api/v1/reports/advanced",
      "/api/v1/export"
    ]
  }
}

3. Protect Your Routes

Use the validate_access and get_api_key dependencies to secure your endpoints.

import json
from fastapi import APIRouter, Depends, Request, HTTPException
from subs_webhook import validate_access, get_api_key

# Load plan configuration
with open('plans_config.json', 'r') as f:
    sub_plan_permissions = json.load(f)

# Create a protected router
router = APIRouter(
    prefix="/api/v1",
    dependencies=[
        # 1. Checks for 'X-API-Key' header
        Depends(get_api_key),
        # 2. Validates Key, Checks Expiry, and Verifies Route Access
        Depends(validate_access(sub_plan_permissions))
    ]
)

@router.get("/reports/advanced")
async def get_advanced_report(request: Request):
    """
    This route is only accessible if the user's active plan
    includes '/api/v1/reports/advanced' in its 'routes' list.
    """

    # Access context injected by the middleware
    current_plan = request.state.plan
    features = request.state.plan_features

    limit = features.get('rate_limit', 50)

    return {
        "message": f"Welcome user on {current_plan}",
        "authorized_limit": limit,
        "data": [...]
    }

app.include_router(router)

---

🔗 Webhook Setup (Pabbly)

To automate subscription management, you must point your payment provider or middleware (Pabbly Connect) to the webhook endpoint.

Endpoint: POST {your-domain}{prefix}/subscription-webhook/pabbly
Example: https://api.myapp.com/api/v1/subscription-webhook/pabbly

Supported Events

The system expects a JSON payload. It handles logic for:

1. New Users: Creates a Profile when a customer is created.
2. New Subscriptions: Links a Subscription to the user.
3. Renewals/Invoices: Updates current_period_end and sets status to live.
4. Cancellations: Revokes access immediately or at period end (depending on logic).

---

🔐 Manually Creating API Keys

Since the system relies on webhooks for incoming data, you might need to manually seed your first user and API key for testing.

You can use a script like this to access the database directly:

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from subs_webhook.db.subscriptions.models import Profile, Subscription, ApiKey
from datetime import datetime, timedelta

# Settings
DB_URL = "sqlite+aiosqlite:///./subscriptions.db"

async def create_seed_user():
    engine = create_async_engine(DB_URL)
    AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    async with AsyncSessionLocal() as db:
        # 1. Create Profile
        user = Profile(email="[e-mail]", pabbly_customer_id="cus_12345")
        db.add(user)
        await db.flush() # Populate user.id

        # 2. Create Subscription
        sub = Subscription(
            id="sub_test_001",
            user_id=user.id,
            plan_code="pro_tier", # Must match your JSON config key
            status="live",
            current_period_end=datetime.utcnow() + timedelta(days=365)
        )
        db.add(sub)

        # 3. Create API Key
        api_key = ApiKey(
            key="sk_live_SECRET_KEY_123",
            user_id=user.id,
            status="active"
        )
        db.add(api_key)

        await db.commit()
        print(f"Created User ID: {user.id}")
        print(f"API Key: sk_live_SECRET_KEY_123")

    await engine.dispose()

if __name__ == "__main__":
    asyncio.run(create_seed_user())

---

🧩 Architecture & Data Flow

1. Request: Client sends request with header X-API-Key: sk_....
2. Redis Check: Middleware checks Redis for a cached user_id and plan_list associated with that key.
   • Hit: Proceed to step 4.
   • Miss: Proceed to step 3.
3. DB Lookup:
   • Query ApiKey table (must be active).
   • Join Profile.
   • Join Subscription (must be live and not expired).
   • Calculate cache TTL based on the earliest expiration date.
   • Store in Redis.
4. Route Guard: The system checks if the requested URL path exists in the routes array for any of the user's active plans.
5. Context Injection: If allowed, request.state.plan and request.state.plan_features are populated for use in the endpoint.

---

⚠️ Important Notes

1. SQLite WAL Mode: The library automatically enables Write-Ahead Logging for SQLite to handle concurrent async reads/writes better.
2. Route Matching: Currently, the system uses Exact Matching for route permissions. If your config allows /api/rates, a request to /api/rates/history will be denied unless explicitly added to the list.
3. Cache Invalidation: When a webhook updates a subscription (e.g., cancellation, renewal, or plan change), the system automatically invalidates the specific user's cache in Redis. This ensures that access rights are updated immediately on the next API call, rather than waiting for the cache TTL to expire.
4. Plan Code Matching: The keys in your plans_config.json (e.g., "pro_tier", "basic") must match the plan_code sent by your payment provider (Pabbly). If they do not match, the system will recognize the user but find no mapped permissions, resulting in a 403 error.

---

⚠️ Common Issues & Troubleshooting

• "Missing X-API-Key header": Ensure your client is sending the header exactly as X-API-Key (case-insensitive in HTTP/2, but good practice to be consistent).
• 403 Forbidden (even with a valid key):
  • Check if the subscription status is live in the subscriptions table.
  • Ensure current_period_end is a date in the future.
  • Verify that the specific route path (e.g., /api/v1/rates/analytics) is explicitly listed in the routes array for the user's plan in your config.
• SQLite Database Locks: The system enables WAL (Write-Ahead Logging) mode automatically. Ensure the directory containing your .db file has write permissions so SQLite can create the necessary -wal and -shm temporary files.

---

📄 License

Copyright (c) 2026 Anthony Mugendi.

This software is released under the MIT License. https://opensource.org/licenses/MIT