FastAPI에서 사용자 정의 미들웨어를 빌드하여 API 제어 강화

소개

백엔드 개발의 세계에서 강력하고 확장 가능한 API를 구축하는 것은 종종 엔드포인트를 정의하는 것 이상을 포함합니다. 애플리케이션이 성장함에 따라 거의 모든 들어오는 요청 또는 나가는 응답에 대해 수행해야 하는 반복적인 작업에 직면하게 될 것입니다. 이러한 작업에는 인증, 로깅, 오류 처리, 사용자 정의 헤더 추가 또는 요청/응답 본문 수정 등이 포함될 수 있습니다. 이러한 기능은 개별 엔드포인트 핸들러에 흩어져 있을 수 있지만, 이 접근 방식은 빠르게 코드 중복, 유지 관리성 감소 및 관심사 분리의 명확성 부족으로 이어집니다. 이것이 바로 미들웨어라는 개념이 빛나는 곳입니다. 미들웨어는 요청이 핵심 애플리케이션 로직에 도달하기 훨씬 전이나 로직이 작업을 완료한 후에 전역 수준에서 요청 및 응답을 가로채고 처리할 수 있는 강력하고 우아하며 중앙 집중식 메커니즘을 제공합니다. 미들웨어를 활용함으로써 API의 제어, 유연성 및 아키텍처 명확성을 크게 향상시킬 수 있습니다. 이 글에서는 이러한 일반적인 문제를 해결하고 백엔드 개발 프로세스를 간소화하기 위해 두 가지 인기 있는 Python 웹 프레임워크인 FastAPI 및 Starlette에서 사용자 정의 미들웨어를 구축하는 실용적인 측면에 대해 자세히 살펴보겠습니다.

웹 프레임워크에서 미들웨어 이해하기

구현 세부 정보로 뛰어들기 전에 관련 핵심 개념에 대한 공통된 이해를 확립해 보겠습니다.

** 미들웨어란 무엇인가? ** 본질적으로 미들웨어는 요청과 애플리케이션의 핵심 로직 간, 또는 애플리케이션의 핵심 로직과 응답 간의 중개자 역할을 하는 소프트웨어 조각입니다. 이를 요청(및 해당 응답)이 통과하는 일련의 검문소 또는 필터로 생각하십시오. 각 미들웨어 조각은 특정 작업을 수행하고, 요청 또는 응답을 수정할 가능성이 있으며, 체인의 다음 계층으로 전달할 수 있습니다.

** Starlette 및 ASGI ** FastAPI는 경량 ASGI 프레임워크인 Starlette 위에 구축됩니다. ASGI(Asynchronous Server Gateway Interface)는 비동기 작업을 처리하도록 설계된 WSGI의 영적 계승자입니다. Starlette(및 확장하여 FastAPI)의 미들웨어는 본질적으로 비동기적입니다. 이는 미들웨어 함수가 애플리케이션의 호출을 처리할 때 async이며 await 작업을 수행한다는 것을 의미합니다.

** 주요 미들웨어 특성: **

• 가로채기: 미들웨어는 들어오는 요청과 나가는 응답 모두를 가로챌 수 있습니다.
• 실행 순서: 미들웨어가 정의된 순서는 중요합니다. 요청은 위에서 아래로 미들웨어 체인을 이동하고, 응답은 아래에서 위로 이동합니다.
• 수정: 미들웨어는 요청 헤더, 본문, 쿼리 매개변수와 응답 상태 코드, 헤더, 본문을 수정할 수 있습니다.
• 단축: 미들웨어는 예를 들어 핵심 애플리케이션 로직에 도달하기도 전에 오류 응답을 반환함으로써 요청-응답 주기를 조기에 종료할 수 있습니다.

사용자 정의 미들웨어 빌드

실용적인 예제를 통해 사용자 정의 미들웨어를 만드는 방법을 살펴보겠습니다. 일반적인 미들웨어 사용 사례의 다양한 유형을 다룰 것입니다.

요청 시간 로깅

각 요청이 처리되는 데 걸리는 시간을 기록하는 것은 성능 모니터링 및 디버깅에 매우 중요하며 매우 일반적인 사용 사례입니다.

# main.py
from fastapi import FastAPI, Request, Response
import time
import logging

# 로깅 구성
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    logger.info(f"Request: {request.method} {request.url.path} processed in {process_time:.4f}s")
    return response

@app.get("/")
async def read_root():
    return {"message": "Hello FastAPI!"}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # 일부 처리 시간 시뮬레이션
    await asyncio.sleep(0.1)
    return {"item_id": item_id}

이 예제에서는:

• add_process_time_header라는 async 함수를 정의합니다.
• request: Request와 call_next를 인수로 받습니다. call_next는 결국 라우트 핸들러로 이어지는 미들웨어 체인의 "다음" 작업을 나타내는 호출 가능 개체입니다.
• start_time을 기록하고, await call_next(request)를 사용하여 제어를 다음 미들웨어나 라우트 핸들러로 전달한 다음, 응답을 받은 후 process_time을 계산합니다.
• 마지막으로 응답에 사용자 정의 헤더 X-Process-Time을 추가하고 시간을 기록합니다.

이것을 main.py로 저장하고 uvicorn main:app --reload로 실행합니다. 그런 다음 요청을 합니다: curl -v http://localhost:8000/items/123. 응답에서 X-Process-Time 헤더를 볼 수 있습니다.

사용자 정의 인증 미들웨어

특정 X-API-Key 헤더가 존재하고 유효해야 하는 간단한 토큰 기반 인증 시스템이 있다고 상상해 봅시다.

