Python structlog

 

Standard Library Logging

이전 글에서 작성했던 파이썬 내장 라이브러리 logging으로 구현했던 것을 structlog로 구현할 수 있다.

Python logging

추가적으로 미들웨어를 조작해서 trace_id 까지 로그에 남기도록 할 수 있다.

• logging.py
  

  import logging.config
  
  import structlog
  from typing import Any
  
  from config import settings
  
  
  Logger = structlog.stdlib.BoundLogger
  
  
  class Logging:
      # utc=False 설정을 해야 실행 중인 서버 시각이 찍힌다.
      timestamper = structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False)
  
      @classmethod
      def get_level(cls) -> str:
          return settings.DEBUG # 환경에 맞게 분리할 수도 있다.
  
      @classmethod
      def get_processors(cls) -> list[Any]:
          return [
              structlog.contextvars.merge_contextvars,
              structlog.stdlib.add_log_level,
              structlog.stdlib.add_logger_name,
              structlog.stdlib.PositionalArgumentsFormatter(),
              cls.timestamper,
              structlog.processors.UnicodeDecoder(),
              structlog.processors.StackInfoRenderer(),
              structlog.processors.format_exc_info,
              structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
          ]
  
      @classmethod
      def extract_from_record(cls, _, __, event_dict):
          """
          Extract thread and process names and add them to the event dict.
          """
          record = event_dict["_record"]
          event_dict["thread_name"] = record.threadName
          event_dict["process_name"] = record.processName
          return event_dict
  
      @classmethod
      def configure_stdlib(cls) -> None:
          level = cls.get_level()
          logging.config.dictConfig(
              {
                  "version": 1,
                  "disable_existing_loggers": True,
                  "formatters": {
                      "이름 지정": {
                          "()": structlog.stdlib.ProcessorFormatter,
                          "processors": [
                              cls.extract_from_record,
                              structlog.stdlib.ProcessorFormatter.remove_processors_meta,
                              structlog.processors.JSONRenderer(),
                          ],
                          "foreign_pre_chain": [
                              structlog.contextvars.merge_contextvars,
                              structlog.stdlib.add_log_level,
                              structlog.stdlib.add_logger_name,
                              structlog.stdlib.PositionalArgumentsFormatter(),
                              structlog.stdlib.ExtraAdder(),
                              cls.timestamper,
                              structlog.processors.UnicodeDecoder(),
                              structlog.processors.StackInfoRenderer(),
                              structlog.processors.format_exc_info,
                          ],
                      },
                  },
                  "handlers": {
                      "default": {
                          "level": level,
                          "class": "logging.StreamHandler",
                          "formatter": "이름 지정",
                      },
                      "file": {
                          "level": level,
                          "class": "logging.handlers.TimedRotatingFileHandler",
                          "filename": "../logs/logfile.log",
                          "when": "midnight",
                          "interval": 1,
                          "formatter": "이름 지정",
                      }
                  },
                  "loggers": {
                      "": {
                          "handlers": ["default", "file"],
                          "level": level,
                          "propagate": False,
                      },
                  },
              }
          )
  
      @classmethod
      def configure_structlog(cls) -> None:
          structlog.configure_once(
              processors=cls.get_processors(),
              wrapper_class=structlog.stdlib.BoundLogger,
              logger_factory=structlog.stdlib.LoggerFactory(),
              cache_logger_on_first_use=True,
          )
  
      @classmethod
      def configure(cls) -> None:
          cls.configure_stdlib()
          cls.configure_structlog()
  
  def configure() -> None:
      Logging.configure()

• middlewares.py
  

  import uuid
  
  import structlog
  from starlette.types import ASGIApp, Receive, Scope, Send
  
  class LogTraceIdMiddleware:
      def __init__(self, app: ASGIApp) -> None:
          self.app = app
  
      async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
          if scope["type"] != "http":
              return await self.app(scope, receive, send)
  
          structlog.contextvars.bind_contextvars(
              trace_id=str(uuid.uuid4()),
              method=scope["method"],
              path=scope["path"],
          )
  
          await self.app(scope, receive, send)
  
          structlog.contextvars.unbind_contextvars("trace_id", "method", "path")​

• main.py
  

  import structlog
  from fastapi import FastAPI
  
  from middlewares import LogTraceIdMiddleware
  from config.logging import Logger
  from config.logging import configure as configure_logging
  
  
  s_logger: Logger = structlog.get_logger()
  configure_logging()
  
  app = FastAPI()
  app.add_middleware(LogTraceIdMiddleware)​

주의 사항은 로그에 한글을 아직 나타낼 방법을 찾지 못했다.

유니코드로 나타나기에 왠만하면 영어로 로그를 남겨보자.