[FastAPI] 요청 단위 로깅 구현하기 - Middleware, ContextVar, Decorator

 

 

 

문제상황: 요청 단위 로깅 및 쓰레드 로컬

이번에 로깅 시스템을 개선하는 작업을 진행했습니다. 원래는 아주 단순한 수준의 로깅만 되어 있었는데, 실제로 장애나 예외 상황이 발생했을 때 특정 요청의 흐름을 추적하기가 너무 어려웠습니다. 예를 들어, 같은 시점에 여러 요청이 들어오면 로그가 섞여버리기 때문에 "이 로그가 어떤 요청에서 나온 건지" 확인하려면 로그 타임스탬프나 유추에 의존해야 했습니다. 그래서 이번에 요청 단위로 UUID 같은 고유한 ID를 부여하고, 그걸 모든 로그에 자동으로 포함시키는 방식을 적용했습니다.

 

자바/스프링에서는 AOP와 ThreadLocal을 지원하기 때문에 이러한 로깅을 손쉽게 적용할 수 있습니다. 요청이 들어올 때마다 ThreadLocal에 request ID를 저장하고, 로깅 시 해당 값을 꺼내서 함께 출력할 수 있습니다. 이번에 FastAPI에서 로깅을 적용할 때 위와 같은 컨셉을 어떻게 적용할 수 있는지 알아보았습니다.

요약

역할/개념	Spring	FastAPI
요청 전후 처리	Interceptor	Middleware
요청 단위 데이터 저장	ThreadLocal	ContextVar
서비스 함수 단위 공통 로직 (단면 관심사)	AOP	Decorator

 

해결방안 (1): ContextVar로 비동기 컨텍스트 관리

FastAPI는 ASGI 기반 비동기 서버이므로, ThreadLocal을 그대로 사용할 수 없습니다. 한 요청이 여러 코루틴으로 나뉘어 실행될 수 있기 때문에, ThreadLocal에 저장하면 요청 간 데이터가 섞일 위험이 있습니다. 대신 ContextVar를 사용하면, 코루틴 간에도 안전하게 요청 단위 데이터를 유지할 수 있습니다.

 

from contextvars import ContextVar

_request_id: ContextVar[str] = ContextVar("request_id", default="")

class RequestContext:
    @staticmethod
    def set_request_id(request_id: str) -> None:
        _request_id.set(request_id)

    @staticmethod
    def get_request_id() -> Optional[str]:
        return _request_id.get()

 

해결방안 (2): Middleware로 요청 단위 감싸기

FastAPI에서는 미들웨어를 사용하여 요청(Request)과 응답(Response) 전후에 작업을 추가할 수 있습니다. 이때 요청이 들어올 때 UUID를 생성하고 ContextVar에 저장하면, 요청 단위 로깅이 가능해집니다. 추가로 시간을 측정하여 전체 요청이 처리되는데 소요되는 시간을 확인할 수 있습니다.

 

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

from request_context import RequestContext  # RequestContext가 정의된 모듈

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

@app.middleware("http")
async def request_logging_middleware(request: Request, call_next):
    # 1. 요청 UUID 생성 및 RequestContext에 저장
    request_id = str(uuid.uuid4())
    RequestContext.set_request_id(request_id)

    # 2. 요청 처리 시작 시간 기록
    start_time = time.time()

    # 3. 요청 처리
    response = await call_next(request)

    # 4. 요청 처리 종료 시간 기록 및 소요 시간 계산
    elapsed_time = time.time() - start_time

    # 5. 로그 출력
    logger.info(f"[{RequestContext.get_request_id()}] {request.method} {request.url.path} completed in {elapsed_time:.3f}s")

    return response

 

해결방안 (3): 데코레이터를 활용한 AOP 구현

미들웨어는 Spring에서 Interceptor 처럼 요청 전체의 시작과 끝만 감쌀 수 있기 때문에, Spring의 AOP처럼 특정 서비스 함수 단위의 로깅이나 공통 로직을 적용하려면 데코레이터를 사용할 수 있습니다.

 

import functools
import time
import logging
from request_context import RequestContext  # RequestContext 모듈 import

logger = logging.getLogger(__name__)

def log_execution(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        request_id = RequestContext.get_request_id()  # ContextVar 접근 캡슐화
        start = time.time()
        logger.info(f"[{request_id}] Start {func.__name__}")
        
        result = await func(*args, **kwargs)
        
        elapsed = time.time() - start
        logger.info(f"[{request_id}] End {func.__name__} ({elapsed:.3f}s)")
        return result
    return wrapper

# 사용 예시
@log_execution
async def process_payment(user_id: str, amount: int):
    # 서비스 로직
    ...

 

결론

이번 작업을 통해 Spring에서의 ThreadLocal + Interceptor + AOP 패턴을 FastAPI 환경에 맞게 구현해보았습니다.

• ContextVar: 요청 단위 컨텍스트 관리
  • FastAPI에서는 ContextVar를 사용하여 비동기 코루틴 환경에서도 요청 단위 데이터를 안전하게 유지할 수 있습니다.
  • 추가로 파이썬에서는 threading.local()을 통해 ThreadLocal 저장소 또한 지원합니다.
• Middleware: 요청 전체 단위 로깅
  • 미들웨어를 활용하면 요청 시작과 종료 시점에 UUID를 부여하고 처리 시간을 기록할 수 있습니다.
  • 로그가 요청 단위로 구분되므로 장애 분석과 디버깅이 훨씬 수월해집니다.
• Decorator: 서비스 함수 단위 공통 로직 적용
  • AOP처럼 특정 서비스 함수 단위 로직을 적용하고 싶다면, Python의 데코레이터를 활용할 수 있습니다.
  • 미들웨어와 함께 사용하면 요청 단위 + 함수 단위 로깅을 모두 커버할 수 있습니다.