Repository navigation
Expand file tree
/
Copy pathfile_logger.py
More file actions
337 lines (260 loc) · 11.1 KB
/
Copy pathfile_logger.py
File metadata and controls
337 lines (260 loc) · 11.1 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
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
import logging
import structlog
from pathlib import Path
from typing import Optional
def configure_file_logger(
logger_name: str,
log_file_path: str | Path,
log_level: int = logging.INFO,
log_format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
encoding: str = "utf-8",
) -> structlog.BoundLogger:
"""
Configure a structlog logger to log messages only to a file.
Args:
logger_name: Name of the logger
log_file_path: Path to the log file
log_level: Logging level (default: INFO)
log_format: Format string for log messages
encoding: File encoding (default: utf-8)
Returns:
structlog.BoundLogger: Configured structlog logger
Example:
>>> logger = configure_file_logger("my_app", "logs/app.log")
>>> logger.info("Application started")
>>> logger.error("An error occurred", error_code=500)
"""
# Convert path to Path object if it's a string
log_file_path = Path(log_file_path)
# Create log directory if it doesn't exist
log_file_path.parent.mkdir(parents=True, exist_ok=True)
# Get the standard library logger
std_logger = logging.getLogger(logger_name)
std_logger.setLevel(log_level)
# Remove only FileHandler handlers to avoid duplicates while preserving other handlers
for handler in std_logger.handlers[:]:
if isinstance(handler, logging.FileHandler):
std_logger.removeHandler(handler)
# Create a simple file handler
file_handler = logging.FileHandler(
filename=log_file_path,
encoding=encoding,
)
# Create formatter
formatter = logging.Formatter(log_format)
file_handler.setFormatter(formatter)
# Add handler to logger
std_logger.addHandler(file_handler)
# Prevent propagation to root logger to avoid console output
std_logger.propagate = False
# Get the structlog logger
logger = structlog.get_logger(logger_name)
return logger
def configure_structured_file_logger(
logger_name: str,
log_file_path: str | Path,
log_level: int = logging.INFO,
encoding: str = "utf-8",
include_timestamp: bool = True,
include_logger_name: bool = True,
include_level: bool = True,
) -> structlog.BoundLogger:
"""
Configure a structlog logger with structured logging to a file.
Args:
logger_name: Name of the logger
log_file_path: Path to the log file
log_level: Logging level (default: INFO)
encoding: File encoding (default: utf-8)
include_timestamp: Whether to include timestamp in logs (default: True)
include_logger_name: Whether to include logger name in logs (default: True)
include_level: Whether to include log level in logs (default: True)
Returns:
structlog.BoundLogger: Configured structlog logger with structured logging
Example:
>>> logger = configure_structured_file_logger("my_app", "logs/app.json")
>>> logger.info("User logged in", user_id=123, ip="192.168.1.1")
"""
import json
# Convert path to Path object if it's a string
log_file_path = Path(log_file_path)
# Create log directory if it doesn't exist
log_file_path.parent.mkdir(parents=True, exist_ok=True)
# Get the standard library logger
std_logger = logging.getLogger(logger_name)
std_logger.setLevel(log_level)
# Remove only FileHandler handlers to avoid duplicates while preserving other handlers
for handler in std_logger.handlers[:]:
if isinstance(handler, logging.FileHandler):
std_logger.removeHandler(handler)
# Create a simple file handler
file_handler = logging.FileHandler(
filename=log_file_path,
encoding=encoding,
)
# Create JSON formatter for structured logging
class JSONFormatter(logging.Formatter):
def format(self, record):
log_entry = {
"message": record.getMessage(),
}
if include_timestamp:
log_entry["timestamp"] = self.formatTime(record)
if include_logger_name:
log_entry["logger"] = record.name
if include_level:
log_entry["level"] = record.levelname
# Add any extra fields from structlog
if hasattr(record, "structlog"):
log_entry.update(record.structlog)
return json.dumps(log_entry)
formatter = JSONFormatter()
file_handler.setFormatter(formatter)
# Add handler to logger
std_logger.addHandler(file_handler)
# Prevent propagation to root logger to avoid console output
std_logger.propagate = False
# Get the structlog logger
logger = structlog.get_logger(logger_name)
return logger
def get_file_logger(
logger_name: str,
log_file_path: str | Path,
structured: bool = False,
**kwargs
) -> structlog.BoundLogger:
"""
Convenience function to get a file logger with either standard or structured logging.
Args:
logger_name: Name of the logger
log_file_path: Path to the log file
structured: Whether to use structured (JSON) logging (default: False)
**kwargs: Additional arguments passed to the configuration function
Returns:
structlog.BoundLogger: Configured structlog logger
Example:
>>> # Standard logging
>>> logger = get_file_logger("app", "logs/app.log")
>>>
>>> # Structured logging
>>> logger = get_file_logger("app", "logs/app.json", structured=True)
"""
if structured:
return configure_structured_file_logger(logger_name, log_file_path, **kwargs)
else:
return configure_file_logger(logger_name, log_file_path, **kwargs)
def reset_logger_to_defaults(logger_name: str) -> structlog.BoundLogger:
"""
Reset a named structlog logger to its default configuration.
This function removes all custom handlers and resets the logger to use
the default structlog configuration, which typically outputs to console.
Args:
logger_name: Name of the logger to reset
Returns:
structlog.BoundLogger: Reset structlog logger
Example:
>>> # Configure a file logger
>>> logger = configure_file_logger("my_app", "logs/app.log")
>>> logger.info("This goes to file")
>>>
>>> # Reset to defaults (console output)
>>> logger = reset_logger_to_defaults("my_app")
>>> logger.info("This goes to console")
"""
# Get the standard library logger
std_logger = logging.getLogger(logger_name)
# Remove all existing handlers
for handler in std_logger.handlers[:]:
std_logger.removeHandler(handler)
# Reset logger level to default (NOTSET)
std_logger.setLevel(logging.NOTSET)
# Re-enable propagation to parent loggers
std_logger.propagate = True
# Get the structlog logger (this will use default structlog configuration)
logger = structlog.get_logger(logger_name)
return logger
def reset_all_loggers_to_defaults() -> None:
"""
Reset all loggers to their default configuration.
This function removes all custom handlers from all loggers and resets
them to use the default structlog configuration.
Example:
>>> # Configure multiple file loggers
>>> logger1 = configure_file_logger("app1", "logs/app1.log")
>>> logger2 = configure_file_logger("app2", "logs/app2.log")
>>>
>>> # Reset all loggers to defaults
>>> reset_all_loggers_to_defaults()
>>>
>>> # Now all loggers will output to console by default
>>> logger1.info("This goes to console")
>>> logger2.info("This also goes to console")
"""
# Get all existing loggers
loggers = [logging.getLogger(name) for name in logging.root.manager.loggerDict]
# Also include the root logger
loggers.append(logging.getLogger())
for logger in loggers:
# Remove all existing handlers
for handler in logger.handlers[:]:
logger.removeHandler(handler)
# Reset logger level to default
logger.setLevel(logging.NOTSET)
# Re-enable propagation
logger.propagate = True
def get_logger_info(logger_name: str) -> dict:
"""
Get information about a logger's current configuration.
Args:
logger_name: Name of the logger to inspect
Returns:
dict: Information about the logger's configuration
Example:
>>> logger = configure_file_logger("my_app", "logs/app.log")
>>> info = get_logger_info("my_app")
>>> print(info)
>>> # Output: {'name': 'my_app', 'level': 20, 'handlers': 1, 'propagate': False}
"""
std_logger = logging.getLogger(logger_name)
return {
"name": logger_name,
"level": std_logger.level,
"handlers": len(std_logger.handlers),
"propagate": std_logger.propagate,
"handler_types": [type(handler).__name__ for handler in std_logger.handlers],
}
# Example usage and testing
if __name__ == "__main__":
# Example 1: Standard file logging
logger1 = configure_file_logger("test_app", "logs/test.log")
logger1.info("This is a test message")
logger1.error("This is an error message", error_code=500)
# Example 2: Structured file logging
logger2 = configure_structured_file_logger("test_app_structured", "logs/test.json")
logger2.info("User action", user_id=123, action="login", ip="192.168.1.1")
logger2.error("Database error", error_code=500, table="users", query="SELECT *")
# Example 3: Using convenience function
logger3 = get_file_logger("convenience_app", "logs/convenience.log")
logger3.info("Using convenience function")
logger4 = get_file_logger("convenience_structured", "logs/convenience.json", structured=True)
logger4.info("Structured logging with convenience", event="test", data={"key": "value"})
# Example 4: Demonstrating reset functionality
print("\n=== Testing Reset Functionality ===")
# Show logger info before reset
print("Before reset:")
print(f"test_app logger info: {get_logger_info('test_app')}")
# Reset specific logger
reset_logger = reset_logger_to_defaults("test_app")
reset_logger.info("This message goes to console (after reset)")
# Show logger info after reset
print("After reset:")
print(f"test_app logger info: {get_logger_info('test_app')}")
# Example 5: Reset all loggers
print("\n=== Resetting All Loggers ===")
reset_all_loggers_to_defaults()
# All loggers now use default configuration
logger1.info("This also goes to console now")
logger2.info("This also goes to console now")
print("\nLog files created successfully!")
print("Check the 'logs' directory for the generated log files.")
print("After reset, all loggers output to console by default.")