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
|
# -*- 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 ._levels import LogLevel
class LoggingFile(object):
"""
File-like object that turns C{write()} calls into logging events.
Note that because event formats are C{unicode}, C{bytes} received via
C{write()} are converted to C{unicode}, which is the opposite of what
C{file} does.
@ivar softspace: File-like L{'softspace' attribute <file.softspace>}; zero
or one.
@type softspace: L{int}
"""
softspace = 0
def __init__(self, logger, level=LogLevel.info, encoding=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()}.
@type encoding: L{str}
@param log: The logger to send events to.
@type log: L{Logger}
"""
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):
"""
Read-only property. Is the file closed?
@return: true if closed, otherwise false.
@rtype: L{bool}
"""
return self._closed
@property
def encoding(self):
"""
Read-only property. File encoding.
@return: an encoding.
@rtype: L{str}
"""
return self._encoding
@property
def mode(self):
"""
Read-only property. File mode.
@return: "w"
@rtype: L{str}
"""
return "w"
@property
def newlines(self):
"""
Read-only property. Types of newlines encountered.
@return: L{None}
@rtype: L{None}
"""
return None
@property
def name(self):
"""
The name of this file; a repr-style string giving information about its
namespace.
@return: A file name.
@rtype: L{str}
"""
return (
"<{0} {1}#{2}>".format(
self.__class__.__name__,
self.log.namespace,
self.level.name,
)
)
def close(self):
"""
Close this file so it can no longer be written to.
"""
self._closed = True
def flush(self):
"""
No-op; this file does not buffer.
"""
pass
def fileno(self):
"""
Returns an invalid file descriptor, since this is not backed by an FD.
@return: C{-1}
@rtype: L{int}
"""
return -1
def isatty(self):
"""
A L{LoggingFile} is not a TTY.
@return: C{False}
@rtype: L{bool}
"""
return False
def write(self, string):
"""
Log the given message.
@param string: Data to write.
@type string: L{bytes} in this file's preferred encoding or L{unicode}
"""
if self._closed:
raise ValueError("I/O operation on closed file")
if isinstance(string, bytes):
string = string.decode(self._encoding)
lines = (self._buffer + string).split("\n")
self._buffer = lines[-1]
lines = lines[0:-1]
for line in lines:
self.log.emit(self.level, format=u"{log_io}", log_io=line)
def writelines(self, lines):
"""
Log each of the given lines as a separate message.
@param lines: Data to write.
@type lines: iterable of L{unicode} or L{bytes} in this file's
declared encoding
"""
for line in lines:
self.write(line)
def _unsupported(self, *args):
"""
Template for unsupported operations.
@param args: Arguments.
@type args: tuple of L{object}
"""
raise IOError("unsupported operation")
read = _unsupported
next = _unsupported
readline = _unsupported
readlines = _unsupported
xreadlines = _unsupported
seek = _unsupported
tell = _unsupported
truncate = _unsupported
|
