This repository was archived by the owner on Jun 5, 2025. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 90
Expand file tree
/
Copy pathlogging.py
More file actions
180 lines (150 loc) · 5.7 KB
/
logging.py
File metadata and controls
180 lines (150 loc) · 5.7 KB
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
import datetime
import json
import logging
import sys
from typing import Any, Optional
from .config import LogFormat, LogLevel
class JSONFormatter(logging.Formatter):
"""Custom formatter that outputs log records as JSON."""
def __init__(self) -> None:
"""Initialize the JSON formatter."""
super().__init__()
self.default_time_format = "%Y-%m-%dT%H:%M:%S"
self.default_msec_format = "%s.%03dZ"
def format(self, record: logging.LogRecord) -> str:
"""Format the log record as a JSON string.
Args:
record: The log record to format
Returns:
str: JSON formatted log entry
"""
# Create the base log entry
log_entry: dict[str, Any] = {
"timestamp": self.formatTime(record, self.default_time_format),
"level": record.levelname,
"module": record.module,
"message": record.getMessage(),
"extra": {},
}
# Add extra fields from the record
extra_attrs = {}
for key, value in record.__dict__.items():
if key not in {
"args",
"asctime",
"created",
"exc_info",
"exc_text",
"filename",
"funcName",
"levelname",
"levelno",
"lineno",
"module",
"msecs",
"msg",
"name",
"pathname",
"process",
"processName",
"relativeCreated",
"stack_info",
"thread",
"threadName",
"extra",
}:
extra_attrs[key] = value
# Handle the explicit extra parameter if present
if hasattr(record, "extra"):
try:
if isinstance(record.extra, dict):
extra_attrs.update(record.extra)
except Exception:
extra_attrs["unserializable_extra"] = str(record.extra)
# Add all extra attributes to the log entry
if extra_attrs:
try:
json.dumps(extra_attrs) # Test if serializable
log_entry["extra"] = extra_attrs
except (TypeError, ValueError):
# If serialization fails, convert values to strings
serializable_extra = {}
for key, value in extra_attrs.items():
try:
json.dumps({key: value}) # Test individual value
serializable_extra[key] = value
except (TypeError, ValueError):
serializable_extra[key] = str(value)
log_entry["extra"] = serializable_extra
# Handle exception info if present
if record.exc_info:
log_entry["extra"]["exception"] = self.formatException(record.exc_info)
# Handle stack info if present
if record.stack_info:
log_entry["extra"]["stack_info"] = self.formatStack(record.stack_info)
return json.dumps(log_entry)
class TextFormatter(logging.Formatter):
"""Standard text formatter with consistent timestamp format."""
def __init__(self) -> None:
"""Initialize the text formatter."""
super().__init__(
fmt="%(asctime)s - %(levelname)s - %(name)s - %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S.%03dZ",
)
def formatTime( # noqa: N802
self, record: logging.LogRecord, datefmt: Optional[str] = None
) -> str:
"""Format the time with millisecond precision.
Args:
record: The log record
datefmt: The date format string (ignored as we use a fixed format)
Returns:
str: Formatted timestamp
"""
ct = datetime.datetime.fromtimestamp(record.created, datetime.UTC)
return ct.strftime(self.datefmt)
def setup_logging(
log_level: Optional[LogLevel] = None, log_format: Optional[LogFormat] = None
) -> None:
"""Configure the logging system.
Args:
log_level: The logging level to use. Defaults to INFO if not specified.
log_format: The log format to use. Defaults to JSON if not specified.
This configures two handlers:
- stderr_handler: For ERROR, CRITICAL, and WARNING messages
- stdout_handler: For INFO and DEBUG messages
"""
if log_level is None:
log_level = LogLevel.INFO
if log_format is None:
log_format = LogFormat.JSON
# Create formatters
json_formatter = JSONFormatter()
text_formatter = TextFormatter()
formatter = json_formatter if log_format == LogFormat.JSON else text_formatter
# Create handlers for stdout and stderr
stdout_handler = logging.StreamHandler(sys.stdout)
stderr_handler = logging.StreamHandler(sys.stderr)
# Set formatters
stdout_handler.setFormatter(formatter)
stderr_handler.setFormatter(formatter)
# Configure log routing
stdout_handler.addFilter(lambda record: record.levelno <= logging.INFO)
stderr_handler.addFilter(lambda record: record.levelno > logging.INFO)
# Get root logger and configure it
root_logger = logging.getLogger()
root_logger.setLevel(log_level.value)
# Remove any existing handlers and add our new ones
root_logger.handlers.clear()
root_logger.addHandler(stdout_handler)
root_logger.addHandler(stderr_handler)
# Create a logger for our package
logger = logging.getLogger("codegate")
logger.debug(
"Logging initialized",
extra={
"log_level": log_level.value,
"log_format": log_format.value,
"handlers": ["stdout", "stderr"],
},
)