File size: 4,140 Bytes
de49715 | 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 | """
Structured JSON Logging Configuration
Provides consistent logging across the application
"""
import logging
import json
import sys
from datetime import datetime
from typing import Any, Dict, Optional
class JSONFormatter(logging.Formatter):
"""Custom JSON formatter for structured logging"""
def format(self, record: logging.LogRecord) -> str:
"""Format log record as JSON"""
log_data = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
# Add extra fields if present
if hasattr(record, 'provider'):
log_data['provider'] = record.provider
if hasattr(record, 'endpoint'):
log_data['endpoint'] = record.endpoint
if hasattr(record, 'duration'):
log_data['duration_ms'] = record.duration
if hasattr(record, 'status'):
log_data['status'] = record.status
if hasattr(record, 'http_code'):
log_data['http_code'] = record.http_code
# Add exception info if present
if record.exc_info:
log_data['exception'] = self.formatException(record.exc_info)
# Add stack trace if present
if record.stack_info:
log_data['stack_trace'] = self.formatStack(record.stack_info)
return json.dumps(log_data)
def setup_logger(name: str, level: str = "INFO") -> logging.Logger:
"""
Setup a logger with JSON formatting
Args:
name: Logger name
level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
Returns:
Configured logger instance
"""
logger = logging.getLogger(name)
# Clear any existing handlers
logger.handlers = []
# Set level
logger.setLevel(getattr(logging, level.upper()))
# Create console handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(getattr(logging, level.upper()))
# Set JSON formatter
json_formatter = JSONFormatter()
console_handler.setFormatter(json_formatter)
# Add handler to logger
logger.addHandler(console_handler)
# Prevent propagation to root logger
logger.propagate = False
return logger
def log_api_request(
logger: logging.Logger,
provider: str,
endpoint: str,
duration_ms: float,
status: str,
http_code: Optional[int] = None,
level: str = "INFO"
):
"""
Log an API request with structured data
Args:
logger: Logger instance
provider: Provider name
endpoint: API endpoint
duration_ms: Request duration in milliseconds
status: Request status (success/error)
http_code: HTTP status code
level: Log level
"""
log_level = getattr(logging, level.upper())
extra = {
'provider': provider,
'endpoint': endpoint,
'duration': duration_ms,
'status': status,
}
if http_code:
extra['http_code'] = http_code
message = f"{provider} - {endpoint} - {status} - {duration_ms}ms"
logger.log(log_level, message, extra=extra)
def log_error(
logger: logging.Logger,
provider: str,
error_type: str,
error_message: str,
endpoint: Optional[str] = None,
exc_info: bool = False
):
"""
Log an error with structured data
Args:
logger: Logger instance
provider: Provider name
error_type: Type of error
error_message: Error message
endpoint: API endpoint (optional)
exc_info: Include exception info
"""
extra = {
'provider': provider,
'error_type': error_type,
}
if endpoint:
extra['endpoint'] = endpoint
message = f"{provider} - {error_type}: {error_message}"
logger.error(message, extra=extra, exc_info=exc_info)
# Global application logger
app_logger = setup_logger("crypto_monitor", level="INFO")
|