FastAPI 요청-응답 로깅하기

app = Flask() @app.before_first_request ... @app.before_request ... @app.after_request ... @app.teardown_request ... @app.teardown_appcontext ...

# Ref: https://stackoverflow.com/questions/69670125/how-to-log-raw-http-request-response-in-python-fastapi from json import JSONDecodeError import json import logging from typing import Callable, Awaitable, Tuple, Dict, List from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request from starlette.responses import Response, StreamingResponse from starlette.types import Scope, Message from fastapi import BackgroundTasks # Set up your custom logger here # Below is for example fix for what you want def log_handler(data: dict): # data handling ... class RequestWithBody(Request): """Creation of new request with body""" def __init__(self, scope: Scope, body: bytes) -> None: super().__init__(scope, self._receive) self._body = body self._body_returned = False async def _receive(self) -> Message: if self._body_returned: return {"type": "http.disconnect"} else: self._body_returned = True return {"type": "http.request", "body": self._body, "more_body": False} class CustomLoggingMiddleware(BaseHTTPMiddleware): """ Use of custom middleware since reading the request body and the response consumes the bytestream. Hence this approach to basically generate a new request/response when we read the attributes for logging. """ async def _get_response_params(self, response: StreamingResponse) -> Tuple[bytes, Dict[str, str], int]: """Getting the response parameters of a response and create a new response.""" response_byte_chunks: List[bytes] = [] response_status: List[int] = [] response_headers: List[Dict[str, str]] = [] async def send(message: Message) -> None: if message["type"] == "http.response.start": response_status.append(message["status"]) response_headers.append({k.decode("utf8"): v.decode("utf8") for k, v in message["headers"]}) else: response_byte_chunks.append(message["body"]) await response.stream_response(send) content = b"".join(response_byte_chunks) return content, response_headers[0], response_status[0] async def dispatch( # type: ignore self, request: Request, call_next: Callable[[Request], Awaitable[StreamingResponse]] ) -> Response: # Store request body in a variable and generate new request as it is consumed. request_body_bytes = await request.body() request_with_body = RequestWithBody(request.scope, request_body_bytes) # Store response body in a variable and generate new response as it is consumed. response = await call_next(request_with_body) response_content_bytes, response_headers, response_status = await self._get_response_params(response) # Logging # If there is no request body handle exception, otherwise convert bytes to JSON. try: req_body = json.loads(request_body_bytes) except JSONDecodeError: req_body = "" resp_body = json.loads(response.body.decode("utf8")) # Logging of relevant variables. log_data = { # added from auth middleware 'user_id': request.user.user_id, 'group_id': request.user.group_id, "authority": request.user.user_authority, # basic infomation "ip": request.client.host, "port": request.client.port, 'method': request.method, "path": request.url.path, # custom infomation "agent": dict(request.headers.items())["user-agent"], "request_body": req_body, "response_body": response_body, "response_status": response.status_code } # add BackgroudTask pre_background = response.background response.background = BackgroundTasks() if pre_background: response.background = BackgroundTasks([pre_background]) response.background.add_task(func=log_handler, data={ "req-resp": data }) # Finally, return the newly instantiated response values return Response(response_content_bytes, response_status, response_headers) # Register Middleware # app.py from fastapi import FastAPI # First way app = FastAPI(...) app.add_middleware(CustomLoggingMiddleware) # Second way from fastapi import FastAPI from fastapi.middleware import Middleware app = FastAPI( middleware=[Middleware(TimeHeaderMiddleware)] )

from fastapi import Request, BackgroundTasks def log_req_and_resp(func): @wraps(func) async def wrapper(request: Request, *args, **kwargs): resp = await func(request, *args, **kwargs) body_content = await request.body() # 바디 내용을 문자열로 변환 body_str = body_content.decode("utf-8") # JSON 문자열을 파이썬 딕셔너리로 변환 body_dict = json.loads(body_str) log_data = { # added from auth middleware 'user_id': request.user.user_id, 'group_id': request.user.group_id, "authority": request.user.user_authority, # basic infomation "ip": request.client.host, "port": request.client.port, 'method': request.method, "path": request.url.path, # custom infomation "agent": dict(request.headers.items())["user-agent"], # Below line raise Exception cause response has no status_code 'response_status': resp.status_code, } # Below 2 lines raise Exception cause resp is not starlette's Response resp.background = BackgroundTasks() resp.background.add_task(log_handler, log_data) # handle data log_handler(log_data) return resp # <= {"message": "OK"} return wrapper @app.get("/test") @log_req_and_resp async def test(): ... return {"message": "OK"}

# log_router import time from typing import Callable from fastapi import Request, Response, BackgroundTasks from fastapi.routing import APIRoute from dependency_injector.wiring import Provide, inject class LogRoute(APIRoute): def get_route_handler(self) -> Callable: original_route_handler = super().get_route_handler() async def custom_route_handler(request: Request) -> Response: response: Response = await original_route_handler(request) body_content = await request.body() req_body = json.loads(body_content.decode("utf-8")) resp_body = json.loads(response.body.decode("utf8")) log_data = { # added from auth middleware 'user_id': request.user.user_id, 'group_id': request.user.group_id, "authority": request.user.user_authority, # basic infomation "ip": request.client.host, "port": request.client.port, 'method': request.method, "path": request.url.path, "agent": dict(request.headers.items())["user-agent"], 'response_status': response.status_code, "req_body": req_body, "resp_body": resp_body } pre_background = response.background response.background = BackgroundTasks() if pre_background: response.background = BackgroundTasks([pre_background]) response.background.add_task(func=log_handler, data={ "req-resp": log_data }) return response return custom_route_handler # api.py some_router = APIRouter(route_class=LogRoute) # some_router로 decorated된 view는 이제 로깅이 됩니다. @some_router.get("/some") async def some_view(...): ... app.include_router(some_router)

# app.py app_ = FastAPI(...) @app_.exception_handler(Exception) async def base_exception_handler(request: Request, exc: Exception): """ 따로 명시 되지 않은 모든 에러를 잡아내는 핸들러 """ try: try: log_data = { "user_id": request.user.user_id, "group_id": request.user.group_id, "authority": request.user.user_authority, "ip": request.client.host if request.client else None, "port": request.client.port if request.client else None, "method": request.method, "path": request.url.path, "agent": dict(request.headers.items())["user-agent"], "response_status": HTTPStatus.INTERNAL_SERVER_ERROR, } # authentication middleware에서 Unhandled Error가 발생한 경우, # request.user 접근시 AssertionError가 발생합니다. except AssertionError: log_data = { "user_id": None, "group_id": None, "authority": None, "ip": request.client.host if request.client else None, "port": request.client.port if request.client else None, "method": request.method, "path": request.url.path, "agent": dict(request.headers.items())["user-agent"], "response_status": HTTPStatus.INTERNAL_SERVER_ERROR, } await log_handler(log_data) return JSONResponse( status_code=HTTPStatus.INTERNAL_SERVER_ERROR, content={ "code": "99999", "message": "RUNTIME_ERROR", }, ) except Exception as e: ...