Build an “OpenAI-Style” Paywall in FastAPI

Redis rate-limiting + tiered API keys + monthly usage/credits — Dockerized, production-ish demo

🎬 “The Invisible Gatekeeper”**

“Every developer dreams of building an API that people actually use. But what happens when they do? Suddenly, your traffic spikes. Thousands of requests hit your endpoints. Your costs rise. Your database gasps for air. And that’s when most APIs break — not because the code failed, but because there was no gatekeeper.

In this project, we build that gatekeeper — a real-world Paywall System inspired by how OpenAI and other SaaS giants manage access, control costs, and scale fairly. It’s built with FastAPI, powered by Redis, and tracked by SQLite.

So how does it work? Each user — or client app — gets a unique API key. Behind the scenes, that key maps to a tier — Free, Pro, or Enterprise. Every incoming request passes through a smart middleware layer that checks: 1️⃣ Is the key valid? 2️⃣ What tier does it belong to? 3️⃣ Has this user hit their limit?

That check happens at lightning speed inside Redis, an in-memory data store that holds short-lived counters for each endpoint and user. Redis keeps your API safe from floods and DDoS-like spikes. It knows exactly how many calls each key made this minute, and enforces your 429 “Too Many Requests” response — instantly.

But we didn’t stop there. While Redis handles short-term rate limits, a lightweight SQLite database tracks long-term usage — monthly request totals, user credits, and quotas. That’s what lets you simulate a real SaaS billing model: Free users might get 100 requests a month, Pro users 10,000, and Enterprise customers — unlimited with analytics.

Together, Redis and SQLite form a perfect duo — speed and persistence. Redis protects your runtime, SQLite records your history.

And the best part? It’s fully containerized. You can run it anywhere — locally, on a VPS, or scale horizontally in Kubernetes. Redis handles global counters shared across replicas, while SQLite can evolve into PostgreSQL or MySQL for enterprise-grade tracking without touching your core logic.

This isn’t just another FastAPI demo — it’s the blueprint for how modern SaaS APIs survive and scale. You’re not just learning code — you’re learning how to build digital infrastructure that protects your product, your users, and your business.

Welcome to the FastAPI Paywall Project — where every request earns its right to exist.”

This hands-on guide walks you through a fresh FastAPI project that implements:

API-key tiers: free, pro, enterprise
Per-endpoint rate limiting in Redis (fixed window; overridable per tier/endpoint)
Monthly usage tracking in SQLite (SQLModel)
Simulated billing: monthly quotas + optional per-request credits
Mock/real OpenAI call on /chat
Docker Compose for easy spin-up

You’ll get the full folder layout, code, and how to run + test. Explanations focus on what matters for a techie building SaaS APIs.

📁 Project Structure

fastapi-paywall-demo/
├─ docker-compose.yml
├─ .env.example
├─ README.md
└─ app/
   ├─ Dockerfile
   ├─ requirements.txt
   ├─ config.py
   ├─ tiers.py
   ├─ limiter.py
   ├─ billing.py
   ├─ openai_client.py
   └─ main.py

docker-compose.yml — boots the API and Redis together.
.env.example — all knobs (API keys, tier rules, overrides).
app/Dockerfile + requirements.txt — minimal, fast builds.
config.py — env settings, DB path, flags (mock/real OpenAI).
tiers.py — parse API_KEYS, tier defaults, endpoint overrides.
limiter.py — Redis fixed-window limiter + rate headers.
billing.py — SQLite models for Account & Usage, monthly quota + credits.
openai_client.py — safe mock; flips to real OpenAI with a key.
main.py — FastAPI routes, middleware, auth dependency, orchestration.

🔑 Environment & Compose

`.env.example` (copy to `.env`)

APP_ENV=dev
APP_HOST=0.0.0.0
APP_PORT=8000

REDIS_URL=redis://redis:6379/0

# OpenAI - mocked by default
OPENAI_API_KEY=
USE_MOCK_OPENAI=1

# API key → tier mapping
API_KEYS=FREE123:free;PRO456:pro;ENT789:enterprise

# TIER_DEFAULTS: requests,window_secs,monthly_quota,per_request_cost_credits
TIER_DEFAULTS=free:10,60,100,0;pro:60,60,1000000,0;enterprise:300,60,1000000,0

# Optional per-endpoint overrides
ENDPOINT_OVERRIDES=chat|free=5,60;chat|pro=30,60;chat|enterprise=180,60

Notes

