# main_in_memory.py
from fastapi import FastAPI, Request, HTTPException, status
from collections import defaultdict
import time
import asyncio

app = FastAPI()

# { "client_ip": { "timestamp": [request_times...], "count": N } }
# 또는 더 간단하게: { "client_ip": { "last_reset_time": float, "count": int } }
# 여기서는 더 간단한 고정 윈도우 카운터 방식을 사용합니다.
rate_limit_data = defaultdict(lambda: {"last_reset_time": 0.0, "count": 0})
RATE_LIMIT_WINDOW_SECONDS = 60  # 60초
RATE_LIMIT_MAX_REQUESTS = 5     # 60초당 최대 5회 요청

# 동시성 문제를 방지하기 위한 락 (FastAPI는 기본적으로 단일 스레드이지만, 비동기 작업 시 필요할 수 있음)
rate_limit_lock = asyncio.Lock()

async def get_client_ip(request: Request) -> str:
    """클라이언트 IP 주소를 가져옵니다."""
    # X-Forwarded-For 헤더를 우선적으로 확인하여 프록시 뒤에 있을 경우 실제 클라이언트 IP를 얻습니다.
    # 하지만 이 헤더는 조작될 수 있으므로, 보안이 중요한 서비스에서는 추가적인 검증이 필요합니다.
    return request.headers.get("X-Forwarded-For", request.client.host)

@app.get("/")
async def read_root(request: Request):
    client_ip = await get_client_ip(request)
    current_time = time.time()

    async with rate_limit_lock:
        client_data = rate_limit_data[client_ip]

        # 윈도우가 재설정될 시간인지 확인
        if current_time - client_data["last_reset_time"] > RATE_LIMIT_WINDOW_SECONDS:
            client_data["last_reset_time"] = current_time
            client_data["count"] = 1
        else:
            client_data["count"] += 1

        if client_data["count"] > RATE_LIMIT_MAX_REQUESTS:
            remaining_time = int(RATE_LIMIT_WINDOW_SECONDS - (current_time - client_data["last_reset_time"]))
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail=f"Too many requests. Try again in {remaining_time} seconds."
            )
    
    # Rate Limit 관련 헤더 추가 (선택 사항이지만 권장)
    response_headers = {
        "X-RateLimit-Limit": str(RATE_LIMIT_MAX_REQUESTS),
        "X-RateLimit-Remaining": str(RATE_LIMIT_MAX_REQUESTS - client_data["count"]),
        "X-RateLimit-Reset": str(int(client_data["last_reset_time"] + RATE_LIMIT_WINDOW_SECONDS))
    }
    
    return {"message": f"Hello, {client_ip}! Request successful.", "headers": response_headers}

@app.get("/items/{item_id}")
async def read_item(item_id: int, request: Request):
    client_ip = await get_client_ip(request)
    current_time = time.time()

    async with rate_limit_lock:
        client_data = rate_limit_data[client_ip]

        if current_time - client_data["last_reset_time"] > RATE_LIMIT_WINDOW_SECONDS:
            client_data["last_reset_time"] = current_time
            client_data["count"] = 1
        else:
            client_data["count"] += 1

        if client_data["count"] > RATE_LIMIT_MAX_REQUESTS:
            remaining_time = int(RATE_LIMIT_WINDOW_SECONDS - (current_time - client_data["last_reset_time"]))
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail=f"Too many requests. Try again in {remaining_time} seconds."
            )
            
    response_headers = {
        "X-RateLimit-Limit": str(RATE_LIMIT_MAX_REQUESTS),
        "X-RateLimit-Remaining": str(RATE_LIMIT_MAX_REQUESTS - client_data["count"]),
        "X-RateLimit-Reset": str(int(client_data["last_reset_time"] + RATE_LIMIT_WINDOW_SECONDS))
    }

    return {"item_id": item_id, "message": f"Item data for {client_ip}", "headers": response_headers}

docker run --name my-redis -p 6379:6379 -d redis

pip install redis

# main_redis.py
from fastapi import FastAPI, Request, HTTPException, Depends, status
import redis.asyncio as redis
import time
from typing import Optional

app = FastAPI()

# Redis 클라이언트 초기화
# 실제 환경에서는 환경 변수 등을 통해 Redis URL을 관리하는 것이 좋습니다.
REDIS_URL = "redis://localhost:6379/0"
redis_client: Optional[redis.Redis] = None

# Rate Limiting 설정 (초당 요청 수)
RATE_LIMIT_WINDOW_SECONDS = 60  # 60초
RATE_LIMIT_MAX_REQUESTS = 5     # 60초당 최대 5회 요청

@app.on_event("startup")
async def startup_event():
    global redis_client
    redis_client = redis.from_url(REDIS_URL, encoding="utf-8", decode_responses=True)
    print("Redis client connected.")

@app.on_event("shutdown")
async def shutdown_event():
    if redis_client:
        await redis_client.close()
        print("Redis client closed.")

