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
|
# -*- test-case-name: twisted.logger.test.test_io -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
File-like object that logs.
"""
import sys
from typing import AnyStr, Iterable, Optional
from constantly import NamedConstant
from incremental import Version
from twisted.python.deprecate import deprecatedProperty
from ._levels import LogLevel
from ._logger import Logger
class LoggingFile:
"""
File-like object that turns C{write()} calls into logging events.
Note that because event formats are L{str}, C{bytes} received via C{write()}
are converted to C{str}, which is the opposite of what C{file} does.
@ivar softspace: Attribute to make this class more file-like under Python 2;
value is zero or one. Do not use.
"""
_softspace = 0
@deprecatedProperty(Version("Twisted", 21, 2, 0))
def softspace(self):
return self._softspace
@softspace.setter # type: ignore[no-redef]
def softspace(self, value):
self._softspace = value
def __init__(
self,
logger: Logger,
level: NamedConstant = LogLevel.info,
encoding: Optional[str] = None,
) -> None:
"""
@param logger: the logger to log through.
@param level: the log level to emit events with.
@param encoding: The encoding to expect when receiving bytes via
C{write()}. If L{None}, use C{sys.getdefaultencoding()}.
"""
self.level = level
self.log = logger
if encoding is None:
self._encoding = sys.getdefaultencoding()
else:
self._encoding = encoding
self._buffer = ""
self._closed = False
@property
def closed(self) -> bool:
"""
Read-only property. Is the file closed?
@return: true if closed, otherwise false.
"""
return self._closed
@property
def encoding(self) -> str:
"""
Read-only property. File encoding.
@return: an encoding.
"""
return self._encoding
@property
def mode(self) -> str:
"""
Read-only property. File mode.
@return: "w"
"""
return "w"
@property
def newlines(self) -> None:
"""
Read-only property. Types of newlines encountered.
@return: L{None}
"""
return None
@property
def name(self) -> str:
"""
The name of this file; a repr-style string giving information about its
namespace.
@return: A file name.
"""
return "<{} {}#{}>".format(
self.__class__.__name__,
self.log.namespace,
self.level.name,
)
def close(self) -> None:
"""
Close this file so it can no longer be written to.
"""
self._closed = True
def flush(self) -> None:
"""
No-op; this file does not buffer.
"""
pass
def fileno(self) -> int:
"""
Returns an invalid file descriptor, since this is not backed by an FD.
@return: C{-1}
"""
return -1
def isatty(self) -> bool:
"""
A L{LoggingFile} is not a TTY.
@return: C{False}
"""
return False
def write(self, message: AnyStr) -> None:
"""
Log the given message.
@param message: The message to write.
"""
if self._closed:
raise ValueError("I/O operation on closed file")
if isinstance(message, bytes):
text = message.decode(self._encoding)
else:
text = message
lines = (self._buffer + text).split("\n")
self._buffer = lines[-1]
lines = lines[0:-1]
for line in lines:
self.log.emit(self.level, format="{log_io}", log_io=line)
def writelines(self, lines: Iterable[AnyStr]) -> None:
"""
Log each of the given lines as a separate message.
@param lines: Data to write.
"""
for line in lines:
self.write(line)
def _unsupported(self, *args: object) -> None:
"""
Template for unsupported operations.
@param args: Arguments.
"""
raise OSError("unsupported operation")
read = _unsupported
next = _unsupported
readline = _unsupported
readlines = _unsupported
xreadlines = _unsupported
seek = _unsupported
tell = _unsupported
truncate = _unsupported
|
