tegwick
1fa0f1e84a
Some checks failed
Test Suite / unit-tests (3.11) (push) Has been cancelled
Test Suite / unit-tests (3.12) (push) Has been cancelled
Test Suite / integration-tests (push) Has been cancelled
Test Suite / e2e-tests (push) Has been cancelled
Test Suite / performance-tests (push) Has been cancelled
Test Suite / code-quality (push) Has been cancelled
Test Suite / security-scan (push) Has been cancelled
Test Suite / test-summary (push) Has been cancelled
- Replace deprecated datetime.utcnow() with datetime.now(timezone.utc) across all domain models, services, infrastructure, and test files - Add missing timezone imports to all affected files - Fix pytest.ini configuration format from [tool:pytest] to [pytest] - Remove warning suppressions to expose actual issues - Ensure proper pytest marker registration for smoke tests Results: - 305 passed, 2 skipped, 0 warnings (down from 111 warnings) - All functionality preserved with modern datetime API usage - Improved code quality by addressing root causes vs suppression 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
302 lines
10 KiB
Python
302 lines
10 KiB
Python
"""
|
|
Custom log formatters for MarkiTect.
|
|
|
|
Provides structured formatting for development and production environments
|
|
with context-aware logging capabilities.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import traceback
|
|
from datetime import datetime, timezone
|
|
from typing import Dict, Any, Optional
|
|
|
|
from .context import get_current_log_context
|
|
|
|
|
|
class BaseFormatter(logging.Formatter):
|
|
"""Base formatter with common functionality."""
|
|
|
|
def __init__(self, enable_context: bool = True, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.enable_context = enable_context
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
"""Format the log record with context information."""
|
|
# Add context information if enabled
|
|
if self.enable_context:
|
|
self._add_context_to_record(record)
|
|
|
|
# Add standard fields
|
|
self._add_standard_fields(record)
|
|
|
|
return super().format(record)
|
|
|
|
def _add_context_to_record(self, record: logging.LogRecord) -> None:
|
|
"""Add context information to log record."""
|
|
context = get_current_log_context()
|
|
if context:
|
|
record.correlation_id = context.correlation_id
|
|
record.operation_id = context.operation_id
|
|
record.operation_type = context.operation_type
|
|
record.user_id = context.user_id
|
|
record.request_id = context.request_id
|
|
|
|
# Add custom fields
|
|
for key, value in context.custom_fields.items():
|
|
setattr(record, f'ctx_{key}', value)
|
|
else:
|
|
record.correlation_id = None
|
|
record.operation_id = None
|
|
record.operation_type = None
|
|
record.user_id = None
|
|
record.request_id = None
|
|
|
|
def _add_standard_fields(self, record: logging.LogRecord) -> None:
|
|
"""Add standard fields to log record."""
|
|
record.timestamp = datetime.now(timezone.utc).isoformat() + 'Z'
|
|
record.logger_name = record.name
|
|
record.level_name = record.levelname
|
|
record.thread_name = record.threadName
|
|
record.process_id = record.process
|
|
|
|
# Add exception information if present
|
|
if record.exc_info and record.exc_info != (None, None, None):
|
|
if isinstance(record.exc_info, tuple) and len(record.exc_info) == 3:
|
|
record.exception_type = record.exc_info[0].__name__ if record.exc_info[0] else None
|
|
record.exception_message = str(record.exc_info[1]) if record.exc_info[1] else None
|
|
record.stack_trace = traceback.format_exception(*record.exc_info)
|
|
else:
|
|
# Handle case where exc_info is True but we need to get current exception
|
|
import sys
|
|
exc_info = sys.exc_info()
|
|
if exc_info[0] is not None:
|
|
record.exception_type = exc_info[0].__name__
|
|
record.exception_message = str(exc_info[1])
|
|
record.stack_trace = traceback.format_exception(*exc_info)
|
|
else:
|
|
record.exception_type = None
|
|
record.exception_message = None
|
|
record.stack_trace = None
|
|
else:
|
|
record.exception_type = None
|
|
record.exception_message = None
|
|
record.stack_trace = None
|
|
|
|
|
|
class DevelopmentFormatter(BaseFormatter):
|
|
"""
|
|
Human-readable formatter for development environment.
|
|
|
|
Provides colored output and structured information for easy debugging.
|
|
"""
|
|
|
|
# Color codes for different log levels
|
|
COLORS = {
|
|
'DEBUG': '\033[36m', # Cyan
|
|
'INFO': '\033[32m', # Green
|
|
'WARNING': '\033[33m', # Yellow
|
|
'ERROR': '\033[31m', # Red
|
|
'CRITICAL': '\033[41m', # Red background
|
|
'RESET': '\033[0m' # Reset
|
|
}
|
|
|
|
def __init__(self, enable_context: bool = True, enable_colors: bool = True, *args, **kwargs):
|
|
super().__init__(enable_context, *args, **kwargs)
|
|
self.enable_colors = enable_colors and self._supports_color()
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
"""Format record for development environment."""
|
|
super().format(record)
|
|
|
|
# Build the message
|
|
parts = []
|
|
|
|
# Timestamp
|
|
timestamp = datetime.fromtimestamp(record.created).strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
|
|
parts.append(f"[{timestamp}]")
|
|
|
|
# Log level with color
|
|
level = record.levelname
|
|
if self.enable_colors and level in self.COLORS:
|
|
level = f"{self.COLORS[level]}{level:<8}{self.COLORS['RESET']}"
|
|
else:
|
|
level = f"{level:<8}"
|
|
parts.append(level)
|
|
|
|
# Logger name (shortened)
|
|
logger_name = self._shorten_logger_name(record.name)
|
|
parts.append(f"[{logger_name}]")
|
|
|
|
# Context information
|
|
if self.enable_context and hasattr(record, 'correlation_id') and record.correlation_id:
|
|
context_parts = []
|
|
if record.correlation_id:
|
|
context_parts.append(f"cid:{record.correlation_id[:8]}")
|
|
if record.operation_id:
|
|
context_parts.append(f"op:{record.operation_id}")
|
|
if context_parts:
|
|
parts.append(f"({' '.join(context_parts)})")
|
|
|
|
# Main message
|
|
parts.append(record.getMessage())
|
|
|
|
# Exception information
|
|
if record.exc_info:
|
|
parts.append(f"\n{self.formatException(record.exc_info)}")
|
|
|
|
# Performance information
|
|
if hasattr(record, 'duration_ms'):
|
|
parts.append(f"[{record.duration_ms:.2f}ms]")
|
|
|
|
return " ".join(parts)
|
|
|
|
def _shorten_logger_name(self, name: str) -> str:
|
|
"""Shorten logger name for compact display."""
|
|
parts = name.split('.')
|
|
if len(parts) <= 2:
|
|
return name
|
|
|
|
# Keep first and last part, abbreviate middle parts
|
|
first = parts[0]
|
|
last = parts[-1]
|
|
middle = '.'.join(part[0] for part in parts[1:-1])
|
|
|
|
return f"{first}.{middle}.{last}" if middle else f"{first}.{last}"
|
|
|
|
def _supports_color(self) -> bool:
|
|
"""Check if terminal supports color output."""
|
|
import os
|
|
import sys
|
|
|
|
# Check if we're in a terminal
|
|
if not hasattr(sys.stdout, 'isatty') or not sys.stdout.isatty():
|
|
return False
|
|
|
|
# Check environment variables
|
|
if os.getenv('NO_COLOR'):
|
|
return False
|
|
|
|
if os.getenv('FORCE_COLOR'):
|
|
return True
|
|
|
|
# Check terminal type
|
|
term = os.getenv('TERM', '')
|
|
return 'color' in term or term in ('xterm', 'xterm-256color', 'screen')
|
|
|
|
|
|
class ProductionFormatter(BaseFormatter):
|
|
"""
|
|
Structured formatter for production environment.
|
|
|
|
Provides JSON-like structured output optimized for log aggregation systems.
|
|
"""
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
"""Format record for production environment."""
|
|
super().format(record)
|
|
|
|
# Build structured log entry
|
|
log_entry = {
|
|
'timestamp': record.timestamp,
|
|
'level': record.levelname,
|
|
'logger': record.name,
|
|
'message': record.getMessage(),
|
|
'thread': record.thread_name,
|
|
'process': record.process_id
|
|
}
|
|
|
|
# Add context information
|
|
if self.enable_context:
|
|
context_info = {}
|
|
if hasattr(record, 'correlation_id') and record.correlation_id:
|
|
context_info['correlation_id'] = record.correlation_id
|
|
if hasattr(record, 'operation_id') and record.operation_id:
|
|
context_info['operation_id'] = record.operation_id
|
|
if hasattr(record, 'operation_type') and record.operation_type:
|
|
context_info['operation_type'] = record.operation_type
|
|
if hasattr(record, 'user_id') and record.user_id:
|
|
context_info['user_id'] = record.user_id
|
|
if hasattr(record, 'request_id') and record.request_id:
|
|
context_info['request_id'] = record.request_id
|
|
|
|
if context_info:
|
|
log_entry['context'] = context_info
|
|
|
|
# Add custom context fields
|
|
custom_fields = {}
|
|
for attr_name in dir(record):
|
|
if attr_name.startswith('ctx_'):
|
|
field_name = attr_name[4:] # Remove 'ctx_' prefix
|
|
custom_fields[field_name] = getattr(record, attr_name)
|
|
|
|
if custom_fields:
|
|
log_entry['custom'] = custom_fields
|
|
|
|
# Add exception information
|
|
if record.exc_info:
|
|
log_entry['exception'] = {
|
|
'type': record.exception_type,
|
|
'message': record.exception_message,
|
|
'traceback': record.stack_trace
|
|
}
|
|
|
|
# Add performance information
|
|
if hasattr(record, 'duration_ms'):
|
|
log_entry['performance'] = {
|
|
'duration_ms': record.duration_ms
|
|
}
|
|
|
|
# Add location information
|
|
log_entry['source'] = {
|
|
'file': record.pathname,
|
|
'line': record.lineno,
|
|
'function': record.funcName
|
|
}
|
|
|
|
return self._format_structured_entry(log_entry)
|
|
|
|
def _format_structured_entry(self, entry: Dict[str, Any]) -> str:
|
|
"""Format structured entry as string."""
|
|
# Use compact JSON format
|
|
return json.dumps(entry, separators=(',', ':'), ensure_ascii=False, default=str)
|
|
|
|
|
|
class PerformanceFormatter(BaseFormatter):
|
|
"""
|
|
Specialized formatter for performance logging.
|
|
|
|
Optimized for capturing timing, metrics, and performance data.
|
|
"""
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
"""Format record for performance logging."""
|
|
super().format(record)
|
|
|
|
# Performance-focused format
|
|
parts = [
|
|
record.timestamp,
|
|
record.levelname,
|
|
record.name
|
|
]
|
|
|
|
# Context for performance tracking
|
|
if self.enable_context and hasattr(record, 'operation_id') and record.operation_id:
|
|
parts.append(f"op:{record.operation_id}")
|
|
|
|
# Main message
|
|
parts.append(record.getMessage())
|
|
|
|
# Performance metrics
|
|
metrics = []
|
|
if hasattr(record, 'duration_ms'):
|
|
metrics.append(f"duration:{record.duration_ms:.2f}ms")
|
|
if hasattr(record, 'memory_mb'):
|
|
metrics.append(f"memory:{record.memory_mb:.2f}MB")
|
|
if hasattr(record, 'cpu_percent'):
|
|
metrics.append(f"cpu:{record.cpu_percent:.1f}%")
|
|
|
|
if metrics:
|
|
parts.append(f"[{', '.join(metrics)}]")
|
|
|
|
return " | ".join(parts) |