API_KEYS is your PoC identity store.
TIER_DEFAULTS defines rate limit + monthly quota + optional per-request credit cost.
ENDPOINT_OVERRIDES lets you tighten/loosen specific endpoints per tier.

`docker-compose.yml`

version: "3.9"

services:
  api:
    build: ./app
    container_name: paywall_api
    env_file: .env
    ports: ["8001:8000"]        # API on http://localhost:8001
    depends_on: [redis]
    volumes:
      - ./app/data:/app/data    # persist SQLite
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    container_name: paywall_redis
    command: ["redis-server", "--appendonly", "yes"]
    volumes:
      - redisdata:/data
    ports: ["6379:6379"]        # optional expose for local inspection
    restart: unless-stopped

volumes:
  redisdata:

Key idea: We use Redis for hot path (rate limiting) and SQLite for monthly usage and credits (kept simple for demos).

📦 Build Image & Dependencies

`app/requirements.txt`

fastapi==0.114.2
uvicorn[standard]==0.30.6
redis==5.0.8
pydantic==2.9.2
httpx==0.27.2
sqlmodel==0.0.21
python-dotenv==1.0.1

`app/Dockerfile`

FROM python:3.11-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
    curl build-essential \
  && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000
CMD ["sh","-c","uvicorn main:app --host ${APP_HOST:-0.0.0.0} --port ${APP_PORT:-8000}"]

Why sh -c? So ${APP_PORT} and ${APP_HOST} expand from environment (avoids the “not a valid integer” issue).

⚙️ App Settings

`app/config.py`

from pydantic import BaseModel
import os
from pathlib import Path

class Settings(BaseModel):
    app_env: str = os.getenv("APP_ENV", "dev")
    host: str = os.getenv("APP_HOST", "0.0.0.0")
    port: int = int(os.getenv("APP_PORT", "8000"))

    redis_url: str = os.getenv("REDIS_URL", "redis://localhost:6379/0")

    openai_api_key: str | None = os.getenv("OPENAI_API_KEY")
    use_mock_openai: bool = os.getenv("USE_MOCK_OPENAI", "1") == "1"

    api_keys_raw: str = os.getenv("API_KEYS", "FREE123:free;PRO456:pro;ENT789:enterprise")
    tier_defaults_raw: str = os.getenv(
        "TIER_DEFAULTS",
        "free:10,60,100,0;pro:60,60,1000000,0;enterprise:300,60,1000000,0"
    )
    endpoint_overrides_raw: str = os.getenv("ENDPOINT_OVERRIDES", "")

    data_dir: str = os.getenv("DATA_DIR", "/app/data")
    db_url: str = f"sqlite:///{Path(data_dir).joinpath('billing.db')}"

settings = Settings()

Highlights

Centralized configuration object (settings) imported everywhere.
db_url points to /app/data/billing.db (volume-mounted).

🧩 Tiers, Keys, Overrides

`app/tiers.py`

from __future__ import annotations
from typing import Dict, Tuple
from config import settings

# TIER_DEFAULTS: name: limit,window,monthly_quota,cost
def parse_tier_defaults(raw: str) -> Dict[str, tuple[int,int,int,int]]:
    tiers = {}
    if not raw.strip(): return tiers
    for part in raw.split(";"):
        name, spec = part.split(":")
        limit, window, monthly_quota, cost = spec.split(",")
        tiers[name.strip()] = (int(limit), int(window), int(monthly_quota), int(cost))
    return tiers

def parse_api_keys(raw: str) -> Dict[str, str]:
    m = {}
    if not raw.strip(): return m
    for part in raw.split(";"):
        key, tier = part.split(":")
        m[key.strip()] = tier.strip()
    return m

def parse_overrides(raw: str) -> Dict[str, Dict[str, tuple[int,int]]]:
    out: Dict[str, Dict[str, tuple[int,int]]] = {}
    if not raw.strip(): return out
    for part in raw.split(";"):
        scope, spec = part.split("=")
        endpoint, tier = scope.split("|")
        limit, window = spec.split(",")
        out.setdefault(endpoint.strip(), {})[tier.strip()] = (int(limit), int(window))
    return out

TIER_DEFAULTS = parse_tier_defaults(settings.tier_defaults_raw)
API_KEYS = parse_api_keys(settings.api_keys_raw)
ENDPOINT_OVERRIDES = parse_overrides(settings.endpoint_overrides_raw)

def get_tier_for_key(api_key: str | None) -> str | None:
    if not api_key: return None
    return API_KEYS.get(api_key)

