"""Core Logging Module This module provides centralized logging functionality for the Adtlas Core module: - Structured logging with context - Multiple log handlers (file, console, database) - Log formatting and filtering - Performance logging - Security event logging - Audit trail logging - Error tracking and alerting - Log rotation and archiving Usage: from apps.core.logging import ( get_logger, log_user_action, log_security_event, log_performance, log_api_request, AuditLogger, SecurityLogger, PerformanceLogger, ) Configuration: Configure logging in Django settings: LOGGING = { 'version': 1, 'disable_existing_loggers': False, 'formatters': { 'detailed': { 'format': '{levelname} {asctime} {name} {process:d} {thread:d} {message}', 'style': '{', }, }, 'handlers': { 'core_file': { 'level': 'INFO', 'class': 'logging.handlers.RotatingFileHandler', 'filename': 'logs/core.log', 'maxBytes': 1024*1024*10, # 10MB 'backupCount': 5, 'formatter': 'detailed', }, }, 'loggers': { 'core': { 'handlers': ['core_file'], 'level': 'INFO', 'propagate': True, }, }, } Security: - Sensitive data is automatically filtered from logs - Log access is restricted to authorized personnel - Log integrity is maintained through checksums - Security events are logged with high priority """ import json import logging import logging.handlers import os import sys import traceback from datetime import datetime from typing import Any, Dict, List, Optional, Union from pathlib import Path import hashlib import threading from functools import wraps import time from django.conf import settings from django.contrib.auth import get_user_model from django.core.mail import mail_admins from django.utils import timezone from django.db import models from django.http import HttpRequest # Thread-local storage for request context _thread_local = threading.local() User = get_user_model() # ============================================================================= # Log Models for Database Logging # ============================================================================= class LogEntry(models.Model): """ Model to store log entries in database. """ LEVEL_CHOICES = [ ('DEBUG', 'Debug'), ('INFO', 'Info'), ('WARNING', 'Warning'), ('ERROR', 'Error'), ('CRITICAL', 'Critical'), ] timestamp = models.DateTimeField(auto_now_add=True, db_index=True) level = models.CharField(max_length=10, choices=LEVEL_CHOICES, db_index=True) logger_name = models.CharField(max_length=100, db_index=True) message = models.TextField() module = models.CharField(max_length=100, blank=True) function = models.CharField(max_length=100, blank=True) line_number = models.IntegerField(null=True, blank=True) user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='core_log_entries') ip_address = models.GenericIPAddressField(null=True, blank=True) user_agent = models.TextField(blank=True) request_path = models.CharField(max_length=500, blank=True) request_method = models.CharField(max_length=10, blank=True) session_key = models.CharField(max_length=40, blank=True) extra_data = models.JSONField(default=dict, blank=True) stack_trace = models.TextField(blank=True) class Meta: db_table = 'core_log_entries' ordering = ['-timestamp'] indexes = [ models.Index(fields=['timestamp', 'level']), models.Index(fields=['logger_name', 'timestamp']), models.Index(fields=['user', 'timestamp']), ] def __str__(self): return f"{self.timestamp} [{self.level}] {self.logger_name}: {self.message[:100]}" class AuditLogEntry(models.Model): """ Model to store audit log entries. """ timestamp = models.DateTimeField(auto_now_add=True, db_index=True) user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='core_audit_log_entries') action = models.CharField(max_length=100, db_index=True) object_type = models.CharField(max_length=100, blank=True) object_id = models.CharField(max_length=100, blank=True) object_repr = models.CharField(max_length=200, blank=True) changes = models.JSONField(default=dict, blank=True) ip_address = models.GenericIPAddressField(null=True, blank=True) user_agent = models.TextField(blank=True) session_key = models.CharField(max_length=40, blank=True) success = models.BooleanField(default=True) error_message = models.TextField(blank=True) class Meta: db_table = 'core_audit_log_entries' ordering = ['-timestamp'] indexes = [ models.Index(fields=['timestamp', 'action']), models.Index(fields=['user', 'timestamp']), models.Index(fields=['object_type', 'object_id']), ] def __str__(self): user_str = self.user.username if self.user else 'Anonymous' return f"{self.timestamp} {user_str}: {self.action}" class SecurityLogEntry(models.Model): """ Model to store security-related log entries. """ SEVERITY_CHOICES = [ ('LOW', 'Low'), ('MEDIUM', 'Medium'), ('HIGH', 'High'), ('CRITICAL', 'Critical'), ] timestamp = models.DateTimeField(auto_now_add=True, db_index=True) event_type = models.CharField(max_length=100, db_index=True) severity = models.CharField(max_length=10, choices=SEVERITY_CHOICES, db_index=True) description = models.TextField() user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='core_security_log_entries') ip_address = models.GenericIPAddressField(null=True, blank=True) user_agent = models.TextField(blank=True) request_path = models.CharField(max_length=500, blank=True) additional_data = models.JSONField(default=dict, blank=True) resolved = models.BooleanField(default=False) resolved_by = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, related_name='resolved_security_logs' ) resolved_at = models.DateTimeField(null=True, blank=True) class Meta: db_table = 'core_security_log_entries' ordering = ['-timestamp'] indexes = [ models.Index(fields=['timestamp', 'severity']), models.Index(fields=['event_type', 'timestamp']), models.Index(fields=['resolved', 'severity']), ] def __str__(self): return f"{self.timestamp} [{self.severity}] {self.event_type}: {self.description[:100]}" # ============================================================================= # Custom Log Handlers # ============================================================================= class DatabaseLogHandler(logging.Handler): """ Custom log handler that stores logs in database. """ def emit(self, record): try: # Get request context request = getattr(_thread_local, 'request', None) # Create log entry log_entry = LogEntry( level=record.levelname, logger_name=record.name, message=record.getMessage(), module=getattr(record, 'module', ''), function=getattr(record, 'funcName', ''), line_number=getattr(record, 'lineno', None), ) # Add request context if available if request: log_entry.user = getattr(request, 'user', None) if hasattr(request, 'user') and request.user.is_authenticated else None log_entry.ip_address = self._get_client_ip(request) log_entry.user_agent = request.META.get('HTTP_USER_AGENT', '') log_entry.request_path = request.path log_entry.request_method = request.method log_entry.session_key = request.session.session_key if hasattr(request, 'session') else '' # Add extra data if hasattr(record, 'extra_data'): log_entry.extra_data = record.extra_data # Add stack trace for errors if record.exc_info: log_entry.stack_trace = self.format(record) log_entry.save() except Exception: # Don't let logging errors break the application self.handleError(record) def _get_client_ip(self, request): """Get client IP address from request.""" x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0].strip() else: ip = request.META.get('REMOTE_ADDR', '') return ip class SensitiveDataFilter(logging.Filter): """ Filter to remove sensitive data from log records. """ SENSITIVE_KEYS = [ 'password', 'passwd', 'pwd', 'secret', 'token', 'key', 'api_key', 'access_token', 'refresh_token', 'auth_token', 'session_key', 'credit_card', 'ssn', 'social_security', 'bank_account', ] def filter(self, record): if hasattr(record, 'msg') and isinstance(record.msg, str): record.msg = self._filter_sensitive_data(record.msg) if hasattr(record, 'args') and record.args: record.args = tuple( self._filter_sensitive_data(str(arg)) if isinstance(arg, str) else arg for arg in record.args ) return True def _filter_sensitive_data(self, text): """Filter sensitive data from text.""" for key in self.SENSITIVE_KEYS: # Simple pattern matching - can be enhanced with regex if key in text.lower(): # Replace with asterisks text = text.replace(key, '*' * len(key)) return text class StructuredFormatter(logging.Formatter): """ Formatter that outputs structured JSON logs. """ def format(self, record): log_data = { 'timestamp': datetime.fromtimestamp(record.created).isoformat(), 'level': record.levelname, 'logger': record.name, 'message': record.getMessage(), 'module': getattr(record, 'module', ''), 'function': getattr(record, 'funcName', ''), 'line': getattr(record, 'lineno', None), } # Add request context if available request = getattr(_thread_local, 'request', None) if request: log_data['request'] = { 'method': request.method, 'path': request.path, 'user': request.user.username if hasattr(request, 'user') and request.user.is_authenticated else None, 'ip': self._get_client_ip(request), } # Add extra data if hasattr(record, 'extra_data'): log_data['extra'] = record.extra_data # Add exception info if record.exc_info: log_data['exception'] = { 'type': record.exc_info[0].__name__, 'message': str(record.exc_info[1]), 'traceback': traceback.format_exception(*record.exc_info), } return json.dumps(log_data, default=str) def _get_client_ip(self, request): """Get client IP address from request.""" x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0].strip() else: ip = request.META.get('REMOTE_ADDR', '') return ip # ============================================================================= # Logger Factory and Utilities # ============================================================================= def get_logger(name: str) -> logging.Logger: """ Get a configured logger instance. Args: name: Logger name Returns: Configured logger instance """ logger = logging.getLogger(name) # Add sensitive data filter if not already added if not any(isinstance(f, SensitiveDataFilter) for f in logger.filters): logger.addFilter(SensitiveDataFilter()) return logger def set_request_context(request: HttpRequest): """ Set request context for current thread. Args: request: HTTP request object """ _thread_local.request = request def clear_request_context(): """ Clear request context for current thread. """ if hasattr(_thread_local, 'request'): delattr(_thread_local, 'request') def setup_logging(): """ Setup logging configuration. """ # Ensure log directory exists log_dir = getattr(settings, 'LOG_DIR', 'logs') Path(log_dir).mkdir(parents=True, exist_ok=True) # Configure root logger root_logger = logging.getLogger() root_logger.setLevel(logging.INFO) # Add console handler for development if settings.DEBUG: console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(logging.DEBUG) console_formatter = logging.Formatter( '%(asctime)s [%(levelname)s] %(name)s: %(message)s' ) console_handler.setFormatter(console_formatter) root_logger.addHandler(console_handler) # Add file handler file_handler = logging.handlers.RotatingFileHandler( os.path.join(log_dir, 'adtlas.log'), maxBytes=10*1024*1024, # 10MB backupCount=5 ) file_handler.setLevel(logging.INFO) file_formatter = StructuredFormatter() file_handler.setFormatter(file_formatter) root_logger.addHandler(file_handler) # Add database handler for important logs try: db_handler = DatabaseLogHandler() db_handler.setLevel(logging.WARNING) root_logger.addHandler(db_handler) except Exception: # Database might not be ready during startup pass # ============================================================================= # Specialized Loggers # ============================================================================= class AuditLogger: """ Logger for audit events. """ def __init__(self): self.logger = get_logger('audit') def log_action(self, user, action, object_type='', object_id='', object_repr='', changes=None, success=True, error_message='', request=None): """ Log an audit action. Args: user: User performing the action action: Action description object_type: Type of object object_id: Object ID object_repr: Object representation changes: Dictionary of changes success: Whether action was successful error_message: Error message if failed request: HTTP request object """ try: audit_entry = AuditLogEntry( user=user, action=action, object_type=object_type, object_id=str(object_id) if object_id else '', object_repr=object_repr, changes=changes or {}, success=success, error_message=error_message, ) if request: audit_entry.ip_address = self._get_client_ip(request) audit_entry.user_agent = request.META.get('HTTP_USER_AGENT', '') audit_entry.session_key = request.session.session_key if hasattr(request, 'session') else '' audit_entry.save() # Also log to regular logger self.logger.info( f"Audit: {user.username if user else 'Anonymous'} {action} {object_type} {object_id}", extra={ 'extra_data': { 'audit_id': audit_entry.id, 'action': action, 'object_type': object_type, 'object_id': object_id, 'success': success, } } ) except Exception as e: self.logger.error(f"Failed to log audit action: {str(e)}") def _get_client_ip(self, request): """Get client IP address from request.""" x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0].strip() else: ip = request.META.get('REMOTE_ADDR', '') return ip class SecurityLogger: """ Logger for security events. """ def __init__(self): self.logger = get_logger('security') def log_event(self, event_type, description, severity='MEDIUM', user=None, request=None, additional_data=None): """ Log a security event. Args: event_type: Type of security event description: Event description severity: Event severity (LOW, MEDIUM, HIGH, CRITICAL) user: User involved in the event request: HTTP request object additional_data: Additional event data """ try: security_entry = SecurityLogEntry( event_type=event_type, description=description, severity=severity, user=user, additional_data=additional_data or {}, ) if request: security_entry.ip_address = self._get_client_ip(request) security_entry.user_agent = request.META.get('HTTP_USER_AGENT', '') security_entry.request_path = request.path security_entry.save() # Log to regular logger with appropriate level log_level = { 'LOW': logging.INFO, 'MEDIUM': logging.WARNING, 'HIGH': logging.ERROR, 'CRITICAL': logging.CRITICAL, }.get(severity, logging.WARNING) self.logger.log( log_level, f"Security Event [{severity}] {event_type}: {description}", extra={ 'extra_data': { 'security_id': security_entry.id, 'event_type': event_type, 'severity': severity, 'user_id': user.id if user else None, } } ) # Send email for critical events if severity == 'CRITICAL' and not settings.DEBUG: try: mail_admins( f"Critical Security Event: {event_type}", f"Description: {description}\n\n" f"User: {user.username if user else 'Anonymous'}\n" f"Time: {timezone.now()}\n" f"Additional Data: {additional_data}" ) except Exception: pass # Don't let email failures break security logging except Exception as e: self.logger.error(f"Failed to log security event: {str(e)}") def _get_client_ip(self, request): """Get client IP address from request.""" x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0].strip() else: ip = request.META.get('REMOTE_ADDR', '') return ip class PerformanceLogger: """ Logger for performance metrics. """ def __init__(self): self.logger = get_logger('performance') def log_timing(self, operation, duration, context=None): """ Log performance timing. Args: operation: Operation name duration: Duration in seconds context: Additional context data """ self.logger.info( f"Performance: {operation} took {duration:.4f}s", extra={ 'extra_data': { 'operation': operation, 'duration': duration, 'context': context or {}, } } ) def log_query_count(self, operation, query_count, context=None): """ Log database query count. Args: operation: Operation name query_count: Number of queries context: Additional context data """ self.logger.info( f"Queries: {operation} executed {query_count} queries", extra={ 'extra_data': { 'operation': operation, 'query_count': query_count, 'context': context or {}, } } ) def log_memory_usage(self, operation, memory_mb, context=None): """ Log memory usage. Args: operation: Operation name memory_mb: Memory usage in MB context: Additional context data """ self.logger.info( f"Memory: {operation} used {memory_mb:.2f}MB", extra={ 'extra_data': { 'operation': operation, 'memory_mb': memory_mb, 'context': context or {}, } } ) # ============================================================================= # Convenience Functions # ============================================================================= def log_user_action(user, action, object_type='', object_id='', changes=None, request=None): """ Convenience function to log user actions. Args: user: User performing the action action: Action description object_type: Type of object object_id: Object ID changes: Dictionary of changes request: HTTP request object """ audit_logger = AuditLogger() audit_logger.log_action( user=user, action=action, object_type=object_type, object_id=object_id, changes=changes, request=request ) def log_security_event(event_type, description, severity='MEDIUM', user=None, request=None): """ Convenience function to log security events. Args: event_type: Type of security event description: Event description severity: Event severity user: User involved request: HTTP request object """ security_logger = SecurityLogger() security_logger.log_event( event_type=event_type, description=description, severity=severity, user=user, request=request ) def log_performance(operation, duration, context=None): """ Convenience function to log performance metrics. Args: operation: Operation name duration: Duration in seconds context: Additional context """ perf_logger = PerformanceLogger() perf_logger.log_timing(operation, duration, context) def log_api_request(request, response_time=None, status_code=None): """ Log API request details. Args: request: HTTP request object response_time: Response time in seconds status_code: HTTP status code """ logger = get_logger('api') log_data = { 'method': request.method, 'path': request.path, 'query_string': request.META.get('QUERY_STRING', ''), 'user': request.user.username if hasattr(request, 'user') and request.user.is_authenticated else None, 'ip_address': _get_client_ip(request), 'user_agent': request.META.get('HTTP_USER_AGENT', ''), } if response_time is not None: log_data['response_time'] = response_time if status_code is not None: log_data['status_code'] = status_code logger.info( f"API Request: {request.method} {request.path}", extra={'extra_data': log_data} ) def _get_client_ip(request): """Get client IP address from request.""" x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0].strip() else: ip = request.META.get('REMOTE_ADDR', '') return ip # ============================================================================= # Context Managers and Decorators # ============================================================================= class LogContext: """ Context manager for adding context to logs. Usage: with LogContext(operation='user_login', user_id=123): # Your code here logger.info("User logged in") """ def __init__(self, **context): self.context = context self.old_context = None def __enter__(self): self.old_context = getattr(_thread_local, 'log_context', {}) _thread_local.log_context = {**self.old_context, **self.context} return self def __exit__(self, exc_type, exc_val, exc_tb): _thread_local.log_context = self.old_context def with_logging(operation_name: str, log_args: bool = False, log_result: bool = False): """ Decorator to add logging to functions. Args: operation_name: Name of the operation log_args: Whether to log function arguments log_result: Whether to log function result Returns: Decorated function """ def decorator(func): @wraps(func) def wrapper(*args, **kwargs): logger = get_logger(f"{func.__module__}.{func.__name__}") start_time = time.time() log_data = { 'operation': operation_name, 'function': func.__name__, 'module': func.__module__, } if log_args: log_data['args'] = str(args)[:1000] # Limit size log_data['kwargs'] = str(kwargs)[:1000] logger.info(f"Starting {operation_name}", extra={'extra_data': log_data}) try: result = func(*args, **kwargs) duration = time.time() - start_time log_data.update({ 'status': 'success', 'duration': duration, }) if log_result: log_data['result'] = str(result)[:1000] logger.info( f"Completed {operation_name} in {duration:.4f}s", extra={'extra_data': log_data} ) return result except Exception as e: duration = time.time() - start_time log_data.update({ 'status': 'error', 'duration': duration, 'error': str(e), 'error_type': type(e).__name__, }) logger.error( f"Failed {operation_name} after {duration:.4f}s: {str(e)}", extra={'extra_data': log_data}, exc_info=True ) raise return wrapper return decorator # ============================================================================= # Initialize logging on module import # ============================================================================= # Setup logging when module is imported if not getattr(settings, 'LOGGING_SETUP_DONE', False): try: setup_logging() settings.LOGGING_SETUP_DONE = True except Exception: # Don't let logging setup errors break the application pass # ============================================================================= # Export all logging utilities # ============================================================================= __all__ = [ # Models 'LogEntry', 'AuditLogEntry', 'SecurityLogEntry', # Handlers and Formatters 'DatabaseLogHandler', 'SensitiveDataFilter', 'StructuredFormatter', # Logger Factory 'get_logger', 'setup_logging', # Context Management 'set_request_context', 'clear_request_context', # Specialized Loggers 'AuditLogger', 'SecurityLogger', 'PerformanceLogger', # Convenience Functions 'log_user_action', 'log_security_event', 'log_performance', 'log_api_request', # Context Managers and Decorators 'LogContext', 'with_logging', ]