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
|
# vim: se sw=2:
# MIT licensed
# Copyright (c) 2018-2020,2023-2024 lilydjwg <lilydjwg@gmail.com>, et al.
import logging
import os
import io
import traceback
import sys
import structlog
from .httpclient import TemporaryError
def _console_msg(event):
evt = event['event']
if evt == 'up-to-date':
msg = 'up-to-date, version %s' % event['version']
del event['version']
elif evt == 'updated':
if event.get('old_version'):
msg = 'updated from %(old_version)s to %(version)s' % event
else:
msg = 'updated to %(version)s' % event
del event['version'], event['old_version']
else:
msg = evt
if 'name' in event:
msg = f"{event['name']}: {msg}"
del event['name']
event['msg'] = msg
return event
def exc_info(logger, level, event):
if level == 'exception':
event['exc_info'] = True
return event
def filter_nones(logger, level, event):
if 'url' in event and event['url'] is None:
del event['url']
return event
def filter_taskname(logger, level, event):
# added in Python 3.12, not useful to us, but appears as a normal KV.
if 'taskName' in event:
del event['taskName']
return event
def filter_exc(logger, level, event):
exc_info = event.get('exc_info')
if not exc_info:
return event
if exc_info is True:
exc = sys.exc_info()[1]
else:
exc = exc_info
if isinstance(exc, TemporaryError):
if exc.code == 599: # network issues
del event['exc_info']
event['error'] = exc
return event
def stdlib_renderer(logger, level, event):
# return event unchanged for further processing
std_event = _console_msg(event.copy())
try:
logger = logging.getLogger(std_event.pop('logger_name'))
except KeyError:
logger = logging.getLogger()
msg = std_event.pop('msg', std_event.pop('event'))
exc_info = std_event.pop('exc_info', None)
if 'error' in std_event:
std_event['error'] = repr(std_event['error'])
getattr(logger, level)(
msg, exc_info = exc_info, extra=std_event,
)
return event
_renderer = structlog.processors.JSONRenderer(ensure_ascii=False)
def json_renderer(logger, level, event):
event['level'] = level
return _renderer(logger, level, event)
def null_renderer(logger, level, event):
return ''
class _Logger(logging.Logger):
_my_srcfile = os.path.normcase(
stdlib_renderer.__code__.co_filename)
_structlog_dir = os.path.dirname(structlog.__file__)
def findCaller(self, stack_info=False, stacklevel=1):
"""
Find the stack frame of the caller so that we can note the source
file name, line number and function name.
"""
f = logging.currentframe()
#On some versions of IronPython, currentframe() returns None if
#IronPython isn't run with -X:Frames.
if f is not None:
f = f.f_back
orig_f = f
while f and stacklevel > 1:
f = f.f_back
stacklevel -= 1
if not f:
f = orig_f
rv = "(unknown file)", 0, "(unknown function)", None
while hasattr(f, "f_code"):
co = f.f_code
filename = os.path.normcase(co.co_filename)
if filename in [logging._srcfile, self._my_srcfile] \
or filename.startswith(self._structlog_dir):
f = f.f_back
continue
sinfo = None
if stack_info:
sio = io.StringIO()
sio.write('Stack (most recent call last):\n')
traceback.print_stack(f, file=sio)
sinfo = sio.getvalue()
if sinfo[-1] == '\n':
sinfo = sinfo[:-1]
sio.close()
rv = (co.co_filename, f.f_lineno, co.co_name, sinfo)
break
return rv
def fix_logging():
logging.setLoggerClass(_Logger)
|
