본문으로 건너뛰기
Advertisement

로깅 시스템

logging 모듈과 structlog로 레벨·핸들러·포매터를 설계하고 프로덕션 로깅을 구축합니다.

설치​

pip install structlog python-json-logger

logging — 표준 라이브러리​

import logging
import logging.handlers
import sys
from pathlib import Path


# ── 기본 설정 ─────────────────────────────────────────────
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

logger = logging.getLogger(__name__)

logger.debug("디버그 정보")
logger.info("일반 정보")
logger.warning("경고")
logger.error("에러")
logger.critical("치명적 오류")

# 예외 스택 트레이스 포함
try:
    1 / 0
except ZeroDivisionError:
    logger.exception("예외 발생")  # ERROR 레벨 + 스택 트레이스

프로덕션 로깅 설정 (dictConfig)​

import logging
import logging.config
from pathlib import Path

LOG_DIR = Path("logs")
LOG_DIR.mkdir(exist_ok=True)

LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,

    "formatters": {
        "detailed": {
            "format": "%(asctime)s [%(levelname)s] %(name)s:%(lineno)d - %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S",
        },
        "simple": {
            "format": "[%(levelname)s] %(message)s",
        },
        "json": {
            "()": "pythonjsonlogger.jsonlogger.JsonFormatter",
            "format": "%(asctime)s %(name)s %(levelname)s %(message)s",
        },
    },

    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stdout",
            "formatter": "simple",
            "level": "DEBUG",
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "filename": str(LOG_DIR / "app.log"),
            "maxBytes": 10 * 1024 * 1024,  # 10 MB
            "backupCount": 5,
            "formatter": "detailed",
            "encoding": "utf-8",
        },
        "error_file": {
            "class": "logging.handlers.RotatingFileHandler",
            "filename": str(LOG_DIR / "error.log"),
            "maxBytes": 5 * 1024 * 1024,
            "backupCount": 3,
            "formatter": "detailed",
            "level": "ERROR",
            "encoding": "utf-8",
        },
        "timed_file": {
            "class": "logging.handlers.TimedRotatingFileHandler",
            "filename": str(LOG_DIR / "daily.log"),
            "when": "midnight",
            "interval": 1,
            "backupCount": 30,
            "formatter": "json",
            "encoding": "utf-8",
        },
    },

    "loggers": {
        "myapp": {
            "level": "DEBUG",
            "handlers": ["console", "file", "error_file"],
            "propagate": False,
        },
        "uvicorn": {
            "level": "INFO",
            "handlers": ["console"],
            "propagate": False,
        },
        "sqlalchemy.engine": {
            "level": "WARNING",  # SQL 쿼리 로그 억제
            "handlers": ["file"],
            "propagate": False,
        },
    },

    "root": {
        "level": "WARNING",
        "handlers": ["console"],
    },
}

logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger("myapp.service")
logger.info("서버 시작")

structlog — 구조화 로깅​

import structlog
import logging


# ── 설정 ──────────────────────────────────────────────────
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.dev.ConsoleRenderer(),          # 개발: 컬러 출력
        # structlog.processors.JSONRenderer(),    # 프로덕션: JSON
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.DEBUG),
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)

log = structlog.get_logger()


# ── 구조화된 로그 ──────────────────────────────────────────
log.info("사용자 로그인", user_id=42, ip="192.168.1.1", method="jwt")
log.warning("레이트 리밋 초과", user_id=42, limit=100, current=150)
log.error("DB 연결 실패", host="db.prod.internal", port=5432, retry=3)

# 컨텍스트 바인딩
request_log = log.bind(request_id="abc-123", user_id=42)
request_log.info("요청 시작", path="/api/users", method="GET")
request_log.info("쿼리 실행", table="users", duration_ms=12.3)
request_log.info("응답 완료", status=200, duration_ms=45.7)

# contextvars — 요청 전체에 컨텍스트 공유 (async 지원)
from structlog.contextvars import bind_contextvars, clear_contextvars

async def handle_request(request_id: str, user_id: int):
    clear_contextvars()
    bind_contextvars(request_id=request_id, user_id=user_id)

    log.info("처리 시작")  # 자동으로 request_id, user_id 포함
    # ... 로직
    log.info("처리 완료")

FastAPI 로깅 미들웨어​

import time
import uuid
import logging
import structlog
from fastapi import FastAPI, Request, Response

logger = structlog.get_logger()
app = FastAPI()


@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    request_id = str(uuid.uuid4())[:8]
    start = time.perf_counter()

    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        request_id=request_id,
        method=request.method,
        path=request.url.path,
    )

    logger.info("요청 수신")

    try:
        response: Response = await call_next(request)
        duration_ms = (time.perf_counter() - start) * 1000

        logger.info(
            "요청 완료",
            status=response.status_code,
            duration_ms=round(duration_ms, 2),
        )
        response.headers["X-Request-ID"] = request_id
        return response

    except Exception as exc:
        logger.exception("요청 처리 중 예외", exc_info=exc)
        raise

로그 레벨 가이드​

DEBUG    — 개발 중 상세 흐름 (프로덕션 비활성화)
INFO     — 정상 동작 이벤트 (서버 시작, 요청 완료)
WARNING  — 주의 필요하지만 정상 동작 (레이트 리밋 근접, 재시도 성공)
ERROR    — 기능 실패 (DB 오류, 결제 실패) — 알람 필요
CRITICAL — 즉시 대응 필요 (서버 다운, 데이터 손실 위험)

정리​

도구특징추천 상황
logging표준 라이브러리, 유연한 설정모든 프로젝트 기반
structlog구조화 JSON, 컨텍스트 바인딩API 서버, 마이크로서비스
Rich logging컬러 터미널 출력개발 환경
  • 개발 환경: logging.basicConfig 또는 structlog.dev.ConsoleRenderer
  • 프로덕션: dictConfig + JSON 포매터 + 파일 로테이션 + 중앙 로그 수집
Advertisement
Advertisement