mirror of
https://github.com/pacnpal/thrillwiki_django_no_react.git
synced 2025-12-20 08:11:08 -05:00
262 lines
7.2 KiB
Python
262 lines
7.2 KiB
Python
"""
|
|
Centralized logging configuration for ThrillWiki.
|
|
Provides structured logging with proper formatting and context.
|
|
"""
|
|
|
|
import logging
|
|
import sys
|
|
from typing import Dict, Any, Optional
|
|
from django.conf import settings
|
|
from django.utils import timezone
|
|
|
|
|
|
class ThrillWikiFormatter(logging.Formatter):
|
|
"""Custom formatter for ThrillWiki logs with structured output."""
|
|
|
|
def format(self, record):
|
|
# Add timestamp if not present
|
|
if not hasattr(record, "timestamp"):
|
|
record.timestamp = timezone.now().isoformat()
|
|
|
|
# Add request context if available
|
|
if hasattr(record, "request"):
|
|
record.request_id = getattr(record.request, "id", "unknown")
|
|
record.user_id = (
|
|
getattr(record.request.user, "id", "anonymous")
|
|
if hasattr(record.request, "user")
|
|
else "unknown"
|
|
)
|
|
record.path = getattr(record.request, "path", "unknown")
|
|
record.method = getattr(record.request, "method", "unknown")
|
|
|
|
# Structure the log message
|
|
if hasattr(record, "extra_data"):
|
|
record.structured_data = record.extra_data
|
|
|
|
return super().format(record)
|
|
|
|
|
|
def get_logger(name: str) -> logging.Logger:
|
|
"""
|
|
Get a configured logger for ThrillWiki components.
|
|
|
|
Args:
|
|
name: Logger name (usually __name__)
|
|
|
|
Returns:
|
|
Configured logger instance
|
|
"""
|
|
logger = logging.getLogger(name)
|
|
|
|
# Only configure if not already configured
|
|
if not logger.handlers:
|
|
handler = logging.StreamHandler(sys.stdout)
|
|
formatter = ThrillWikiFormatter(
|
|
fmt="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
|
)
|
|
handler.setFormatter(formatter)
|
|
logger.addHandler(handler)
|
|
logger.setLevel(logging.INFO if settings.DEBUG else logging.WARNING)
|
|
|
|
return logger
|
|
|
|
|
|
def log_exception(
|
|
logger: logging.Logger,
|
|
exception: Exception,
|
|
*,
|
|
context: Optional[Dict[str, Any]] = None,
|
|
request=None,
|
|
level: int = logging.ERROR,
|
|
) -> None:
|
|
"""
|
|
Log an exception with structured context.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
exception: Exception to log
|
|
context: Additional context data
|
|
request: Django request object
|
|
level: Log level
|
|
"""
|
|
log_data = {
|
|
"exception_type": exception.__class__.__name__,
|
|
"exception_message": str(exception),
|
|
"context": context or {},
|
|
}
|
|
|
|
if request:
|
|
log_data.update(
|
|
{
|
|
"request_path": getattr(request, "path", "unknown"),
|
|
"request_method": getattr(request, "method", "unknown"),
|
|
"user_id": (
|
|
getattr(request.user, "id", "anonymous")
|
|
if hasattr(request, "user")
|
|
else "unknown"
|
|
),
|
|
}
|
|
)
|
|
|
|
logger.log(
|
|
level,
|
|
f"Exception occurred: {exception}",
|
|
extra={"extra_data": log_data},
|
|
exc_info=True,
|
|
)
|
|
|
|
|
|
def log_business_event(
|
|
logger: logging.Logger,
|
|
event_type: str,
|
|
*,
|
|
message: str,
|
|
context: Optional[Dict[str, Any]] = None,
|
|
request=None,
|
|
level: int = logging.INFO,
|
|
) -> None:
|
|
"""
|
|
Log a business event with structured context.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
event_type: Type of business event
|
|
message: Event message
|
|
context: Additional context data
|
|
request: Django request object
|
|
level: Log level
|
|
"""
|
|
log_data = {"event_type": event_type, "context": context or {}}
|
|
|
|
if request:
|
|
log_data.update(
|
|
{
|
|
"request_path": getattr(request, "path", "unknown"),
|
|
"request_method": getattr(request, "method", "unknown"),
|
|
"user_id": (
|
|
getattr(request.user, "id", "anonymous")
|
|
if hasattr(request, "user")
|
|
else "unknown"
|
|
),
|
|
}
|
|
)
|
|
|
|
logger.log(level, message, extra={"extra_data": log_data})
|
|
|
|
|
|
def log_performance_metric(
|
|
logger: logging.Logger,
|
|
operation: str,
|
|
*,
|
|
duration_ms: float,
|
|
context: Optional[Dict[str, Any]] = None,
|
|
level: int = logging.INFO,
|
|
) -> None:
|
|
"""
|
|
Log a performance metric.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
operation: Operation name
|
|
duration_ms: Duration in milliseconds
|
|
context: Additional context data
|
|
level: Log level
|
|
"""
|
|
log_data = {
|
|
"metric_type": "performance",
|
|
"operation": operation,
|
|
"duration_ms": duration_ms,
|
|
"context": context or {},
|
|
}
|
|
|
|
message = f"Performance: {operation} took {duration_ms:.2f}ms"
|
|
logger.log(level, message, extra={"extra_data": log_data})
|
|
|
|
|
|
def log_api_request(
|
|
logger: logging.Logger,
|
|
request,
|
|
*,
|
|
response_status: Optional[int] = None,
|
|
duration_ms: Optional[float] = None,
|
|
level: int = logging.INFO,
|
|
) -> None:
|
|
"""
|
|
Log an API request with context.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
request: Django request object
|
|
response_status: HTTP response status code
|
|
duration_ms: Request duration in milliseconds
|
|
level: Log level
|
|
"""
|
|
log_data = {
|
|
"request_type": "api",
|
|
"path": getattr(request, "path", "unknown"),
|
|
"method": getattr(request, "method", "unknown"),
|
|
"user_id": (
|
|
getattr(request.user, "id", "anonymous")
|
|
if hasattr(request, "user")
|
|
else "unknown"
|
|
),
|
|
"response_status": response_status,
|
|
"duration_ms": duration_ms,
|
|
}
|
|
|
|
message = f"API Request: {request.method} {request.path}"
|
|
if response_status:
|
|
message += f" -> {response_status}"
|
|
if duration_ms:
|
|
message += f" ({duration_ms:.2f}ms)"
|
|
|
|
logger.log(level, message, extra={"extra_data": log_data})
|
|
|
|
|
|
def log_security_event(
|
|
logger: logging.Logger,
|
|
event_type: str,
|
|
*,
|
|
message: str,
|
|
severity: str = "medium",
|
|
context: Optional[Dict[str, Any]] = None,
|
|
request=None,
|
|
) -> None:
|
|
"""
|
|
Log a security-related event.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
event_type: Type of security event
|
|
message: Event message
|
|
severity: Event severity (low, medium, high, critical)
|
|
context: Additional context data
|
|
request: Django request object
|
|
"""
|
|
log_data = {
|
|
"security_event": True,
|
|
"event_type": event_type,
|
|
"severity": severity,
|
|
"context": context or {},
|
|
}
|
|
|
|
if request:
|
|
log_data.update(
|
|
{
|
|
"request_path": getattr(request, "path", "unknown"),
|
|
"request_method": getattr(request, "method", "unknown"),
|
|
"user_id": (
|
|
getattr(request.user, "id", "anonymous")
|
|
if hasattr(request, "user")
|
|
else "unknown"
|
|
),
|
|
"remote_addr": request.META.get("REMOTE_ADDR", "unknown"),
|
|
"user_agent": request.META.get("HTTP_USER_AGENT", "unknown"),
|
|
}
|
|
)
|
|
|
|
# Use WARNING for medium/high, ERROR for critical
|
|
level = logging.ERROR if severity in ["high", "critical"] else logging.WARNING
|
|
|
|
logger.log(level, f"SECURITY: {message}", extra={"extra_data": log_data})
|