def get_rate_limits(endpoint: str, tier: str) -> tuple[int, int]:
    if endpoint in ENDPOINT_OVERRIDES and tier in ENDPOINT_OVERRIDES[endpoint]:
        return ENDPOINT_OVERRIDES[endpoint][tier]
    default = TIER_DEFAULTS.get(tier)
    return (default[0], default[1]) if default else (1, 60)

def get_quota_and_cost(tier: str) -> tuple[int, int]:
    default = TIER_DEFAULTS.get(tier)
    return (default[2], default[3]) if default else (0, 0)

What you get

A simple string → numbers parser so you can tune behavior via .env.
get_rate_limits() and get_quota_and_cost() unify the rules lookup.

🚦 Rate Limiter (Redis, fixed window)

`app/limiter.py`

from __future__ import annotations
from fastapi import HTTPException, Request
import time
import redis
from config import settings

r = redis.from_url(settings.redis_url, decode_responses=True)

def window_key(api_key: str, endpoint: str, window_seconds: int) -> tuple[str, int]:
    now = int(time.time())
    window_start = now - (now % window_seconds)
    window_end = window_start + window_seconds
    ttl = max(0, window_end - now)
    key = f"rl:{api_key}:{endpoint}:{window_start}:{window_seconds}"
    return key, ttl

async def enforce_rate_limit(request: Request, *, api_key: str, endpoint: str, limit: int, window_seconds: int):
    k, ttl = window_key(api_key, endpoint, window_seconds)
    pipe = r.pipeline()
    pipe.incr(k, amount=1)     # atomic increment for this window
    pipe.ttl(k)                # see if TTL exists
    current, current_ttl = pipe.execute()
    if current_ttl == -1:
        r.expire(k, ttl)       # align key expiry with window

    remaining = max(0, limit - current)
    request.state.rate_limit = {
        "limit": limit, "remaining": remaining if remaining > 0 else 0, "reset": ttl
    }

    if current > limit:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded",
            headers={
                "X-RateLimit-Limit": str(limit),
                "X-RateLimit-Remaining": "0",
                "X-RateLimit-Reset": str(ttl),
                "Retry-After": str(ttl),
            },
        )

def attach_headers(response, request):
    rl = getattr(request.state, "rate_limit", None)
    if rl:
        response.headers["X-RateLimit-Limit"] = str(rl["limit"])
        response.headers["X-RateLimit-Remaining"] = str(rl["remaining"])
        response.headers["X-RateLimit-Reset"] = str(rl["reset"])

Why fixed-window? It’s the simplest to understand and demo. You can later swap to token bucket for smoother bursts.

💳 Billing, Quotas, Credits

`app/billing.py`

from __future__ import annotations
from datetime import datetime
from typing import Optional, Tuple
from sqlmodel import SQLModel, Field, Session, create_engine, select
from config import settings
from tiers import get_quota_and_cost

engine = create_engine(settings.db_url, echo=False, connect_args={"check_same_thread": False})

class Account(SQLModel, table=True):
    api_key: str = Field(primary_key=True)
    tier: str
    status: str = "active"      # active | suspended
    credits: int = 0            # demo credits (if you charge per request)

