FastAPI rate limit dengan Redis token bucket

API publik tanpa rate limit = undangan abuse. FastAPI sering deploy multi-instance di Kubernetes, jadi rate limit in-memory tidak cukup. Token bucket di Redis lebih akurat dari sliding counter dan handle burst dengan natural. Snippet ini middleware untuk per-IP dan per-user limit.

Kode

# rate_limit.py
import time
from dataclasses import dataclass
from typing import Optional

import redis.asyncio as redis
from fastapi import HTTPException, Request, status

# Lua script atomic — refill + consume token
TOKEN_BUCKET_LUA = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local cost = tonumber(ARGV[4])

local data = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(data[1])
local last_refill = tonumber(data[2])

if tokens == nil then
  tokens = capacity
  last_refill = now
end

-- Refill berdasarkan elapsed time
local elapsed = math.max(0, now - last_refill)
tokens = math.min(capacity, tokens + elapsed * refill_rate)

local allowed = 0
local retry_after = 0

if tokens >= cost then
  tokens = tokens - cost
  allowed = 1
else
  local kekurangan = cost - tokens
  retry_after = math.ceil(kekurangan / refill_rate)
end

redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
redis.call('EXPIRE', key, 3600)

return {allowed, tokens, retry_after}
"""


@dataclass
class RateLimitResult:
    allowed: bool
    tokens_remaining: float
    retry_after_seconds: int


class TokenBucketLimiter:
    def __init__(
        self,
        redis_client: redis.Redis,
        capacity: int,
        refill_per_second: float,
        prefix: str = "rl",
    ) -> None:
        self.r = redis_client
        self.capacity = capacity
        self.refill = refill_per_second
        self.prefix = prefix
        self._script_sha: Optional[str] = None

    async def _ensure_script(self) -> str:
        if self._script_sha is None:
            self._script_sha = await self.r.script_load(TOKEN_BUCKET_LUA)
        return self._script_sha

    async def consume(self, identifier: str, cost: int = 1) -> RateLimitResult:
        sha = await self._ensure_script()
        key = f"{self.prefix}:{identifier}"
        now = time.time()

        result = await self.r.evalsha(
            sha, 1, key, self.capacity, self.refill, now, cost
        )
        allowed, tokens_remaining, retry_after = result

        return RateLimitResult(
            allowed=bool(allowed),
            tokens_remaining=float(tokens_remaining),
            retry_after_seconds=int(retry_after),
        )


def get_identifier(request: Request) -> str:
    """Pakai user_id kalau authenticated, fallback ke IP."""
    user = getattr(request.state, "user", None)
    if user and hasattr(user, "id"):
        return f"user:{user.id}"
    # Behind proxy (nginx, CF) — trust X-Forwarded-For setelah validasi
    ip = request.headers.get("X-Forwarded-For", "").split(",")[0].strip()
    ip = ip or request.client.host
    return f"ip:{ip}"


async def rate_limit_dependency(
    request: Request,
    limiter: TokenBucketLimiter,
) -> None:
    """FastAPI Depends — raise 429 kalau exceeded."""
    ident = get_identifier(request)
    result = await limiter.consume(ident, cost=1)

    if not result.allowed:
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded",
            headers={
                "Retry-After": str(result.retry_after_seconds),
                "X-RateLimit-Remaining": str(int(result.tokens_remaining)),
            },
        )

Pemakaian

# main.py
from contextlib import asynccontextmanager

import redis.asyncio as redis
from fastapi import Depends, FastAPI, Request

from rate_limit import TokenBucketLimiter, rate_limit_dependency


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)
    # 100 token capacity, refill 10/detik → burst 100, sustain 10 req/s
    app.state.limiter = TokenBucketLimiter(
        app.state.redis, capacity=100, refill_per_second=10.0
    )
    yield
    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)


async def rate_limit(request: Request) -> None:
    await rate_limit_dependency(request, request.app.state.limiter)


@app.get("/api/produk", dependencies=[Depends(rate_limit)])
async def list_produk():
    return {"produk": ["Indomie", "Aqua", "Beng-Beng"]}


@app.get("/api/heavy", dependencies=[Depends(rate_limit)])
async def heavy_endpoint(request: Request):
    # Endpoint berat — consume 5 token sekaligus
    ident = request.headers.get("X-API-Key", "anon")
    result = await request.app.state.limiter.consume(f"key:{ident}", cost=5)
    if not result.allowed:
        raise HTTPException(429, "Heavy endpoint exhausted")
    return {"data": [...]}

# Test rate limit
for i in $(seq 1 120); do
  curl -s -o /dev/null -w "%{http_code}\n" http://localhost:8000/api/produk
done
# Output: pertama 100 dapat 200, sisanya 429

Kapan dipakai

• API publik dengan free tier vs paid tier limit berbeda.
• Login endpoint — limit per IP supaya tidak brute force.
• AI / LLM endpoint — quota harian per user.
• Search endpoint yang berat di DB.
• Webhook receiver — protect dari flood upstream.

Catatan

• Lua script atomic — Redis eksekusi single-threaded, jadi script jalan tanpa race condition. Setara CAS tapi cleaner.
• script_load + evalsha — kirim Lua sekali, eksekusi pakai SHA hash. Hemat bandwidth.
• Capacity vs refill rate — capacity = burst tolerance, refill = sustained rate. User benefit kalau low traffic biasa, burst pas spike.
• X-Forwarded-For trust — hanya trust kalau di belakang proxy yang kamu kontrol (nginx, Cloudflare). Tanpa proxy, user spoof header dengan mudah.
• Headers response — kirim Retry-After dan X-RateLimit-Remaining supaya client bisa back-off dengan benar.
• Multi-tier limit — combine per-IP (untuk anti abuse) + per-user (untuk plan limit) + per-endpoint (untuk expensive endpoint).

Token bucket gak handle “rate limit per route group” otomatis. Untuk skema kompleks (free user X req/menit ke /search, Y ke /upload), bikin limiter terpisah per group atau pakai library seperti slowapi.