File size: 1,991 Bytes
255cbd1 | 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 | """
Logging utility with structured logging support
"""
import logging
import sys
from typing import Optional
from datetime import datetime
def get_logger(
name: str,
level: int = logging.INFO,
log_file: Optional[str] = None
) -> logging.Logger:
"""
Get or create a logger with consistent formatting
Args:
name: Logger name (usually __name__)
level: Logging level
log_file: Optional file path for file logging
Returns:
Configured logger instance
"""
logger = logging.getLogger(name)
# Avoid adding handlers multiple times
if logger.handlers:
return logger
logger.setLevel(level)
# Create formatter
formatter = logging.Formatter(
fmt='%(asctime)s | %(levelname)-8s | %(name)s | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# Console handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(level)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
# File handler (optional)
if log_file:
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(level)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
return logger
class JobContextAdapter(logging.LoggerAdapter):
"""
Logger adapter that adds job context to all log messages
"""
def process(self, msg, kwargs):
extra = kwargs.get('extra', {})
extra.update(self.extra)
kwargs['extra'] = extra
return msg, kwargs
def get_job_logger(job_id: str, level: int = logging.INFO) -> logging.Logger:
"""
Get a logger with job context
Args:
job_id: Unique job identifier
level: Logging level
Returns:
Logger with job context
"""
logger = get_logger(f"job.{job_id}", level)
return JobContextAdapter(logger, {'job_id': job_id})
|