FASTAPI 로 구현해보는 Vertical Slice Architecture - 3 (File Log 추가)

기본적으로 어떤 app 을 만들던 운영중에 문제점을 파악하기 위해 또는 모니터링을 위해 logging 을 하게 된다.

python 도 기본적인 log system 이 있다. 

 

features-->hello 로 이동해서 endpoints.py 를 수정하자

endpoints.py

...
import logging

# 로그 생성
logger = logging.getLogger("main")

# 로그의 출력 기준 설정
logger.setLevel(logging.INFO)

# log 출력 형식
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# console handler
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
                  
# log를 파일에 출력
file_handler = logging.FileHandler('my.log')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)


@router.get("/", tags=["hello"])
async def read_root():
    mediator = Mediator()
    request = GetHello()
    logger.info("get read_root")
    return await mediator.send_async(request)


@router.post("/", tags=["hello"])
async def read_root():
    mediator = Mediator()
    request = PostHello()
    logger.info("post read_root")
    return await mediator.send_async(request)

위와 같이 log 를 선언하고 get 과 post 에서 logger.info 를 통해 logging 을 하고 있다. 

이런형태로 file 에 logging 을 남길 수 있다. 

 

이제 logger 생성을 위한 여러 정보들을 log.ini 쪽으로 이동해서 작업해 보자

[loggers]
keys=root

[handlers]
keys=consoleHandler,fileHandler

[formatters]
keys=fileFormatter,consoleFormatter

[logger_root]
level=INFO
handlers=consoleHandler,fileHandler

[handler_consoleHandler]
class=StreamHandler
level=INFO
formatter=consoleFormatter
args=(sys.stdout,)

[handler_fileHandler]
class=FileHandler
level=INFO
formatter=fileFormatter
args=('my.log',)

[formatter_fileFormatter]
format=%(asctime)s - %(name)s - %(levelname)s - %(message)s
datefmt=

[formatter_consoleFormatter]
format=%(levelname)s - %(message)s
datefmt=

이전에 endpoints.py 에서 logging 설정 작업을 했었는데 이부분을 삭제하자.

그리고 main.py 에서 log.ini 를 사용하여 logging 하도록 수정해 보자

 

endpoints.py

from fastapi import APIRouter
from mediatr import Mediator

from .get import GetHello
from .post import PostHello
import logging

router = APIRouter()

# 로그 생성
logger = logging.getLogger()

@router.get("/", tags=["hello"])
async def read_root():
    mediator = Mediator()
    request = GetHello()
    logger.info("get read_root")
    return await mediator.send_async(request)


@router.post("/", tags=["hello"])
async def read_root():
    mediator = Mediator()
    request = PostHello()
    logger.info("post read_root")
    return await mediator.send_async(request)

main.py

from fastapi import FastAPI
from .features.hello import endpoints
from prisma import Prisma
import logging
app = FastAPI(title="FastAPI, vertical slice architecture")

# 로그 생성
logging.config.fileConfig("log.ini")
logger = logging.getLogger()

app.include_router(endpoints.router)
db = Prisma()


@app.on_event("startup")
async def startup_event():
    await db.connect()
    logger.info("start")
    print("Startup event")


@app.on_event("shutdown")
async def shutdown_event():
    await db.disconnect()
    print("Shutdown event")

이제 실행해 보자

uvicorn src.main:app --reload

my.log 가 생성될 것이다. 

내용을 보면 우리가 작성한 log 가 logging 된것을 볼 수 있다. 

 

이제 request 와 response 를 기록하는 방법을 알아보자

middleware 를 통해서 처리하는 방법이 있으나 issue 가 있는 걸로 알고 있다. 

그래서 api route 에서 처리 하는 방식을 활용한다. 

 

apiroute.py

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()


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:
        message: 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()
            message["body"] = request_body.decode("UTF-8")

        logger.info(f"request {message}")

    @staticmethod
    def _response_log(request: Request, response: Response) -> None:
        message: Dict[str, str] = {
	        "httpMethod": request.method,
	        "url": request.url.path,
	        "body": response.body.decode("UTF-8")
				}
		
        logger.info(f"response {message}")

 

그리고 endpoints.py 를 다음과 같이 수정한다.

endpoints.py

from fastapi import APIRouter
from mediatr import Mediator

from src.shared.api_logger.apiroute import LoggingAPIRoute

from .get import GetHello
from .post import PostHello
import logging

router = APIRouter(route_class=LoggingAPIRoute)

# 로그 생성
logger = logging.getLogger()

@router.get("/", tags=["hello"])
async def read_root():
    mediator = Mediator()
    request = GetHello()
    logger.info("get read_root")
    return await mediator.send_async(request)


@router.post("/", tags=["hello"])
async def read_root():
    mediator = Mediator()
    request = PostHello()
    logger.info("post read_root")
    return await mediator.send_async(request)

 

 

 

관련영상

https://youtu.be/qY8MdbXIMWU