class Usage(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    api_key: str
    year_month: str             # e.g., "2025-10"
    requests: int = 0

def init_db():
    SQLModel.metadata.create_all(engine)

def ensure_account(api_key: str, tier: str):
    with Session(engine) as s:
        acc = s.get(Account, api_key)
        if not acc:
            s.add(Account(api_key=api_key, tier=tier, status="active", credits=0))
            s.commit()

def ym_now() -> str:
    d = datetime.utcnow()
    return f"{d.year:04d}-{d.month:02d}"

def get_usage(s: Session, api_key: str, year_month: str) -> Usage:
    u = s.exec(select(Usage).where(Usage.api_key==api_key, Usage.year_month==year_month)).first()
    if not u:
        u = Usage(api_key=api_key, year_month=year_month, requests=0)
        s.add(u); s.commit(); s.refresh(u)
    return u

def add_credits(api_key: str, amount: int) -> int:
    with Session(engine) as s:
        acc = s.get(Account, api_key)
        if not acc: raise ValueError("Unknown API key")
        acc.credits += amount
        s.add(acc); s.commit()
        return acc.credits

def get_account(api_key: str) -> Account | None:
    with Session(engine) as s:
        return s.get(Account, api_key)

def billable_check_and_increment(api_key: str, tier: str) -> Tuple[int, int, int]:
    """
    Enforce monthly quota + optional per-request credits.
    Return (usage_after, monthly_quota, credits_after).
    Raise RuntimeError for quota/credits or ValueError when suspended.
    """
    monthly_quota, per_req_cost = get_quota_and_cost(tier)
    with Session(engine) as s:
        acc = s.get(Account, api_key)
        if not acc: raise ValueError("Account missing")
        if acc.status != "active": raise ValueError("Account suspended")

        u = get_usage(s, api_key, ym_now())
        if u.requests + 1 > monthly_quota:
            raise RuntimeError("Monthly quota exceeded")  # → 402

        if per_req_cost > 0:
            if acc.credits < per_req_cost:
                raise RuntimeError("Insufficient credits")
            acc.credits -= per_req_cost
            s.add(acc)

        u.requests += 1
        s.add(u)
        s.commit()
        return (u.requests, monthly_quota, acc.credits)

Design choices

SQLite is enough to demonstrate monthly usage tracking without running a DB service.
Monthly key is YYYY-MM; rotate naturally each month.

🤖 OpenAI Client (Mockable)

`app/openai_client.py`

from __future__ import annotations
from config import settings
import httpx

OPENAI_API_BASE = "https://api.openai.com/v1"

async def chat_complete(user_prompt: str) -> str:
    if settings.use_mock_openai or not settings.openai_api_key:
        return f"[MOCKED] You said: {user_prompt[:80]}..."
    headers = {
        "Authorization": f"Bearer {settings.openai_api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": "gpt-4o-mini",
        "messages": [
            {"role":"system","content":"You are concise."},
            {"role":"user","content": user_prompt}
        ],
        "temperature": 0.2
    }
    async with httpx.AsyncClient(timeout=30) as client:
        res = await client.post(f"{OPENAI_API_BASE}/chat/completions", headers=headers, json=payload)
        res.raise_for_status()
        data = res.json()
        return data["choices"][0]["message"]["content"].strip()

🛠️ FastAPI App

`app/main.py`

from __future__ import annotations
from typing import Optional
from fastapi import FastAPI, Request, Depends, Header, HTTPException
from fastapi.responses import JSONResponse
from config import settings
from tiers import get_tier_for_key, get_rate_limits, API_KEYS
from limiter import enforce_rate_limit, attach_headers
from billing import init_db, ensure_account, billable_check_and_increment, add_credits, get_account
from openai_client import chat_complete

app = FastAPI(title="FastAPI Paywall Demo", version="1.0.0")

@app.on_event("startup")
def startup():
    init_db()
    # Pre-create accounts for configured keys (for demo simplicity)
    for k, tier in API_KEYS.items():
        ensure_account(k, tier)

@app.middleware("http")
async def add_headers(request: Request, call_next):
    response = await call_next(request)
    attach_headers(response, request)   # add X-RateLimit-* if present
    return response

# Dependency: require X-API-Key and map to tier
async def require_api_key(x_api_key: Optional[str] = Header(default=None, alias="X-API-Key")) -> tuple[str, str]:
    tier = get_tier_for_key(x_api_key or "")
    if not x_api_key or not tier:
        raise HTTPException(status_code=401, detail="Invalid or missing API key")
    return x_api_key, tier

@app.get("/health")
async def health():
    return {"status": "ok"}

@app.get("/whoami")
async def whoami(api: tuple[str,str] = Depends(require_api_key)):
    api_key, tier = api
    acc = get_account(api_key)
    credits = acc.credits if acc else 0
    return {"api_key": api_key, "tier": tier, "credits": credits}

@app.post("/chat")
async def chat(payload: dict, request: Request, api: tuple[str,str] = Depends(require_api_key)):
    api_key, tier = api
    # 1) Cheap check: rate limit (429)
    limit, window = get_rate_limits("chat", tier)
    await enforce_rate_limit(request, api_key=api_key, endpoint="chat", limit=limit, window_seconds=window)

    # 2) Paywall: monthly quota + credits (402 or 403)
    try:
        month_used, monthly_quota, credits_after = billable_check_and_increment(api_key, tier)
    except RuntimeError as e:
        raise HTTPException(status_code=402, detail=str(e))
    except ValueError as e:
        raise HTTPException(status_code=403, detail=str(e))

    prompt = (payload or {}).get("prompt", "")
    if not prompt:
        raise HTTPException(status_code=422, detail="Missing 'prompt' in body")
    answer = await chat_complete(prompt)
    return {
        "tier": tier,
        "usage_this_month": month_used,
        "monthly_quota": monthly_quota,
        "credits_after": credits_after,
        "answer": answer
    }

@app.post("/ingest")
async def ingest(payload: dict, request: Request, api: tuple[str,str] = Depends(require_api_key)):
    api_key, tier = api
    limit, window = get_rate_limits("ingest", tier)
    await enforce_rate_limit(request, api_key=api_key, endpoint="ingest", limit=limit, window_seconds=window)

    try:
        month_used, monthly_quota, _ = billable_check_and_increment(api_key, tier)
    except RuntimeError as e:
        raise HTTPException(status_code=402, detail=str(e))
    except ValueError as e:
        raise HTTPException(status_code=403, detail=str(e))

    item = (payload or {}).get("item", {})
    return {"stored": True, "item": item, "usage_this_month": month_used, "monthly_quota": monthly_quota}

# Demo helper: add credits to the calling API key
@app.post("/billing/add-credits/{amount}")
async def addcredits(amount: int, api: tuple[str,str] = Depends(require_api_key)):
    api_key, _ = api
    new_balance = add_credits(api_key, amount)
    return {"credits": new_balance}

Execution order for /chat

Auth: require_api_key maps header → tier (401 if invalid).
Rate limit: fast Redis check (429 if exceeded).
Paywall: monthly quota & credits (402/403 if exceeded/suspended).
Work: call OpenAI (mock by default).
Headers: middleware adds X-RateLimit-* to the response.

▶️ Run It

git clone <your-repo-or-folder>
cd fastapi-paywall-demo
cp .env.example .env
docker compose up --build
# Open docs: http://localhost:8001/docs

🧪 Test It (curl)

Health

curl -s http://localhost:8001/health | jq

Who am I

curl -s -H "X-API-Key: FREE123" http://localhost:8001/whoami | jq

Rate-limited /chat for Free (override 5/60s) Watch headers & 429 on overflow:

for i in {1..7}; do
  echo "Request $i"
  curl -s -i -H "X-API-Key: FREE123" -H "Content-Type: application/json" \
    -d '{"prompt":"Hi"}' http://localhost:8001/chat | sed -n '1,40p'
  echo
done

Ingest (uses tier defaults if no override)

curl -s -H "X-API-Key: FREE123" -H "Content-Type: application/json" \
  -d '{"item":{"id":1,"name":"demo"}}' http://localhost:8001/ingest | jq

Credits (if you set a non-zero per-request cost in .env)

curl -s -X POST -H "X-API-Key: PRO456" http://localhost:8001/billing/add-credits/50 | jq

Switch to real OpenAI Edit .env: set OPENAI_API_KEY=<your-key> and USE_MOCK_OPENAI=0, then:

docker compose up -d --build

🧭 Design Notes & Extensions

Why rate-limit before paywall? Rate checks are cheap, protect infra from abuse, and avoid DB write load when already over limit.
Scaling
- Redis centralizes limits across many API instances.
- Move SQLite → Postgres for multi-instance usage tracking.
- Put API keys in a real DB and cache into Redis.
Add Token Bucket Replace fixed window in limiter.py with a token bucket (store tokens + last_refill per key). Better burst handling; same headers.
Enterprise knobs
- Per-model quotas, daily/weekly windows, org-level keys.
- SLA flags, priority queues, Prometheus /metrics.
Security
- Sign keys (prefix/claims), rotate keys.
- Add admin auth to /billing/add-credits (Basic/OAuth2).

🧰 Troubleshooting

401: Missing/unknown X-API-Key. Use one from .env (FREE123, PRO456, ENT789).
429: Rate limit exceeded. Check X-RateLimit-* headers.
402: Monthly quota or credits exceeded (paywall).
Redis errors: Ensure REDIS_URL=redis://redis:6379/0 (Compose network).
Port not expanding: We already use CMD ["sh","-c", ...] in Dockerfile.

SEEING

Excellent — that’s a crucial next step when debugging or demonstrating rate limiting and paywall internals. Let’s break it down cleanly 👇

🧭 How to View Redis Data for Your FastAPI Paywall

Your Docker Compose project already spins up a redis container — the in-memory database where all rate-limiting counters and temporary tokens are stored. You can inspect what’s being stored and how it changes in real time.

⚙️ 1. Connect to the Redis CLI (inside container)

Run this from your project root:

docker exec -it paywall_redis redis-cli

You’ll enter the interactive Redis shell:

127.0.0.1:6379>

Now you can run any Redis command.

🧱 2. List All Keys

To see everything stored by your FastAPI limiter and paywall system:

KEYS *

You’ll see entries like:

1) "rl:FREE123:chat:1728373200:60"
2) "rl:FREE123:ingest:1728373260:60"

