💡 Option 1 - Middleware 사용하기
📌 Middleware란?
A "middleware" is a function that works with every request before it is processed by any specific path operation. And also with every response before returning it.
즉, 미들웨어(middleware)란 모든 요청(request)과 응답(response)가 오고 가기 전, 목적에 맞는 처리를 할 수 있는 함수들이라고 할 수 있다.
middleware는 요청(request)이 특정 엔드포인트에 의해 처리되기 전에, 또 응답(response)이 클라이언트에게 return됙 전에 handle할 수 있다.
middleware을 만들기 위해서는 @app.middleware("http") 데코레이터를 사용하면 된다.
📌 request와 response body 접근하기
As you need to consume the request body from the stream inside the middleware—using either
request.body()orrequest.stream(), as shown in this answer (behind the scenes, the former method actually calls the latter, see here)—then it won't be available when you later pass the request to the corresponding endpoint.
여기처럼 set_body 함수를 사용하여 request body를 접근할 수 있게 해야 한다.
response body의 경우, 여기 방법처럼 하면 된다.
response body를 consume한 후 response를 클라이언트에게 return 하면 된다.
📌 로깅하기
BackgroundTask를 사용하면 된다. (관련 질문1, 관련 질문2)
BackgroundTask는 response가 전송될 경우 딱 한 번 동작한다. 따라서 클라이언트는 response를 받기 전, logging이 완료될 때까지 기다리지 않아도 된다.(response time에 큰 영향을 끼치지 않게 됨)
from fastapi import FastAPI, APIRouter, Response, Request
from starlette.background import BackgroundTask
from fastapi.routing import APIRoute
from starlette.types import Message
from typing import Dict, Any
import logging
app = FastAPI()
logger = logging.getLogger("main")
logging.basicConfig(level=logging.DEBUG)
steam_handler = logging.FileHandler('info.log', mode='w')
logger.addHandler(steam_handler)
def log_info(req_body, res_body):
logging.info(req_body)
logging.info(res_body)
async def set_body(request: Request, body: bytes):
async def receive() -> Message:
return {'type': 'http.request', 'body': body}
request._receive = receive
@app.middleware('http')
async def some_middleware(request: Request, call_next):
req_body = await request.body()
await set_body(request, req_body)
response = await call_next(request)
res_body = b''
async for chunk in response.body_iterator:
res_body += chunk
task = BackgroundTask(log_info, req_body, res_body)
return Response(content=res_body, status_code=response.status_code,
headers=dict(response.headers), media_type=response.media_type, background=task)
@app.post('/')
def main(payload: Dict[Any, Any]):
return payloadpython logging에 대한 지식이 부족했다!! 그래서 네이버 부스트캠프 강의와 구글링을 통해 도움을 얻었다.
-
보통 자신만의 특정한
logger를 따로 만들어서 사용한다. -
logger의 레벨은DEBUG < INFO < WARNING < ERROR < CRITICAL이다.
Python 로깅의 기본 설정은WARNING이고, 하위레벨인INFO와DEBUG은 출력이 안 된다. 이처럼 필요에 따라 레벨을 설정할 수 있다. -
출력 포매팅을 설정 할 수 있다.
asctime: 시간name: 로거이름levelname: 로깅레벨message: 메세지
- 핸들러(Handler)란 로깅한 정보가 출력되는 위치를 설정하는 것을 말한다. 콘솔, 파일, DB, 소켓, 큐 등을 통해 출력하도록 설정할 수 있다. 나는 파일에 출력하도록 설정하였다.
import logging
# 1. logger 생성
logger = logging.getLogger("main")
# 2. logger 레벨 설정
logger.setLevel(logging.DEBUG)
# 또는
logging.basicConfig(level=logging.DEBUG)
# 3. formatting 설정
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 4. handler 설정
# console 출력
stream_hander = logging.StreamHandler()
stream_hander.setFormatter(formatter)
logger.addHandler(stream_hander)
# 파일 출력
file_handler = logging.FileHandler('info.log', mode='w')
logger.addHandler(file_handler)API 서버를 실행하고 요청을 보내면 정상적인 응답과 함께 이렇게 로깅되는 것을 확인할 수 있다.
참고
