【2026年版】APIレート制限の設計と実装：トークンバケット・スライディングウィンドウの実践ガイド

import redis
import time

class FixedWindowRateLimiter:
    """固定ウィンドウカウンターによるレート制限"""

    def __init__(self, redis_client: redis.Redis, max_requests: int, window_seconds: int):
        self.redis = redis_client
        self.max_requests = max_requests
        self.window_seconds = window_seconds

    def is_allowed(self, client_id: str) -> bool:
        current_window = int(time.time() // self.window_seconds)
        key = f"rate_limit:{client_id}:{current_window}"

        pipe = self.redis.pipeline()
        pipe.incr(key)
        pipe.expire(key, self.window_seconds)
        results = pipe.execute()

        current_count = results[0]
        return current_count <= self.max_requests

    def get_remaining(self, client_id: str) -> int:
        current_window = int(time.time() // self.window_seconds)
        key = f"rate_limit:{client_id}:{current_window}"
        count = self.redis.get(key)
        current = int(count) if count else 0
        return max(0, self.max_requests - current)

import redis
import time

class SlidingWindowRateLimiter:
    """スライディングウィンドウカウンターによるレート制限"""

    def __init__(self, redis_client: redis.Redis, max_requests: int, window_seconds: int):
        self.redis = redis_client
        self.max_requests = max_requests
        self.window_seconds = window_seconds

    def is_allowed(self, client_id: str) -> bool:
        now = time.time()
        current_window = int(now // self.window_seconds)
        previous_window = current_window - 1

        # 現在ウィンドウ内の経過割合
        elapsed_ratio = (now % self.window_seconds) / self.window_seconds

        current_key = f"rate_limit:{client_id}:{current_window}"
        previous_key = f"rate_limit:{client_id}:{previous_window}"

        pipe = self.redis.pipeline()
        pipe.get(previous_key)
        pipe.get(current_key)
        results = pipe.execute()

        previous_count = int(results[0]) if results[0] else 0
        current_count = int(results[1]) if results[1] else 0

        # 加重平均でリクエスト数を推定
        estimated_count = previous_count * (1 - elapsed_ratio) + current_count

        if estimated_count >= self.max_requests:
            return False

        # カウントをインクリメント
        pipe = self.redis.pipeline()
        pipe.incr(current_key)
        pipe.expire(current_key, self.window_seconds * 2)
        pipe.execute()

        return True

import redis
import time

class TokenBucketRateLimiter:
    """トークンバケットによるレート制限（Redisアトミック操作）"""

    # Luaスクリプトでアトミック性を保証
    LUA_SCRIPT = """
    local key = KEYS[1]
    local max_tokens = tonumber(ARGV[1])
    local refill_rate = tonumber(ARGV[2])
    local now = tonumber(ARGV[3])
    local requested = tonumber(ARGV[4])

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

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

    -- トークンの補充
    local elapsed = now - last_refill
    tokens = math.min(max_tokens, tokens + elapsed * refill_rate)
    last_refill = now

    local allowed = 0
    if tokens >= requested then
        tokens = tokens - requested
        allowed = 1
    end

    redis.call('hmset', key, 'tokens', tokens, 'last_refill', last_refill)
    redis.call('expire', key, math.ceil(max_tokens / refill_rate) * 2)

    return {allowed, tostring(tokens)}
    """

    def __init__(self, redis_client: redis.Redis, max_tokens: int, refill_rate: float):
        self.redis = redis_client
        self.max_tokens = max_tokens
        self.refill_rate = refill_rate  # トークン/秒
        self.script = self.redis.register_script(self.LUA_SCRIPT)

    def is_allowed(self, client_id: str, tokens_requested: int = 1) -> tuple[bool, float]:
        key = f"token_bucket:{client_id}"
        now = time.time()

        result = self.script(
            keys=[key],
            args=[self.max_tokens, self.refill_rate, now, tokens_requested],
        )

        allowed = bool(result[0])
        remaining_tokens = float(result[1])
        return allowed, remaining_tokens

# プラン別のレート制限設定
RATE_LIMIT_CONFIGS = {
    "free": {
        "max_tokens": 10,
        "refill_rate": 1.0,      # 1リクエスト/秒
    },
    "pro": {
        "max_tokens": 100,
        "refill_rate": 10.0,     # 10リクエスト/秒
    },
    "enterprise": {
        "max_tokens": 1000,
        "refill_rate": 100.0,    # 100リクエスト/秒
    },
}

# エンドポイント別のトークン消費量
TOKEN_COSTS = {
    "/api/v1/chat": 5,          # AI推論は高コスト
    "/api/v1/search": 2,        # 検索は中コスト
    "/api/v1/users": 1,         # CRUD操作は低コスト
}

import redis
import time
from collections import deque

class LeakyBucketRateLimiter:
    """リーキーバケットによるレート制限"""

    def __init__(self, redis_client: redis.Redis, capacity: int, leak_rate: float):
        self.redis = redis_client
        self.capacity = capacity
        self.leak_rate = leak_rate  # リクエスト/秒

    def is_allowed(self, client_id: str) -> bool:
        key = f"leaky_bucket:{client_id}"
        now = time.time()

        pipe = self.redis.pipeline()
        pipe.llen(key)
        pipe.execute()

        queue_size = self.redis.llen(key)

        # 漏れ出た分を削除
        cutoff = now - (self.capacity / self.leak_rate)
        self.redis.zremrangebyscore(key, "-inf", cutoff)

        current_size = self.redis.zcard(key)

        if current_size < self.capacity:
            self.redis.zadd(key, {str(now): now})
            self.redis.expire(key, int(self.capacity / self.leak_rate) * 2)
            return True

        return False

import time
from functools import lru_cache

class HybridRateLimiter:
    """ローカルキャッシュとRedisを併用するハイブリッドレート制限"""

    def __init__(self, redis_limiter, local_check_interval: float = 0.1):
        self.redis_limiter = redis_limiter
        self.local_counts = {}
        self.local_check_interval = local_check_interval
        self.last_sync = {}

    def is_allowed(self, client_id: str) -> bool:
        now = time.time()

        # ローカルカウンターで高速チェック
        if client_id in self.local_counts:
            local = self.local_counts[client_id]
            if local["count"] > local["threshold"]:
                return False

        # 一定間隔でRedisと同期
        last = self.last_sync.get(client_id, 0)
        if now - last >= self.local_check_interval:
            allowed = self.redis_limiter.is_allowed(client_id)
            self.last_sync[client_id] = now
            return allowed

        # ローカルカウンターをインクリメント
        if client_id not in self.local_counts:
            self.local_counts[client_id] = {"count": 0, "threshold": 10}
        self.local_counts[client_id]["count"] += 1

        return True

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
import redis.asyncio as redis

class RateLimitMiddleware(BaseHTTPMiddleware):
    """FastAPI用レート制限ミドルウェア"""

    def __init__(self, app: FastAPI, redis_url: str, default_limit: int = 100):
        super().__init__(app)
        self.redis = redis.from_url(redis_url)
        self.default_limit = default_limit

    async def dispatch(self, request: Request, call_next):
        client_id = self._get_client_id(request)
        endpoint = request.url.path

        # レート制限チェック
        allowed, remaining, reset_at = await self._check_rate_limit(
            client_id, endpoint
        )

        if not allowed:
            return JSONResponse(
                status_code=429,
                content={
                    "error": "Too Many Requests",
                    "message": "レート制限を超過しました。しばらく待ってから再試行してください。",
                    "retry_after": reset_at,
                },
                headers={
                    "X-RateLimit-Limit": str(self.default_limit),
                    "X-RateLimit-Remaining": "0",
                    "X-RateLimit-Reset": str(reset_at),
                    "Retry-After": str(reset_at),
                },
            )

        response = await call_next(request)

        # レート制限情報をヘッダーに追加
        response.headers["X-RateLimit-Limit"] = str(self.default_limit)
        response.headers["X-RateLimit-Remaining"] = str(remaining)
        response.headers["X-RateLimit-Reset"] = str(reset_at)

        return response

    def _get_client_id(self, request: Request) -> str:
        """APIキーまたはIPアドレスからクライアントIDを取得"""
        api_key = request.headers.get("X-API-Key")
        if api_key:
            return f"apikey:{api_key}"
        forwarded = request.headers.get("X-Forwarded-For")
        if forwarded:
            return f"ip:{forwarded.split(',')[0].strip()}"
        return f"ip:{request.client.host}"

    async def _check_rate_limit(self, client_id: str, endpoint: str):
        """スライディングウィンドウカウンターでレート制限チェック"""
        # 実装は上述のSlidingWindowRateLimiterと同様
        pass

class TokenBasedRateLimiter:
    """AI API向けトークンベースのレート制限"""

    def __init__(self, redis_client, max_tokens_per_minute: int):
        self.redis = redis_client
        self.max_tokens = max_tokens_per_minute

    async def check_and_consume(
        self, client_id: str, estimated_tokens: int
    ) -> tuple[bool, int]:
        key = f"ai_tokens:{client_id}"
        window_key = f"{key}:{int(time.time() // 60)}"

        current = await self.redis.get(window_key)
        current_usage = int(current) if current else 0

        if current_usage + estimated_tokens > self.max_tokens:
            remaining = max(0, self.max_tokens - current_usage)
            return False, remaining

        pipe = self.redis.pipeline()
        pipe.incrby(window_key, estimated_tokens)
        pipe.expire(window_key, 120)
        await pipe.execute()

        remaining = self.max_tokens - current_usage - estimated_tokens
        return True, remaining

# プラン別のトークン制限
AI_TOKEN_LIMITS = {
    "free":       {"tokens_per_minute": 10_000,   "tokens_per_day": 100_000},
    "pro":        {"tokens_per_minute": 100_000,  "tokens_per_day": 2_000_000},
    "enterprise": {"tokens_per_minute": 1_000_000, "tokens_per_day": 50_000_000},
}

from prometheus_client import Counter, Histogram, Gauge

# レート制限関連のメトリクス
rate_limit_requests_total = Counter(
    "rate_limit_requests_total",
    "Total rate limit checks",
    ["client_tier", "endpoint", "result"],
)

rate_limit_remaining = Gauge(
    "rate_limit_remaining",
    "Remaining requests in current window",
    ["client_id"],
)

rate_limit_latency = Histogram(
    "rate_limit_check_duration_seconds",
    "Time spent checking rate limits",
    buckets=[0.001, 0.005, 0.01, 0.025, 0.05, 0.1],
)

# serverless.yml (Serverless Framework)
provider:
  name: aws
  runtime: python3.12
  apiGateway:
    usagePlan:
      quota:
        limit: 10000
        period: DAY
      throttle:
        burstLimit: 50
        rateLimit: 100

functions:
  api:
    handler: handler.main
    events:
      - http:
          path: /api/{proxy+}
          method: ANY