Explanation of naming convention:

rl:{api_key}:{endpoint}:{window_start}:{window_size}

For example:

rl:FREE123:chat:1728373200:60
→ API key = FREE123
→ endpoint = chat
→ window started at Unix time 1728373200
→ window length = 60 seconds

🔍 3. View Individual Key Data

Each Redis key stores just a numeric counter for that window:

GET rl:FREE123:chat:1728373200:60

Output example:

"3"

→ means 3 requests made in that 60-second window.

🧹 4. Check TTL (time to reset)

Every key expires automatically at the end of its window.

TTL rl:FREE123:chat:1728373200:60

Output:

→ 45 seconds left until it resets and gets removed.

🔁 5. Watch It Update in Real-Time

Keep one terminal running redis-cli monitor:

docker exec -it paywall_redis redis-cli monitor

You’ll see every Redis command executed by FastAPI, e.g.:

1663178392.241392 [0 172.20.0.2:45750] "INCR" "rl:FREE123:chat:1728373200:60"
1663178392.241422 [0 172.20.0.2:45750] "TTL" "rl:FREE123:chat:1728373200:60"
1663178392.241455 [0 172.20.0.2:45750] "EXPIRE" "rl:FREE123:chat:1728373200:60" "45"

That’s proof your rate limiter is active.