# main_auth.py
from fastapi import FastAPI, Request, HTTPException, status
from starlette.responses import JSONResponse

app = FastAPI()

SECRET_API_KEY = "supersecretkey"

@app.middleware("http")
async def api_key_auth_middleware(request: Request, call_next):
    if request.url.path.startswith("/public"):
        # public 경로는 API 키 없이 액세스 허용
        response = await call_next(request)
        return response

    api_key = request.headers.get("X-API-Key")
    if not api_key or api_key != SECRET_API_KEY:
        return JSONResponse(
            status_code=status.HTTP_401_UNAUTHORIZED,
            content={"detail": "Unauthorized: Invalid or missing X-API-Key"}
        )
    
    response = await call_next(request)
    return response

@app.get("/protected")
async def protected_route():
    return {"message": "Welcome, authorized user!"}

@app.get("/public")
async def public_route():
    return {"message": "This is a public endpoint."}

여기서:

• X-API-Key 헤더를 확인합니다.
• 누락되거나 잘못된 경우 401 Unauthorized 상태 코드가있는 JSONResponse를 즉시 반환하여 요청을 단축하고 protected_route에 도달하는 것을 방지합니다.
• 또한 미들웨어가 조건부로 논리를 적용하는 방법을 보여주어 공개 경로가 인증 검사를 건너뛰도록 허용합니다.

다음으로 테스트하십시오:

• curl http://localhost:8000/protected (Unauthorized)
• curl -H "X-API-Key: wrongkey" http://localhost:8000/protected (Unauthorized)
• curl -H "X-API-Key: supersecretkey" http://localhost:8000/protected (Authorized)
• curl http://localhost:8000/public (API 키 없이 Authorized)

요청 또는 응답 본문 수정 (고급)

FastAPI/Starlette의 Request 객체는 본문에 액세스할 때 소비하므로 요청 본문을 수정하는 것은 까다로울 수 있습니다. 이를 수정하려면 본문을 읽고, 수정하고, 새 Request 객체(또는 모의 객체)를 생성한 다음, 전달해야 하는 경우가 많습니다. 응답 본문을 수정하는 것은 더 간단합니다.

타임스탬프를 추가하여 응답 본문을 수정하는 예를 들어 보겠습니다.

# main_transform.py
import json
from fastapi import FastAPI, Request, Response
from starlette.background import BackgroundTask

app = FastAPI()

@app.middleware("http")
async def add_timestamp_to_response(request: Request, call_next):
    response = await call_next(request)
    
    # JSON 응답만 수정
    if "application/json" in response.headers.get("content-type", ""):
        # 원시 응답 본문 읽기
        response_body = b""
        async for chunk in response.body_iterator:
            response_body += chunk
        
        # 디코딩, 수정 및 다시 인코딩
        try:
            data = json.loads(response_body)
            data["timestamp_utc"] = time.time()
            modified_body = json.dumps(data).encode("utf-8")
            
            # 수정된 본문으로 새 Response 객체 생성
            # 그리고 원본 상태 코드 및 헤더 복사
            return Response(
                content=modified_body,
                status_code=response.status_code,
                media_type="application/json",
                headers=dict(response.headers),
                background=response.background
            )
        except json.JSONDecodeError:
            # 유효한 JSON이 아닌 경우, 원본 응답을 반환
            pass
            
    return response

@app.get("/data")
async def get_data():
    return {"value": 123, "description": "some data"}

이 예제는 더 복잡합니다:

• 원본 응답 본문을 재구성하기 위해 response.body_iterator를 반복합니다. 본문은 스트림이며 한 번만 읽을 수 있기 때문에 이것이 중요합니다.
• 디코딩하고 timestamp_utc 필드를 추가한 다음 다시 인코딩합니다.
• 중요한 것은 수정된 내용으로 새 Response 객체를 반환하면서 원본 status_code, media_type, headers 및 background 작업을 보존한다는 것입니다. 이렇게 하면 원본 응답의 완전성이 유지되고 응답 본문 내용만 변경됩니다.

curl http://localhost:8000/data로 timestamp_utc 필드가 JSON 응답에 추가된 것을 볼 수 있습니다.

애플리케이션 시나리오

사용자 정의 미들웨어는 수많은 시나리오에 적용할 수 있습니다:

• API 키 / 토큰 유효성 검사: 위에서 보듯이 인증 및 권한 부여를 위해.
• 요청/응답 로깅: 디버깅, 감사 및 성능 모니터링을 위해.
• 사용자 정의 헤더 주입: 상관 ID, 추적 정보 또는 보안 헤더 추가.
• 오류 처리: 특정 예외를 전역적으로Catch 하여 일관된 오류 응답 반환.
• 속도 제한: 특정 IP 또는 사용자로부터의 요청 수를 제한하여 남용 방지.
• CORS 관리: FastAPI는 내장 CORS 미들웨어를 제공하지만, 필요한 경우 더 세분화된 사용자 정의 논리를 구축할 수 있습니다.
• 데이터 변환: 들어오는 요청 데이터를 정규화하거나 나가는 응답 데이터를 보강합니다.

결론

FastAPI 및 Starlette의 사용자 정의 미들웨어는 웹 애플리케이션에서 교차 관심사를 관리하는 강력하고 유연한 접근 방식을 제공합니다. 인증, 로깅 및 데이터 조작과 같은 일반적인 기능을 중앙 집중화함으로써 코드 중복을 크게 줄이고 유지 관리성을 개선하며 요청-응답 수명 주기에 대한 세분화된 제어를 얻을 수 있습니다. 이 패턴을 활용하면 더 강력하고 확장 가능하며 아키텍처적으로 깔끔한 API를 구축할 수 있습니다.