Rate Limiter Implementation속도 제한 구현

import time
import threading
from dataclasses import dataclass, field

@dataclass
class TokenBucket:
    capacity: int          # 최대 토큰 수
    refill_rate: float     # 초당 토큰 보충량
    tokens: float = field(init=False)
    last_refill: float = field(init=False)
    lock: threading.Lock = field(default_factory=threading.Lock, init=False)

    def __post_init__(self):
        self.tokens = self.capacity
        self.last_refill = time.monotonic()

    def consume(self, tokens: int = 1) -> bool:
        with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_refill

            # 토큰 보충
            self.tokens = min(
                self.capacity,
                self.tokens + elapsed * self.refill_rate
            )
            self.last_refill = now

            # 토큰 소비
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

# Redis 기반 분산 토큰 버킷
import redis

class RedisTokenBucket:
    def __init__(self, redis_client, key: str, capacity: int, refill_rate: float):
        self.redis = redis_client
        self.key = key
        self.capacity = capacity
        self.refill_rate = refill_rate

    def consume(self, tokens: int = 1) -> bool:
        now = time.time()
        pipe = self.redis.pipeline()

        lua_script = """
        local key = KEYS[1]
        local capacity = tonumber(ARGV[1])
        local rate = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])
        local requested = tonumber(ARGV[4])

        local last = tonumber(redis.call('HGET', key, 'last') or now)
        local tokens = tonumber(redis.call('HGET', key, 'tokens') or capacity)

        local elapsed = now - last
        tokens = math.min(capacity, tokens + elapsed * rate)

        if tokens >= requested then
            tokens = tokens - requested
            redis.call('HMSET', key, 'tokens', tokens, 'last', now)
            redis.call('EXPIRE', key, 3600)
            return 1
        end
        return 0
        """

        result = self.redis.eval(lua_script, 1, self.key,
                                  self.capacity, self.refill_rate, now, tokens)
        return bool(result)

def sliding_window_check(redis_client, user_id: str,
                          limit: int, window: int) -> bool:
    """슬라이딩 윈도우: 최근 window초 내 limit번 제한"""
    now = time.time()
    key = f"rate_limit:{user_id}"

    pipe = redis_client.pipeline()
    # 현재 요청 추가
    pipe.zadd(key, {str(now): now})
    # 윈도우 밖 데이터 삭제
    pipe.zremrangebyscore(key, '-inf', now - window)
    # 현재 카운트
    pipe.zcard(key)
    pipe.expire(key, window)
    results = pipe.execute()

    return results[2] <= limit

# FastAPI 미들웨어
from fastapi import Request, HTTPException

async def rate_limit_middleware(request: Request, call_next):
    user_id = request.headers.get('X-User-ID', request.client.host)

    if not sliding_window_check(redis_client, user_id, limit=100, window=60):
        raise HTTPException(status_code=429, detail="Too Many Requests")

    response = await call_next(request)
    response.headers['X-RateLimit-Remaining'] = str(get_remaining(user_id))
    return response