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
|
# -*- test-case-name: twisted.logger.test.test_legacy -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Integration with L{twisted.python.log}.
"""
from typing import TYPE_CHECKING, Any, Callable, Dict, Optional
from zope.interface import implementer
from ._format import formatEvent
from ._interfaces import ILogObserver, LogEvent
from ._levels import LogLevel
from ._stdlib import StringifiableFromEvent, fromStdlibLogLevelMapping
if TYPE_CHECKING:
from twisted.python.log import ILogObserver as ILegacyLogObserver
@implementer(ILogObserver)
class LegacyLogObserverWrapper:
"""
L{ILogObserver} that wraps a L{twisted.python.log.ILogObserver}.
Received (new-style) events are modified prior to forwarding to
the legacy observer to ensure compatibility with observers that
expect legacy events.
"""
def __init__(self, legacyObserver: "ILegacyLogObserver") -> None:
"""
@param legacyObserver: a legacy observer to which this observer will
forward events.
"""
self.legacyObserver = legacyObserver
def __repr__(self) -> str:
return "{self.__class__.__name__}({self.legacyObserver})".format(self=self)
def __call__(self, event: LogEvent) -> None:
"""
Forward events to the legacy observer after editing them to
ensure compatibility.
@param event: an event
"""
# The "message" key is required by textFromEventDict()
if "message" not in event:
event["message"] = ()
if "time" not in event:
event["time"] = event["log_time"]
if "system" not in event:
event["system"] = event.get("log_system", "-")
# Format new style -> old style
if "format" not in event and event.get("log_format", None) is not None:
# Create an object that implements __str__() in order to defer the
# work of formatting until it's needed by a legacy log observer.
event["format"] = "%(log_legacy)s"
event["log_legacy"] = StringifiableFromEvent(event.copy())
# In the old-style system, the 'message' key always holds a tuple
# of messages. If we find the 'message' key here to not be a
# tuple, it has been passed as new-style parameter. We drop it
# here because we render it using the old-style 'format' key,
# which otherwise doesn't get precedence, and the original event
# has been copied above.
if not isinstance(event["message"], tuple):
event["message"] = ()
# From log.failure() -> isError blah blah
if "log_failure" in event:
if "failure" not in event:
event["failure"] = event["log_failure"]
if "isError" not in event:
event["isError"] = 1
if "why" not in event:
event["why"] = formatEvent(event)
elif "isError" not in event:
if event["log_level"] in (LogLevel.error, LogLevel.critical):
event["isError"] = 1
else:
event["isError"] = 0
self.legacyObserver(event)
def publishToNewObserver(
observer: ILogObserver,
eventDict: Dict[str, Any],
textFromEventDict: Callable[[Dict[str, Any]], Optional[str]],
) -> None:
"""
Publish an old-style (L{twisted.python.log}) event to a new-style
(L{twisted.logger}) observer.
@note: It's possible that a new-style event was sent to a
L{LegacyLogObserverWrapper}, and may now be getting sent back to a
new-style observer. In this case, it's already a new-style event,
adapted to also look like an old-style event, and we don't need to
tweak it again to be a new-style event, hence this checks for
already-defined new-style keys.
@param observer: A new-style observer to handle this event.
@param eventDict: An L{old-style <twisted.python.log>}, log event.
@param textFromEventDict: callable that can format an old-style event as a
string. Passed here rather than imported to avoid circular dependency.
"""
if "log_time" not in eventDict:
eventDict["log_time"] = eventDict["time"]
if "log_format" not in eventDict:
text = textFromEventDict(eventDict)
if text is not None:
eventDict["log_text"] = text
eventDict["log_format"] = "{log_text}"
if "log_level" not in eventDict:
if "logLevel" in eventDict:
try:
level = fromStdlibLogLevelMapping[eventDict["logLevel"]]
except KeyError:
level = None
elif "isError" in eventDict:
if eventDict["isError"]:
level = LogLevel.critical
else:
level = LogLevel.info
else:
level = LogLevel.info
if level is not None:
eventDict["log_level"] = level
if "log_namespace" not in eventDict:
eventDict["log_namespace"] = "log_legacy"
if "log_system" not in eventDict and "system" in eventDict:
eventDict["log_system"] = eventDict["system"]
observer(eventDict)
|