🧠 6. List and Inspect Everything via Python Shell (optional)

If you prefer using Python instead of CLI, run this:

docker exec -it paywall_api python

Inside the Python shell:

import redis
r = redis.from_url("redis://redis:6379/0", decode_responses=True)
r.keys("*")

Then inspect:

for k in r.keys("*"):
    print(k, r.get(k), "TTL:", r.ttl(k))

💾 7. Understanding What Redis Stores (in this demo)

Key pattern	Purpose	Example Value
`rl:{api_key}:{endpoint}:{window}:{window_size}`	rate-limiter counters	`"5"`
(you can add) `tokens:{api_key}`	for token-bucket variant (if implemented)	`"7"`
(future) `credits_cache:{api_key}`	cached credit balance	`"120"`

Only rate-limit counters are currently persisted. All other data (accounts, usage, credits) are in SQLite, not Redis.

🧰 8. Bonus: Inspect SQLite Usage Data Too

To see monthly usage/credits data (stored in /app/data/billing.db):

docker exec -it paywall_api sqlite3 /app/data/billing.db

Then run:

.tables
SELECT * FROM account;
SELECT * FROM usage;
.exit

Example:

api_key     tier       status   credits
----------  ----------  -------  --------
FREE123     free        active   0
PRO456      pro         active   50

🧼 9. Reset Everything

If you want to start clean (new month or fresh rate limits):

docker exec -it paywall_redis redis-cli FLUSHALL
docker exec -it paywall_api rm -f /app/data/billing.db
docker compose restart

🚀 10. GUI Alternatives

Prefer a graphical interface?

RedisInsight (by Redis Labs) — free desktop app: https://redis.io/insight
Connect to redis://localhost:6379 (with your compose running). It shows keys, TTLs, and live updates visually.

✅ Summary

Goal	Command
Enter Redis shell	`docker exec -it paywall_redis redis-cli`
List all keys	`KEYS *`
Show counter for a key	`GET <key>`
TTL for a key	`TTL <key>`
Monitor all ops	`redis-cli monitor`
Inspect SQLite usage	`sqlite3 /app/data/billing.db`

✅ Wrap-Up

You now have a fully working, Dockerized FastAPI paywall:

tiered API keys → who can do how much
Redis rate limiting → protects your infra
SQLite usage/credits → monetizes access
Mock/real OpenAI → demo or production ready

This is the backbone of an OpenAI-style SaaS API. Plug in a real DB + billing provider (e.g., Stripe) and you’re on your way to a production system.

DEBUG

Perfect — let’s add a clean, developer-only /debug/redis endpoint that lists what’s inside Redis: rate-limit counters, TTLs, and optionally any cached billing data later.

This endpoint will help you inspect the real-time state of your API usage and limits without touching the CLI.

🧩 File: `app/debug_routes.py`

Create a new file inside the app/ folder:

# app/debug_routes.py
from fastapi import APIRouter, Depends, HTTPException
from limiter import r   # Redis client instance
from tiers import API_KEYS
from config import settings

