File size: 4,741 Bytes
53bec59 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | """
Production-Grade Structured Logging
"""
import logging
import sys
import json
from datetime import datetime
from typing import Any, Dict
from pathlib import Path
from pythonjsonlogger import jsonlogger
from .config import settings
class CustomJsonFormatter(jsonlogger.JsonFormatter):
"""Custom JSON formatter with additional fields"""
def add_fields(self, log_record: Dict[str, Any], record: logging.LogRecord, message_dict: dict):
super().add_fields(log_record, record, message_dict)
# Add timestamp
log_record['timestamp'] = datetime.utcnow().isoformat()
# Add log level
log_record['level'] = record.levelname
# Add application context
log_record['app'] = settings.APP_NAME
log_record['version'] = settings.APP_VERSION
log_record['environment'] = settings.ENVIRONMENT
# Add request ID if available (will be set by middleware)
if hasattr(record, 'request_id'):
log_record['request_id'] = record.request_id
# Add user ID if available
if hasattr(record, 'user_id'):
log_record['user_id'] = record.user_id
def setup_logging():
"""Configure application logging"""
# Create logger
logger = logging.getLogger()
logger.setLevel(getattr(logging, settings.LOG_LEVEL.upper()))
# Remove existing handlers
logger.handlers = []
# Console handler
console_handler = logging.StreamHandler(sys.stdout)
if settings.LOG_FORMAT == "json":
console_formatter = CustomJsonFormatter(
'%(timestamp)s %(level)s %(name)s %(message)s'
)
else:
console_formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
console_handler.setFormatter(console_formatter)
logger.addHandler(console_handler)
# File handler (if configured)
if settings.LOG_FILE:
log_file = Path(settings.LOG_FILE)
log_file.parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(log_file)
if settings.LOG_FORMAT == "json":
file_formatter = CustomJsonFormatter(
'%(timestamp)s %(level)s %(name)s %(message)s'
)
else:
file_formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)
return logger
# Create module-level logger
logger = setup_logging()
def log_api_request(
method: str,
path: str,
status_code: int,
duration_ms: float,
user_id: str = None,
request_id: str = None
):
"""Log API request with structured data"""
logger.info(
"API Request",
extra={
"method": method,
"path": path,
"status_code": status_code,
"duration_ms": duration_ms,
"user_id": user_id,
"request_id": request_id,
"event_type": "api_request"
}
)
def log_prediction(
model_type: str,
input_size: int,
confidence: float,
duration_ms: float,
cached: bool = False,
user_id: str = None
):
"""Log ML prediction with metrics"""
logger.info(
"ML Prediction",
extra={
"model_type": model_type,
"input_size": input_size,
"confidence": confidence,
"duration_ms": duration_ms,
"cached": cached,
"user_id": user_id,
"event_type": "prediction"
}
)
def log_error(
error: Exception,
context: Dict[str, Any] = None,
user_id: str = None,
request_id: str = None
):
"""Log error with full context"""
logger.error(
f"Error: {str(error)}",
extra={
"error_type": type(error).__name__,
"error_message": str(error),
"context": context or {},
"user_id": user_id,
"request_id": request_id,
"event_type": "error"
},
exc_info=True
)
if __name__ == "__main__":
# Test logging
logger.info("Application starting")
logger.debug("Debug message")
logger.warning("Warning message")
logger.error("Error message")
log_api_request("GET", "/api/v1/health", 200, 5.2, request_id="test-123")
log_prediction("deepfake", 1024, 0.95, 125.5, cached=False, user_id="user-1")
|