본문으로 건너뛰기
Advertisement

미들웨어와 CORS

미들웨어는 모든 요청·응답을 가로채 로깅, 인증, 압축, CORS 처리를 수행합니다.

기본 미들웨어​

import time
import uuid
from fastapi import FastAPI, Request, Response

app = FastAPI()


# @app.middleware("http"): 모든 HTTP 요청·응답 처리
@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    # 요청 처리 전
    request_id = str(uuid.uuid4())[:8]
    start_time = time.perf_counter()

    print(f"[{request_id}] → {request.method} {request.url.path}")

    # 다음 미들웨어 또는 라우트 핸들러 호출
    response: Response = await call_next(request)

    # 응답 처리 후
    elapsed = (time.perf_counter() - start_time) * 1000
    print(f"[{request_id}] ← {response.status_code} ({elapsed:.1f}ms)")

    # 응답 헤더 추가
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Process-Time"] = f"{elapsed:.1f}ms"

    return response


@app.get("/items/{item_id}")
def get_item(item_id: int):
    return {"item_id": item_id}

CORS 설정​

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# CORS: 브라우저의 크로스 오리진 요청 허용
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",    # 개발 환경
        "https://myapp.com",        # 프로덕션
    ],
    allow_credentials=True,         # 쿠키 허용
    allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
    allow_headers=["*"],            # 모든 헤더 허용
)


# 개발 환경: 모든 오리진 허용 (프로덕션에서 사용 금지)
def create_dev_app():
    dev_app = FastAPI()
    dev_app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_methods=["*"],
        allow_headers=["*"],
    )
    return dev_app


@app.get("/api/data")
def get_data():
    return {"data": "cross-origin accessible"}

GZip과 TrustedHost 미들웨어​

from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()

# GZip: 응답 압축 (minimum_size 이상일 때)
app.add_middleware(GZipMiddleware, minimum_size=1000)

# TrustedHost: 허용된 호스트만 접근
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com", "localhost"],
)


@app.get("/large-data")
def large_data():
    return {"data": "x" * 5000}  # GZip 압축됨

Starlette BaseHTTPMiddleware​

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response, JSONResponse

app = FastAPI()


class RateLimitMiddleware(BaseHTTPMiddleware):
    """요청 빈도 제한 (시뮬레이션)"""

    def __init__(self, app, max_requests: int = 100):
        super().__init__(app)
        self.max_requests = max_requests
        self._request_counts: dict[str, int] = {}

    async def dispatch(self, request: Request, call_next) -> Response:
        client_ip = request.client.host if request.client else "unknown"
        count = self._request_counts.get(client_ip, 0) + 1
        self._request_counts[client_ip] = count

        if count > self.max_requests:
            return JSONResponse(
                status_code=429,
                content={"detail": "Too Many Requests"},
            )

        response = await call_next(request)
        response.headers["X-RateLimit-Remaining"] = str(self.max_requests - count)
        return response


class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """보안 헤더 추가"""

    async def dispatch(self, request: Request, call_next) -> Response:
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        return response


app.add_middleware(RateLimitMiddleware, max_requests=1000)
app.add_middleware(SecurityHeadersMiddleware)


@app.get("/")
def root():
    return {"message": "Hello"}

미들웨어 실행 순서​

from fastapi import FastAPI, Request
import time

app = FastAPI()


@app.middleware("http")
async def middleware_a(request: Request, call_next):
    print("A: 요청 전")
    response = await call_next(request)
    print("A: 응답 후")
    return response


@app.middleware("http")
async def middleware_b(request: Request, call_next):
    print("B: 요청 전")
    response = await call_next(request)
    print("B: 응답 후")
    return response


# 실행 순서 (LIFO — 마지막 등록이 먼저 실행):
# B: 요청 전
# A: 요청 전
# [라우트 핸들러]
# A: 응답 후
# B: 응답 후


@app.get("/")
def root():
    print("라우트 핸들러")
    return {}

정리​

미들웨어용도
@app.middleware("http")커스텀 HTTP 미들웨어
CORSMiddleware크로스 오리진 요청 허용
GZipMiddleware응답 압축
TrustedHostMiddleware호스트 화이트리스트
BaseHTTPMiddlewareStarlette 기반 커스텀 미들웨어

미들웨어는 모든 요청에 공통 처리가 필요할 때 사용합니다. 요청별 처리는 의존성 주입을 선호하세요.

Advertisement
Advertisement