router = APIRouter(prefix="/debug", tags=["debug"])

@router.get("/redis")
async def debug_redis():
    """
    Returns all Redis keys, their values, and TTLs.
    Only for debugging—disable in production.
    """
    try:
        keys = r.keys("*")
        data = []
        for k in keys:
            val = r.get(k)
            ttl = r.ttl(k)
            data.append({
                "key": k,
                "value": val,
                "ttl_seconds": ttl,
            })
        return {
            "redis_url": settings.redis_url,
            "keys_found": len(keys),
            "entries": data,
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

🧠 What It Does

Uses the same Redis client (r) from your limiter module.
Fetches all keys (r.keys('*')).
For each key:
- Gets the current value (GET).
- Gets the remaining TTL (TTL).
Returns a JSON structure like this:

{
  "redis_url": "redis://redis:6379/0",
  "keys_found": 2,
  "entries": [
    {
      "key": "rl:FREE123:chat:1728373200:60",
      "value": "3",
      "ttl_seconds": 37
    },
    {
      "key": "rl:FREE123:ingest:1728373260:60",
      "value": "2",
      "ttl_seconds": 58
    }
  ]
}

⚙️ Integrate It — Modify `app/main.py`

At the bottom of your imports:

from debug_routes import router as debug_router

Then register the router near the bottom of the file (after other endpoints):

app.include_router(debug_router)

✅ Test It

After rebuilding or restarting your stack:

docker compose up --build

Now open in browser or curl:

curl -s http://localhost:8001/debug/redis | jq

Sample output:

{
  "redis_url": "redis://redis:6379/0",
  "keys_found": 3,
  "entries": [
    {"key":"rl:FREE123:chat:1728373200:60","value":"4","ttl_seconds":45},
    {"key":"rl:FREE123:ingest:1728373260:60","value":"1","ttl_seconds":52},
    {"key":"rl:PRO456:chat:1728373300:60","value":"9","ttl_seconds":57}
  ]
}

🧱 Optional Enhancements

Enhancement	Description
Tier Filter	Add a query param like `/debug/redis?tier=free` to show only that tier’s keys.
Admin Key Protection	Require an `X-Debug-Key` header to view results.
Auto-clean	Add a `DELETE /debug/redis` endpoint to flush all Redis data (for dev resets).

Example header-protection snippet:

from fastapi import Header

@router.get("/redis")
async def debug_redis(x_debug_key: str = Header(None)):
    if x_debug_key != "SECRETDEBUG":
        raise HTTPException(status_code=401, detail="Unauthorized")
    ...

💡 Summary

Task	Endpoint	Example
List all Redis rate-limit keys	`GET /debug/redis`	`/debug/redis`
See key values and TTL	Response JSON	TTL countdown shows reset time
Optional security	Add `X-Debug-Key` header	`-H "X-Debug-Key: SECRETDEBUG"`

Awesome—let’s add a SQLite debug API so you can inspect accounts, monthly usage, and credits from your billing DB right from the browser/curl, just like Redis.

Below I’ll show you how to extend your existing debug_routes.py with /debug/sqlite endpoints, plus quick tests.

1) Extend `app/debug_routes.py`

Append these imports and routes to the same file where you added /debug/redis:

# app/debug_routes.py  (append below existing code)

from typing import Optional, Literal
from sqlmodel import Session, select
from billing import engine, Account, Usage, ym_now

# Helper: serialize SQLModel rows safely
def _account_dict(a: Account):
    return {
        "api_key": a.api_key,
        "tier": a.tier,
        "status": a.status,
        "credits": a.credits,
    }

def _usage_dict(u: Usage):
    return {
        "id": u.id,
        "api_key": u.api_key,
        "year_month": u.year_month,
        "requests": u.requests,
    }

