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
|
try:
from errno import EBADF
except ImportError:
EBADF = 9
from io import TextIOWrapper
class cancel_wait_ex(IOError):
def __init__(self):
super(cancel_wait_ex, self).__init__(
EBADF, 'File descriptor was closed in another greenlet')
class FileObjectClosed(IOError):
def __init__(self):
super(FileObjectClosed, self).__init__(
EBADF, 'Bad file descriptor (FileObject was closed)')
class FileObjectBase(object):
"""
Internal base class to ensure a level of consistency
between FileObjectPosix and FileObjectThread
"""
# List of methods we delegate to the wrapping IO object, if they
# implement them and we do not.
_delegate_methods = (
# General methods
'flush',
'fileno',
'writable',
'readable',
'seek',
'seekable',
'tell',
# Read
'read',
'readline',
'readlines',
'read1',
# Write
'write',
'writelines',
'truncate',
)
# Whether we are translating universal newlines or not.
_translate = False
def __init__(self, io, closefd):
"""
:param io: An io.IOBase-like object.
"""
self._io = io
# We don't actually use this property ourself, but we save it (and
# pass it along) for compatibility.
self._close = closefd
if self._translate:
# This automatically handles delegation.
self.translate_newlines(None)
else:
self._do_delegate_methods()
io = property(lambda s: s._io,
# Historically we either hand-wrote all the delegation methods
# to use self.io, or we simply used __getattr__ to look them up at
# runtime. This meant people could change the io attribute on the fly
# and it would mostly work (subprocess.py used to do that). We don't recommend
# that, but we still support it.
lambda s, nv: setattr(s, '_io', nv) or s._do_delegate_methods())
def _do_delegate_methods(self):
for meth_name in self._delegate_methods:
meth = getattr(self._io, meth_name, None)
implemented_by_class = hasattr(type(self), meth_name)
if meth and not implemented_by_class:
setattr(self, meth_name, self._wrap_method(meth))
elif hasattr(self, meth_name) and not implemented_by_class:
delattr(self, meth_name)
def _wrap_method(self, method):
"""
Wrap a method we're copying into our dictionary from the underlying
io object to do something special or different, if necessary.
"""
return method
def translate_newlines(self, mode, *text_args, **text_kwargs):
wrapper = TextIOWrapper(self._io, *text_args, **text_kwargs)
if mode:
wrapper.mode = mode
self.io = wrapper
self._translate = True
@property
def closed(self):
"""True if the file is closed"""
return self._io is None
def close(self):
if self._io is None:
return
io = self._io
self._io = None
self._do_close(io, self._close)
def _do_close(self, fobj, closefd):
raise NotImplementedError()
def __getattr__(self, name):
if self._io is None:
raise FileObjectClosed()
return getattr(self._io, name)
def __repr__(self):
return '<%s _fobj=%r%s>' % (self.__class__.__name__, self.io, self._extra_repr())
def _extra_repr(self):
return ''
|
