Search

FastAPI 요청-응답 로깅(w/ AOP)

Intro

웹 서비스를 안정적으로 운영하기 위해서는 어플리케이션에서 발생하는 로그를 잘 기록하는 것이 중요하다.
로그를 활용하면 어플리케이션에서 발생하는 다양한 문제를 파악하기 용이하고, 서비스를 지속적으로 모니터링 할 수 있다.
다만 비지니스 로직과 무관한 로그를 컨트롤러(controller) 안에서 처리하는 것은 올바르지 않다.
비지니스 로직과 무관한 로그 코드로 인해 전체적인 코드 가독성을 해칠뿐 아니라 불필요한 코드 중복이 발생한다.
또 유연하지 못한 구조로 인해 변경에 매우 취약하다.
만약 n개의 컨트롤러에서 각각 로그를 발생시키고 있을 때, 로그의 포맷을 일괄적으로 변경한다고 하자.
모든 로그의 포맷을 변경하기 위해 n개의 컨트롤러 코드를 일일이 수정해야 한다.

AOP(aspect-oriented programming)

관점 지향 프로그래밍을 적용하면 위와 같은 문제를 해결할 수 있다.
AOP는 JAVA Spring 생태계에는 흔히 알려져 있지만 파이썬 개발자들에게는 다소 생소한 개념일 수 있다.
AOP의 핵심은 시스템 전반에 넓게 펴져있지만 핵심 비지니스 로직과 무관한 횡단 관심사를 모듈화하는 것이다.
위에 설명한 로그와 같은 경우가 대표적인 예시입니다.

FastAPI 요청-응답 로깅

웹 어플리케이션을 통해 처리되는 HTTP 요청과 응답을 로깅한다고 하자.
일반적으로 AOP를 적용하기 위해 FastAPI 미들웨어에서 로깅을 처리하고자 할 것이다.
미들웨어는 모든 요청과 응답 사이에서 동작하기 때문에 매우 합리적인 선택이 될 수 있다.
그런데 미들웨어에서 처리할 경우 강제적으로 모든 요청과 응답에 대해 일관적인 로깅을 적용할 수 밖에 없다.
로깅이 필요하지 않은 특정 API을 제외시키기 번거롭고, 일부만 다른 로깅 포맷을 적용하기도 어렵다.
이런 경우 APIRoute 클래스를 오버라이드하여 해결할 수 있다.
import logging from typing import Any, Callable, Dict from fastapi.routing import APIRoute from starlette.requests import Request from starlette.responses import Response logger = logging.getLogger(__name__) class LoggingAPIRoute(APIRoute): def get_route_handler(self) -> Callable: original_route_handler = super().get_route_handler() async def custom_route_handler(request: Request) -> Response: await self._request_log(request) response: Response = await original_route_handler(request) self._response_log(request, response) return response return custom_route_handler @staticmethod def _has_json_body(request: Request) -> bool: if ( request.method in ("POST", "PUT", "PATCH") and request.headers.get("content-type") == "application/json" ): return True return False async def _request_log(self, request: Request) -> None: extra: Dict[str, Any] = { "httpMethod": request.method, "url": request.url.path, "headers": request.headers, "queryParams": request.query_params, } if self._has_json_body(request): request_body = await request.body() extra["body"] = request_body.decode("UTF-8") logger.info("request", extra=extra) @staticmethod def _response_log(request: Request, response: Response) -> None: extra: Dict[str, str] = { "httpMethod": request.method, "url": request.url.path, "body": response.body.decode("UTF-8") } logger.info("response", extra=extra)
Python
복사
위와 같이 로깅에 사용할 메소드를 정의한 뒤 original_route_handler() 메소드 전후에서 호출하면 된다.
예시 코드는 _has_json_body()라는 메소드를 정의하여 HTTP method가 POST, PUT, PATCH면서 content-typeapplication/json인 경우에 대해서만 요청의 body를 로깅하도록 구현했다.
정의한 APIRoute 클래스는 다음과 같이 라우터에 연결하여 특정 API에 대해서만 동작하도록 할 수 있다.
router = APIRouter(route_class=LoggingAPIRoute)
Python
복사
만약 다른 로그 포맷이 필요한 경우, 새로운 APIRoute 클래스를 정의하여 사용하면 된다.