async def get_client_ip(request: Request) -> str:
    """클라이언트 IP 주소를 가져옵니다."""
    # X-Forwarded-For 헤더를 우선적으로 확인하여 프록시 뒤에 있을 경우 실제 클라이언트 IP를 얻습니다.
    # 실제 서비스에서는 이 헤더가 조작될 수 있으므로, 로드 밸런서/프록시 설정에 따라 신뢰할 수 있는 IP를 얻는 로직이 필요합니다.
    # 예: request.headers.get("X-Real-IP") 또는 특정 IP 범위만 신뢰
    return request.headers.get("X-Forwarded-For", request.client.host)

async def rate_limit_dependency(
    request: Request,
    limit: int = RATE_LIMIT_MAX_REQUESTS,
    window: int = RATE_LIMIT_WINDOW_SECONDS
):
    """
    FastAPI 의존성 주입으로 Rate Limiting을 적용합니다.
    고정 윈도우 카운터 알고리즘을 Redis로 구현합니다.
    """
    if not redis_client:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Redis client not initialized.")

    client_id = await get_client_ip(request)
    # Redis 키는 "rate_limit:{client_id}:{window_start_timestamp}" 형태로 생성
    # 고정 윈도우이므로, 현재 시간을 윈도우 크기로 나눈 몫을 사용하여 윈도우 시작 시간을 계산합니다.
    current_window_start = int(time.time() / window) * window
    key = f"rate_limit:{client_id}:{current_window_start}"

    # Redis의 INCR 명령은 원자적으로 카운터를 증가시키고 현재 값을 반환합니다.
    # EXPIRE 명령은 키의 만료 시간을 설정합니다.
    # pipeline을 사용하여 두 명령을 하나의 트랜잭션처럼 실행하여 원자성을 보장합니다.
    pipe = redis_client.pipeline()
    pipe.incr(key)
    pipe.expire(key, window) # 윈도우가 끝나면 자동으로 키가 삭제됩니다.
    
    current_count, _ = await pipe.execute()

    if current_count > limit:
        # 남은 시간 계산: 다음 윈도우 시작 시간 - 현재 시간
        next_window_start = current_window_start + window
        remaining_time = int(next_window_start - time.time())
        
        # Rate Limit 관련 헤더 추가
        request.state.rate_limit_headers = {
            "X-RateLimit-Limit": str(limit),
            "X-RateLimit-Remaining": "0",
            "X-RateLimit-Reset": str(next_window_start)
        }
        
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail=f"Too many requests. Try again in {remaining_time} seconds."
        )
    
    # Rate Limit 관련 헤더를 request.state에 저장하여 응답에 추가할 수 있도록 합니다.
    request.state.rate_limit_headers = {
        "X-RateLimit-Limit": str(limit),
        "X-RateLimit-Remaining": str(limit - current_count),
        "X-RateLimit-Reset": str(current_window_start + window)
    }

    return True # 제한에 걸리지 않았다면 True 반환

@app.middleware("http")
async def add_rate_limit_headers(request: Request, call_next):
    """
    미들웨어를 사용하여 Rate Limiting 관련 HTTP 응답 헤더를 추가합니다.
    """
    response = await call_next(request)
    if hasattr(request.state, "rate_limit_headers"):
        for header, value in request.state.rate_limit_headers.items():
            response.headers[header] = value
    return response

@app.get("/", dependencies=[Depends(rate_limit_dependency)])
async def read_root():
    return {"message": "Hello, world! This is a rate-limited endpoint."}

@app.get("/protected", dependencies=[Depends(rate_limit_dependency)])
async def protected_endpoint():
    return {"message": "This is a protected endpoint with default rate limits."}

@app.get("/custom-limit", dependencies=[Depends(rate_limit_dependency(limit=2, window=10))])
async def custom_limit_endpoint():
    """
    특정 엔드포인트에 대해 맞춤형 Rate Limiting을 적용합니다.
    10초당 2회 요청으로 제한됩니다.
    """
    return {"message": "This endpoint has a custom rate limit (2 requests per 10 seconds)."}

# Rate Limiting이 적용되지 않는 엔드포인트
@app.get("/no-limit")
async def no_limit_endpoint():
    return {"message": "This endpoint has no rate limit."}

pip install fastapi-limiter[redis]

# main_fastapi_limiter.py
from fastapi import FastAPI, Request, Depends
from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter
import redis.asyncio as redis

app = FastAPI()

REDIS_URL = "redis://localhost:6379/0"

@app.on_event("startup")
async def startup():
    redis_connection = redis.from_url(REDIS_URL, encoding="utf-8", decode_responses=True)
    await FastAPILimiter.init(redis_connection)

@app.get("/", dependencies=[Depends(RateLimiter(times=5, seconds=60))])
async def read_root(request: Request):
    return {"message": "Hello, world! This is a rate-limited endpoint (5 req/min)."}

@app.get("/protected", dependencies=[Depends(RateLimiter(times=2, seconds=10))])
async def protected_endpoint(request: Request):
    return {"message": "This endpoint has a custom rate limit (2 req/10s)."}

@app.get("/no-limit")
async def no_limit_endpoint():
    return {"message": "This endpoint has no rate limit."}