1. FastAPI 로깅 아키텍처
기본 구조
FastAPI는 Python 표준 logging 모듈을 기반으로 하며, 여러 로거가 계층적으로 구성됩니다:
Root Logger
├── uvicorn (ASGI 서버)
│ ├── uvicorn.error
│ └── uvicorn.access
├── fastapi (프레임워크)
└── your_app (애플리케이션)
├── your_app.services
├── your_app.repositories
└── your_app.utils2. 로그 레벨 상세 분석
레벨별 용도와 기준
🔴 CRITICAL (50)
# 언제 사용?
- 애플리케이션이 완전히 중단될 수 있는 치명적 오류
- 데이터 손실이나 보안 침해
- 즉시 대응이 필요한 상황
# 예시
logger.critical("Database connection pool exhausted - service unavailable")
logger.critical("Security breach detected - unauthorized admin access")
logger.critical("Payment gateway down - revenue impact")🟠 ERROR (40)
# 언제 사용?
- 기능이 실패했지만 애플리케이션은 계속 실행
- 사용자에게 영향을 주는 오류
- 개발팀이 알아야 하는 문제
# 예시
logger.error(f"Failed to send email to {user_email}: {str(e)}")
logger.error(f"Payment processing failed for order {order_id}")
logger.error(f"External API call failed: {api_endpoint}")🟡 WARNING (30)
# 언제 사용?
- 예상치 못한 상황이지만 처리 가능
- 성능 저하나 리소스 부족 징후
- 설정 문제나 deprecated 기능 사용
# 예시
logger.warning(f"Database connection slow: {response_time}ms")
logger.warning(f"Unusual user activity: {user_id} - {action_count} requests/min")
logger.warning("Using deprecated API endpoint")🔵 INFO (20)
# 언제 사용?
- 주요 비즈니스 이벤트
- 시스템 상태 변화
- 사용자 행동 추적
# 예시
logger.info(f"User {user_id} logged in from {ip_address}")
logger.info(f"Order {order_id} created successfully")
logger.info("Application started successfully")⚪ DEBUG (10)
# 언제 사용?
- 개발/디버깅 시에만
- 상세한 실행 흐름
- 변수 값이나 내부 상태
# 예시
logger.debug(f"Processing request: {request_data}")
logger.debug(f"Database query: {sql_query}")
logger.debug(f"Function called with params: {params}")3. FastAPI 로깅 설정 방법
기본 설정
# main.py
import logging
from fastapi import FastAPI
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[
logging.FileHandler("app.log"),
logging.StreamHandler()
]
)
app = FastAPI()
logger = logging.getLogger(__name__)
@app.on_event("startup")
async def startup_event():
logger.info("Application startup completed")고급 설정 (권장)
# logging_config.py
import logging
import logging.config
from datetime import datetime
import os
LOGGING_CONFIG = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"default": {
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
},
"detailed": {
"format": "%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(funcName)s - %(message)s",
},
"json": {
"format": '{"timestamp": "%(asctime)s", "logger": "%(name)s", "level": "%(levelname)s", "message": "%(message)s", "file": "%(filename)s", "line": %(lineno)d}',
}
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "default",
"level": "INFO",
},
"file": {
"class": "logging.handlers.RotatingFileHandler",
"formatter": "detailed",
"filename": "logs/app.log",
"maxBytes": 10485760, # 10MB
"backupCount": 5,
"level": "DEBUG",
},
"error_file": {
"class": "logging.handlers.RotatingFileHandler",
"formatter": "json",
"filename": "logs/error.log",
"maxBytes": 10485760,
"backupCount": 10,
"level": "ERROR",
},
},
"loggers": {
"": { # root logger
"handlers": ["console", "file"],
"level": "INFO",
"propagate": False,
},
"uvicorn": {
"handlers": ["console"],
"level": "INFO",
"propagate": False,
},
"uvicorn.error": {
"handlers": ["error_file"],
"level": "ERROR",
"propagate": False,
},
"your_app": {
"handlers": ["console", "file", "error_file"],
"level": "DEBUG",
"propagate": False,
},
},
}
def setup_logging():
# 로그 디렉토리 생성
os.makedirs("logs", exist_ok=True)
logging.config.dictConfig(LOGGING_CONFIG)4. 환경별 로깅 전략
Development 환경
# 모든 로그 출력, 상세한 디버그 정보
LOGGING_LEVEL = "DEBUG"
LOG_TO_CONSOLE = True
LOG_TO_FILE = True
DETAILED_FORMAT = TrueStaging 환경
# WARNING 이상만 출력, 성능 테스트 고려
LOGGING_LEVEL = "WARNING"
LOG_TO_CONSOLE = True
LOG_TO_FILE = True
PERFORMANCE_LOGGING = TrueProduction 환경
# ERROR 이상만 출력, 보안과 성능 최적화
LOGGING_LEVEL = "ERROR"
LOG_TO_CONSOLE = False # 컨테이너 환경에서는 True
LOG_TO_FILE = True
STRUCTURED_LOGGING = True # JSON 형태
EXTERNAL_LOGGING = True # ELK, CloudWatch 등5. 실무 로깅 패턴
서비스 레이어 로깅
# services/user_service.py
import logging
logger = logging.getLogger(__name__)
class UserService:
async def create_user(self, user_data: dict):
logger.info(f"Creating user: {user_data.get('email')}")
try:
# 비즈니스 로직
user = await self.user_repo.create(user_data)
logger.info(f"User created successfully: {user.id}")
return user
except ValidationError as e:
logger.warning(f"User creation validation failed: {str(e)}")
raise
except Exception as e:
logger.error(f"Failed to create user: {str(e)}", exc_info=True)
raise InternalServerError("사용자 생성에 실패했습니다.")API 엔드포인트 로깅
# routers/users.py
import logging
from fastapi import APIRouter, Request
logger = logging.getLogger(__name__)
router = APIRouter()
@router.post("/users")
async def create_user(request: Request, user_data: UserCreate):
# 요청 로깅 (개인정보 제외)
logger.info(f"POST /users - IP: {request.client.host}")
try:
user = await user_service.create_user(user_data.dict())
# 성공 로깅
logger.info(f"User created - ID: {user.id}")
return {"success": True, "user_id": user.id}
except ValidationError as e:
logger.warning(f"User creation validation error: {str(e)}")
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"User creation failed: {str(e)}")
raise HTTPException(status_code=500, detail="Internal server error")미들웨어 로깅
# middleware/logging_middleware.py
import time
import logging
from fastapi import Request
logger = logging.getLogger(__name__)
async def log_requests(request: Request, call_next):
start_time = time.time()
# 요청 로깅
logger.info(f"Request: {request.method} {request.url.path}")
response = await call_next(request)
# 응답 로깅
process_time = time.time() - start_time
logger.info(
f"Response: {response.status_code} - "
f"Time: {process_time:.4f}s - "
f"Path: {request.url.path}"
)
# 느린 요청 경고
if process_time > 2.0:
logger.warning(f"Slow request detected: {process_time:.4f}s - {request.url.path}")
return response6. 보안 및 개인정보 고려사항
민감한 정보 필터링
import re
class SensitiveDataFilter(logging.Filter):
def filter(self, record):
# 비밀번호, 토큰 등 마스킹
if hasattr(record, 'msg'):
record.msg = re.sub(r'password["\']:\s*["\'][^"\']*["\']', 'password: "***"', str(record.msg))
record.msg = re.sub(r'token["\']:\s*["\'][^"\']*["\']', 'token: "***"', str(record.msg))
return True
# 로거에 필터 추가
logger.addFilter(SensitiveDataFilter())GDPR 준수 로깅
def log_user_action(user_id: str, action: str, details: dict = None):
# 개인정보는 해시화하거나 제외
safe_details = {
k: v for k, v in (details or {}).items()
if k not in ['email', 'phone', 'name']
}
logger.info(f"User action - ID: {hash(user_id)}, Action: {action}, Details: {safe_details}")7. 성능 최적화
조건부 로깅
# 비용이 높은 로그는 레벨 체크 후 실행
if logger.isEnabledFor(logging.DEBUG):
expensive_debug_info = generate_debug_data()
logger.debug(f"Debug info: {expensive_debug_info}")비동기 로깅
from logging.handlers import QueueHandler
import asyncio
import queue
# 비동기 로그 처리기
log_queue = queue.Queue()
queue_handler = QueueHandler(log_queue)
logger.addHandler(queue_handler)8. 권장 로깅 수준별 가이드라인
🟢 개발환경에서 권장
- DEBUG: 모든 함수 호출, 변수 값
- INFO: 비즈니스 로직 흐름
- WARNING: 예상 가능한 예외상황
- ERROR: 모든 예외와 스택트레이스
🟡 스테이징환경에서 권장
- INFO: 주요 비즈니스 이벤트만
- WARNING: 성능 경고, 리소스 부족
- ERROR: 모든 에러 (스택트레이스 포함)
🔴 프로덕션환경에서 권장
- WARNING: 시스템 이상 징후
- ERROR: 사용자 영향 있는 오류
- CRITICAL: 즉시 대응 필요한 장애
9. 모니터링 및 알림 연동
Structured Logging (JSON)
import json
import logging
class JSONFormatter(logging.Formatter):
def format(self, record):
log_entry = {
"timestamp": self.formatTime(record),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"module": record.module,
"function": record.funcName,
"line": record.lineno
}
if record.exc_info:
log_entry["exception"] = self.formatException(record.exc_info)
return json.dumps(log_entry, ensure_ascii=False)외부 시스템 연동
# Slack 알림 핸들러
class SlackHandler(logging.Handler):
def emit(self, record):
if record.levelno >= logging.ERROR:
# Slack으로 즉시 알림 전송
send_slack_alert(self.format(record))
# CloudWatch 로그 전송
class CloudWatchHandler(logging.Handler):
def emit(self, record):
# AWS CloudWatch로 로그 전송
send_to_cloudwatch(self.format(record))이렇게 설정하면 안정적이고 확장 가능한 로깅 시스템을 구축할 수 있습니다! 🎯
.