Designing API Pricing Tiers for Python Micro-SaaS

import os
from dataclasses import dataclass

# Load environment variables for tier configuration
TIER_CONFIG = {
 "free": {"limit": 1000, "price": 0, "burst": 5},
 "pro": {"limit": 50000, "price": 29.00, "burst": 20},
 "enterprise": {"limit": 500000, "price": 199.00, "burst": 100}
}

@dataclass
class CostBaseline:
 compute_ms: float
 network_kb: float
 third_party_calls: int

def calculate_request_cost(baseline: CostBaseline) -> float:
 """Estimate infrastructure cost per request based on profiling data."""
 compute_cost = baseline.compute_ms * 0.000002 # $2/ms per vCPU-hour
 network_cost = baseline.network_kb * 0.000001 # $1/GB egress
 external_cost = baseline.third_party_calls * 0.005 # Avg SaaS API cost
 return compute_cost + network_cost + external_cost

# Apply foundational unit economics from [Building & Monetizing API-Driven Micro-SaaS](/building-monetizing-api-driven-micro-saas/) to avoid underpricing. 
# Always price at least 3x your calculated per-request cost to absorb overhead, retries, and support.

import os
import time
import redis
from typing import Optional

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
RATE_WINDOW = int(os.getenv("RATE_WINDOW_SECONDS", 3600))

# Initialize connection pool for production resilience
redis_pool = redis.ConnectionPool.from_url(REDIS_URL, max_connections=20, socket_timeout=2)

def check_rate_limit(api_key: str, tier_limit: int) -> tuple[bool, int]:
 """Atomic sliding window rate limiter using Redis sorted sets."""
 r = redis.Redis(connection_pool=redis_pool)
 now = time.time()
 key = f"ratelimit:{api_key}"
 
 try:
 # Pipeline for atomicity and reduced network round-trips
 pipe = r.pipeline()
 pipe.zremrangebyscore(key, 0, now - RATE_WINDOW)
 pipe.zcard(key)
 results = pipe.execute()
 
 current_count = results[1]
 
 if current_count >= tier_limit:
 return False, 0
 
 # Add current request with timestamp as score
 pipe = r.pipeline()
 pipe.zadd(key, {f"{now}:{os.urandom(4).hex()}": now})
 pipe.expire(key, RATE_WINDOW)
 pipe.execute()
 
 return True, tier_limit - (current_count + 1)
 except redis.RedisError as e:
 # Fail open or closed based on your tolerance. 
 # For billing, fail closed to prevent quota abuse.
 raise RuntimeError(f"Rate limiter unavailable: {e}")

import os
from fastapi import Request, HTTPException
from starlette.responses import JSONResponse
from typing import Callable

def get_tier_status(api_key: str) -> dict:
 """Mock DB lookup. Replace with async SQLAlchemy/Prisma call."""
 return {"status": "active", "limit": 50000, "remaining": 49999}

async def tier_enforcement_middleware(request: Request, call_next: Callable):
 api_key = request.headers.get("X-API-Key")
 if not api_key:
 raise HTTPException(status_code=401, detail="Missing API key")
 
 try:
 tier_status = get_tier_status(api_key)
 except Exception:
 raise HTTPException(status_code=503, detail="Billing service unavailable")
 
 if tier_status["status"] == "suspended":
 return JSONResponse(
 status_code=402,
 content={"error": "Payment required. Upgrade your tier or update billing."}
 )
 
 allowed, remaining = check_rate_limit(api_key, tier_status["limit"])
 if not allowed:
 return JSONResponse(
 status_code=429,
 content={"error": "Rate limit exceeded. Retry after window reset."},
 headers={"Retry-After": str(int(os.getenv("RATE_WINDOW_SECONDS", 3600)))}
 )
 
 response = await call_next(request)
 response.headers["X-RateLimit-Remaining"] = str(remaining)
 return response

import os
import stripe
from fastapi import Request, HTTPException

STRIPE_WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET")
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")

def verify_webhook_signature(payload: bytes, sig_header: str) -> stripe.Event:
 """Validate Stripe webhook signature to prevent spoofing."""
 try:
 event = stripe.Webhook.construct_event(
 payload, sig_header, STRIPE_WEBHOOK_SECRET
 )
 return event
 except (ValueError, stripe.error.SignatureVerificationError) as e:
 raise HTTPException(status_code=400, detail=f"Invalid signature: {e}")

async def handle_stripe_webhook(request: Request):
 payload = await request.body()
 sig_header = request.headers.get("stripe-signature")
 
 event = verify_webhook_signature(payload, sig_header)
 
 # Idempotent processing: check event ID against your DB before mutating state
 # Follow secure integration patterns from [Integrating Stripe with Python APIs](/building-monetizing-api-driven-micro-saas/integrating-stripe-with-python-apis/) to prevent fraud
 if event.type == "customer.subscription.updated":
 sub = event.data.object
 api_key = sub.metadata.get("api_key")
 status = sub.status # active, past_due, canceled
 
 # Update internal registry atomically
 update_tier_status(api_key, status, sub.plan.tier_limit)
 
 return {"status": "processed"}

import time
import asyncio
from typing import Callable, Any

async def retry_with_backoff(func: Callable, max_retries: int = 3, base_delay: float = 1.0) -> Any:
 """Resilient retry logic for billing gateway calls."""
 for attempt in range(max_retries):
 try:
 return await func()
 except (TimeoutError, ConnectionError) as e:
 if attempt == max_retries - 1:
 raise RuntimeError(f"Billing gateway unreachable after {max_retries} attempts") from e
 delay = base_delay * (2 ** attempt)
 await asyncio.sleep(delay)

async def validate_subscription(api_key: str) -> bool:
 """Check cache first, fallback to provider with backoff."""
 cached = await get_cached_tier(api_key)
 if cached and cached["expires_at"] > time.time():
 return cached["status"] == "active"
 
 async def fetch_from_provider():
 return await stripe_subscription_lookup(api_key)
 
 status = await retry_with_backoff(fetch_from_provider)
 await cache_tier(api_key, status, ttl=60)
 return status == "active"