@router.get("/sqlite")
async def debug_sqlite(
    api_key: Optional[str] = None,
    month: Optional[str] = None,           # format "YYYY-MM", defaults to current UTC month
    show: Literal["both", "accounts", "usage"] = "both",
    limit: int = 100,                       # simple cap to avoid huge payloads
):
    """
    Inspect SQLite billing data:
      - Accounts (tier, status, credits)
      - Usage (requests per month)

    Query params:
      - api_key: filter to one key
      - month: "YYYY-MM" (default: current UTC month)
      - show: "both" | "accounts" | "usage"
      - limit: max rows per section
    """
    result = {"db_url": str(engine.url), "month_default": ym_now()}

    with Session(engine) as s:
        # Accounts
        if show in ("both", "accounts"):
            if api_key:
                acc = s.get(Account, api_key)
                accounts = [_account_dict(acc)] if acc else []
            else:
                acc_rows = s.exec(select(Account).limit(limit)).all()
                accounts = [_account_dict(a) for a in acc_rows]
            result["accounts"] = accounts

        # Usage (current month by default or override)
        if show in ("both", "usage"):
            target_month = month or ym_now()
            q = select(Usage).where(Usage.year_month == target_month)
            if api_key:
                q = q.where(Usage.api_key == api_key)
            q = q.limit(limit)
            usage_rows = s.exec(q).all()
            usage = [_usage_dict(u) for u in usage_rows]
            result["usage"] = {"month": target_month, "rows": usage}

    return result


@router.get("/sqlite/all-months")
async def debug_sqlite_all_months(
    api_key: Optional[str] = None,
    limit: int = 500
):
    """
    Inspect usage across ALL months (capped by `limit`).
    Useful to see historical growth for a key or entire system.
    """
    with Session(engine) as s:
        q = select(Usage)
        if api_key:
            q = q.where(Usage.api_key == api_key)
        q = q.limit(limit)
        rows = s.exec(q).all()
        return {
            "db_url": str(engine.url),
            "count": len(rows),
            "entries": [_usage_dict(u) for u in rows],
        }

(Optional) protect debug endpoints behind a header

Add this near the top, and check it in both handlers:

from fastapi import Header

DEBUG_SECRET = "SECRETDEBUG"  # change or move to env if needed

def _require_debug_key(x_debug_key: Optional[str]):
    if x_debug_key != DEBUG_SECRET:
        raise HTTPException(status_code=401, detail="Unauthorized")

Then in each route signature add x_debug_key: Optional[str] = Header(None) and call _require_debug_key(x_debug_key) at the start.

2) Ensure router is registered

You already added this in main.py, but double-check it exists:

from debug_routes import router as debug_router
app.include_router(debug_router)

Rebuild and run:

docker compose up --build

3) Try it out

List accounts + current month usage

curl -s http://localhost:8001/debug/sqlite | jq

Sample output

{
  "db_url": "sqlite:////app/data/billing.db",
  "month_default": "2025-10",
  "accounts": [
    {"api_key":"FREE123","tier":"free","status":"active","credits":0},
    {"api_key":"PRO456","tier":"pro","status":"active","credits":0},
    {"api_key":"ENT789","tier":"enterprise","status":"active","credits":0}
  ],
  "usage": {
    "month": "2025-10",
    "rows": [
      {"id":1,"api_key":"FREE123","year_month":"2025-10","requests":4},
      {"id":2,"api_key":"PRO456","year_month":"2025-10","requests":1}
    ]
  }
}

Filter by API key

curl -s "http://localhost:8001/debug/sqlite?api_key=FREE123" | jq

Only accounts (no usage)

curl -s "http://localhost:8001/debug/sqlite?show=accounts" | jq

Only usage for a specific month

curl -s "http://localhost:8001/debug/sqlite?show=usage&month=2025-11" | jq

All months (historical) for one key

curl -s "http://localhost:8001/debug/sqlite/all-months?api_key=FREE123" | jq

Tips & Good Practices

Keep these debug routes dev-only (wrap in a flag or header as shown).
If the response is large, raise limit= or add pagination params.
For multi-instance/production, move SQLite → Postgres and add indexes on (api_key, year_month).

CONCLUSION

Absolutely 🔥 — here’s a cinematic, emotionally satisfying outro to close your FastAPI Paywall video. It’s written to inspire developers, reinforce technical depth, and leave a sense of mastery and anticipation for your next video. You can record it as voiceover or use it as the final monologue on screen (duration ~45–60 seconds).

🎬 “Build the Gatekeeper”**

“What you’ve just seen isn’t just another API project — it’s a framework for how modern SaaS truly works. Every request, every token, every limit — now has a story, a boundary, and a purpose.

Redis stands guard — lightning-fast, enforcing fairness. SQLite keeps the records — ensuring accountability. Together, they transform a simple FastAPI app into a living, scalable ecosystem — one that protects your backend and powers your business.

So the next time you ship an API, remember: performance is not enough — policy matters. Security matters. Monetization matters.

This is how you build systems that last. This is how you grow from developer to architect.

I’m Sreeprakash Neelakantan — and this was the FastAPI Paywall Project.

Go ahead… build your gatekeeper.”