[Python] Implementing rate limiting with dynamic quotas based on user subscription level in...

I'm building an API that needs to handle different rate limits based on user subscription levels (free, premium, enterprise). I'm using FastAPI with Redis for rate limiting, but struggling with implementing dynamic quotas. Current implementation:

from fastapi import FastAPI, Depends, HTTPException
import redis
from typing import Optional

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, db=0)

async def get_rate_limit(user_id: str) -> int:
# Currently hardcoded, need dynamic solution
return 100 # requests per hour

@app.get("/api/data")
async def get_data(user_id: str = Depends(get_current_user)):
quota = await get_rate_limit(user_id)
# Rate limiting logic here


How can I:

1. Implement dynamic quotas that update when user changes subscription
2. Handle burst requests efficiently
3. Implement a sliding window rather than fixed window
4. Provide users with remaining quota information
5. Handle distributed environment scenarios

I've looked at various rate limiting algorithms but unsure about best practices for dynamic quotas in a distributed system.

What I tried:

1. Implemented basic Redis-based rate limiting:

async def check_rate_limit(user_id: str) -> bool:
key = f"rate_limit:{user_id}"
current = await redis_client.incr(key)
if current == 1:
await redis_client.expire(key, 3600) # 1 hour expiry
return current <= await get_rate_limit(user_id)

1. Added subscription-based quotas:

QUOTA_MAPPING = {
'free': 100,
'premium': 1000,
'enterprise': 5000
}

async def get_rate_limit(user_id: str) -> int:
subscription = await get_user_subscription(user_id)
return QUOTA_MAPPING.get(subscription, 100)

1. Attempted sliding window implementation:

async def check_sliding_window(user_id: str):
now = time.time()
window_start = now - 3600 # 1 hour window

pipeline = redis_client.pipeline()
pipeline.zadd(f"requests:{user_id}", {now: now})
pipeline.zremrangebyscore(f"requests:{user_id}", 0, window_start)
pipeline.zcard(f"requests:{user_id}")
results = await pipeline.execute()

return results[2] <= await get_rate_limit(user_id)


Expected:

• Smooth handling of quota updates when subscription changes
• Accurate tracking in distributed environment
• Proper sliding window implementation without memory leaks
• Efficient handling of burst requests

Actual results:

• Race conditions when updating quotas
• Memory usage grows unbounded with sliding window implementation
• Burst requests cause Redis performance issues
• Distributed systems face synchronization problems

Continue reading...