def __iter__(self): # type: ignore
# Until https://github.com/python/typing/issues/11
# there's no good way to tell mypy about custom
# iterators that subclass io.IOBase.
"""Let this class act as an iterator."""
return self
python类IOBase()的实例源码
def sanitize_for_serialization(obj):
"""
Sanitize an object for Request.
If obj is None, return None.
If obj is str, int, float, bool, return directly.
If obj is datetime.datetime, datetime.date convert to string in iso8601 format.
If obj is list, santize each element in the list.
If obj is dict, return the dict.
If obj is swagger model, return the properties dict.
"""
if isinstance(obj, type(None)):
return None
elif isinstance(obj, (str, int, float, bool, io.IOBase, tuple)):
return obj
elif isinstance(obj, list):
return [NbClientManager.sanitize_for_serialization(sub_obj) for sub_obj in obj]
elif isinstance(obj, (datetime.datetime, datetime.date)):
return obj.isoformat()
else:
if isinstance(obj, dict):
obj_dict = obj
else:
# Convert model obj to dict except attributes `swaggerTypes`, `attributeMap`
# and attributes which value is not None.
# Convert attribute name to json key in model definition for
# request.
obj_dict = {obj.attributeMap[key]: val
for key, val in obj.__dict__.items()
if key != 'swaggerTypes' and key != 'attributeMap' and val is not None}
return {key: NbClientManager.sanitize_for_serialization(val)
for (key, val) in obj_dict.items()}
def read(obj):
"""Context manager for reading data from multiple sources as a file object
Args:
obj (string|Path|file object): Data to read / read from
If obj is a file object, this is just a pass through
If obj is a Path object, this is similar to obj.open()
If obj is a string, this creates a StringIO so
the data can be read like a file object
Returns:
file object: File handle containing data
"""
try: # Python 2 compatibility
is_unicode = isinstance(obj, unicode)
except NameError:
is_unicode = False
is_open = False
if isinstance(obj, Path):
fh = obj.open()
is_open = True
elif isinstance(obj, str) or is_unicode:
fh = StringIO(obj)
fh.name = '<string>'
elif isinstance(obj, IOBase):
fh = obj
else:
raise Exception("Unknown input type {}".format(type(obj).__name__))
try:
yield fh
finally:
if is_open:
fh.close()
def is_file_like_obj(obj):
"""
Helper function to check that the given object implements all public methods of the
:class:`~io.IOBase` abstract class.
"""
return all((hasattr(obj, name) for name in dir(io.IOBase) if not name.startswith('_')))
def test_scratch_file_supports_file_obj_interface(active_scratch_dir, method_name):
"""
Assert that methods of :class:`~scratchdir.ScratchDir` that are expected to return file-like objects
do so and these objects implement, atleast, the :class:`~io.IOBase` interface.
"""
method = getattr(active_scratch_dir, method_name)
assert is_file_like_obj(method())
def __init__(self, file, start):
assert hasattr(file, 'read') and hasattr(file, 'tell') and hasattr(file, 'seek')
assert isinstance(start, int)
self.file, self.start = file, start
self.total_len = total_len(file)
self.len = self.total_len - start
io.IOBase.__init__(self)
try:
self.seek(0)
except:
pass
def store(self, data, blob):
if not isinstance(data, io.IOBase):
raise NotStorable('Could not store data (not of "file").')
filename = getattr(data, 'name', None)
if filename is not None:
blob.consumeFile(filename)
return
def isFile(self, path=None):
# dirty hack to check where file is opened with codecs module
# (because it returns 'instance' type when encoding is specified
if path:
return isinstance(path, io.IOBase) or path.__class__.__name__ == 'StreamReaderWriter'
else:
return 'file' in self.type
def aclose(self):
"""Like :meth:`io.IOBase.close`, but async.
This is also shielded from cancellation; if a cancellation scope is
cancelled, the wrapped file object will still be safely closed.
"""
# ensure the underling file is closed during cancellation
with _core.open_cancel_scope(shield=True):
await trio.run_sync_in_worker_thread(self._wrapped.close)
await _core.checkpoint_if_cancelled()
def _is_file(f):
return isinstance(f, io.IOBase)
def test_feed_download_output(mock_open, connection, feed, start_time,
end_time, feed_download_url, feed_report_csv):
"""Verifies feed download writing to a file."""
mock_open.return_value = mock.MagicMock(spec=io.IOBase)
httpretty.register_uri(
httpretty.POST,
'{}/feed/{}/prepare'.format(
matchlight.MATCHLIGHT_API_URL_V2,
feed.name),
body=json.dumps({'feed_response_id': 1}),
content_type='application/json', status=200)
httpretty.register_uri(
httpretty.POST,
'{}/feed/{}/link'.format(
matchlight.MATCHLIGHT_API_URL_V2,
feed.name),
responses=[
httpretty.Response(
body=json.dumps({'status': 'pending'}),
content_type='application/json',
status=200),
httpretty.Response(
body=json.dumps({
'status': 'ready',
'url': feed_download_url,
}),
content_type='application/json',
status=200),
],
)
body = '\n'.join(feed_report_csv).encode('utf-8')
httpretty.register_uri(
httpretty.GET, feed_download_url,
content_type='text/csv',
body=body)
connection.feeds.download(
feed, start_time, end_time, save_path='/tmp/output')
file_handle = mock_open.return_value.__enter__.return_value
file_handle.write.assert_called_once_with(body)
def isIOBase(obj):
return isinstance(obj, IOBase)
def _get_file_object(csvfile, encoding=None):
if isinstance(csvfile, str):
assert encoding, 'encoding required for file path'
return open(csvfile, 'rt', encoding=encoding, newline='') # <- EXIT!
if hasattr(csvfile, 'mode'):
assert 'b' not in csvfile.mode, "File must be open in text mode ('rt')."
elif issubclass(csvfile.__class__, io.IOBase):
assert issubclass(csvfile.__class__, io.TextIOBase), ("Stream object must inherit "
"from io.TextIOBase.")
return csvfile
def _py2_get_file_object(csvfile, encoding):
if isinstance(csvfile, str):
return open(csvfile, 'rb') # <- EXIT!
if hasattr(csvfile, 'mode'):
assert 'b' in csvfile.mode, ("When using Python 2, file must "
"be open in binary mode ('rb').")
elif issubclass(csvfile.__class__, io.IOBase):
assert not issubclass(csvfile.__class__, io.TextIOBase), ("When using Python 2, "
"must use byte stream "
"(not text stream).")
return csvfile
def from_csv(cls, file, encoding=None, **fmtparams):
"""Create a DataSource from a CSV *file* (a path or file-like
object)::
source = datatest.DataSource.from_csv('mydata.csv')
If *file* is an iterable of files, data will be loaded and
aligned by column name::
files = ['mydata1.csv', 'mydata2.csv']
source = datatest.DataSource.from_csv(files)
"""
if isinstance(file, string_types) or isinstance(file, IOBase):
file = [file]
new_cls = cls.__new__(cls)
temptable = _from_csv(file, encoding, **fmtparams)
new_cls._connection = temptable.connection
new_cls._table = temptable.name
repr_string = '{0}.from_csv({1}{2}{3})'.format(
new_cls.__class__.__name__,
repr(file[0]) if len(file) == 1 else repr(file),
', {0!r}'.format(encoding) if encoding else '',
', **{0!r}'.format(fmtparams) if fmtparams else '',
)
new_cls._repr_string = repr_string
return new_cls
def _is_file(f):
return isinstance(f, io.IOBase)
def file_output(file_object):
"""
Writes strings to a file, making sure there's a newline at the end
:param:
- `file_object`: opened, writable file or name of file to open
"""
from io import IOBase
if not isinstance(file_object, IOBase):
file_object = open(file_object, WRITEABLE)
while True:
line = (yield)
line = line.rstrip(NEWLINE) + NEWLINE
file_object.write(line)
def _is_file(f):
return isinstance(f, io.IOBase)
def _is_file(f):
return isinstance(f, io.IOBase)
def assemble_file(self, real_file):
"""
File is about to be closed, last chance to append data.
@param real_file: Actual output stream.
@type real_file: C{io.IOBase}
"""
real_file.write(self.file.getvalue())
