diff --git a/docs/source/glossary.rst b/docs/source/glossary.rst new file mode 100644 index 0000000000..20ceb1510a --- /dev/null +++ b/docs/source/glossary.rst @@ -0,0 +1,16 @@ +:orphan: + +.. _glossary: + +******** +Glossary +******** + +.. glossary:: + + asynchronous file object + This is an object with an API identical to a :term:`file object`, with + the exception that all non-computational methods are async functions. + + The main way to create an asynchronous file object is by using the + :func:`trio.open_file` function. diff --git a/docs/source/index.rst b/docs/source/index.rst index 7bff4d2a15..3ea681fbdb 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -83,3 +83,4 @@ Vital statistics: * :ref:`genindex` * :ref:`modindex` * :ref:`search` +* :ref:`glossary` diff --git a/docs/source/reference-io.rst b/docs/source/reference-io.rst index c7c3a4df75..66fc348847 100644 --- a/docs/source/reference-io.rst +++ b/docs/source/reference-io.rst @@ -374,12 +374,20 @@ Socket objects * :meth:`~socket.socket.set_inheritable` * :meth:`~socket.socket.get_inheritable` +Asynchronous disk I/O +--------------------- -Async disk I/O --------------- +.. currentmodule:: trio + +.. autofunction:: open_file -`Not implemented yet! `__ +.. autofunction:: wrap_file +Asynchronous path objects +~~~~~~~~~~~~~~~~~~~~~~~~~ + +.. autoclass:: Path + :members: Subprocesses ------------ diff --git a/trio/__init__.py b/trio/__init__.py index 040b2fdf10..be94d9a873 100644 --- a/trio/__init__.py +++ b/trio/__init__.py @@ -50,6 +50,12 @@ from ._network import * __all__ += _network.__all__ +from ._file_io import * +__all__ += _file_io.__all__ + +from ._path import * +__all__ += _path.__all__ + # Imported by default from . import socket from . import abc diff --git a/trio/_file_io.py b/trio/_file_io.py new file mode 100644 index 0000000000..8642c93586 --- /dev/null +++ b/trio/_file_io.py @@ -0,0 +1,168 @@ +from functools import partial +import io + +import trio +from trio import _core +from trio._util import aiter_compat, async_wraps + + +__all__ = ['open_file', 'wrap_file'] + +_FILE_SYNC_ATTRS = { + 'closed', + 'encoding', 'errors', 'fileno', 'isatty', 'newlines', + 'readable', 'seekable', 'writable', + # not defined in *IOBase: + 'buffer', 'raw', 'line_buffering', 'closefd', 'name', 'mode', + 'getvalue', 'getbuffer', +} + +_FILE_ASYNC_METHODS = { + 'flush', + 'read', 'read1', 'readall', 'readinto', 'readline', 'readlines', + 'seek', 'tell', 'truncate', + 'write', 'writelines', + # not defined in *IOBase: + 'readinto1', 'peek', +} + + +class AsyncIOWrapper: + """A generic :class:`~io.IOBase` wrapper that implements the :term:`asynchronous + file object` interface. Wrapped methods that could block are executed in + :meth:`trio.run_in_worker_thread`. + + All properties and methods defined in in :mod:`~io` are exposed by this + wrapper, if they exist in the wrapped file object. + + """ + + def __init__(self, file): + self._wrapped = file + + @property + def wrapped(self): + """object: A reference to the wrapped file object + + """ + + return self._wrapped + + def __getattr__(self, name): + if name in _FILE_SYNC_ATTRS: + return getattr(self._wrapped, name) + if name in _FILE_ASYNC_METHODS: + meth = getattr(self._wrapped, name) + + @async_wraps(self.__class__, self._wrapped.__class__, name) + async def wrapper(*args, **kwargs): + func = partial(meth, *args, **kwargs) + return await trio.run_in_worker_thread(func) + + # cache the generated method + setattr(self, name, wrapper) + return wrapper + + raise AttributeError(name) + + def __dir__(self): + attrs = set(super().__dir__()) + attrs.update(a for a in _FILE_SYNC_ATTRS if hasattr(self.wrapped, a)) + attrs.update(a for a in _FILE_ASYNC_METHODS if hasattr(self.wrapped, a)) + return attrs + + + @aiter_compat + def __aiter__(self): + return self + + async def __anext__(self): + line = await self.readline() + if line: + return line + else: + raise StopAsyncIteration + + async def __aenter__(self): + return self + + async def __aexit__(self, typ, value, traceback): + await self.close() + + async def detach(self): + """Like :meth:`~io.BufferedIOBase.detach`, but async. + + This also re-wraps the result in a new :term:`asynchronous file object` + wrapper. + + """ + + raw = await trio.run_in_worker_thread(self._wrapped.detach) + return wrap_file(raw) + + async def close(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_in_worker_thread(self._wrapped.close) + + await _core.yield_if_cancelled() + + +async def open_file(file, mode='r', buffering=-1, encoding=None, errors=None, + newline=None, closefd=True, opener=None): + """Asynchronous version of :func:`~io.open`. + + Returns: + An :term:`asynchronous file object` + + Example:: + + async with await trio.open_file(filename) as f: + async for line in f: + pass + + assert f.closed + + """ + # python3.5 compat + if isinstance(file, trio.Path): + file = file.__fspath__() + + _file = wrap_file(await trio.run_in_worker_thread(io.open, file, mode, + buffering, encoding, errors, newline, closefd, opener)) + return _file + + +def wrap_file(file): + """This wraps any file object in a wrapper that provides an asynchronous file + object interface. + + Args: + file: a :term:`file object` + + Returns: + An :term:`asynchronous file object` that wraps `file` + + Example:: + + async_file = trio.wrap_file(StringIO('asdf')) + + assert await async_file.read() == 'asdf' + + """ + + def has(attr): + return hasattr(file, attr) and callable(getattr(file, attr)) + + if not (has('close') and (has('read') or has('write'))): + raise TypeError('{} does not implement required duck-file methods: ' + 'close and (read or write)'.format(file)) + + return AsyncIOWrapper(file) diff --git a/trio/_path.py b/trio/_path.py new file mode 100644 index 0000000000..2f88f1d9fa --- /dev/null +++ b/trio/_path.py @@ -0,0 +1,163 @@ +from functools import wraps, partial +import os +import types +import pathlib + +import trio +from trio._util import async_wraps + + +__all__ = ['Path'] + + +# python3.5 compat: __fspath__ does not exist in 3.5, so unwrap any trio.Path +# being passed to any wrapped method +def unwrap_paths(args): + new_args = [] + for arg in args: + if isinstance(arg, Path): + arg = arg._wrapped + new_args.append(arg) + return new_args + + +# re-wrap return value from methods that return new instances of pathlib.Path +def rewrap_path(value): + if isinstance(value, pathlib.Path): + value = Path(value) + return value + + +def _forward_factory(cls, attr_name, attr): + @wraps(attr) + def wrapper(self, *args, **kwargs): + args = unwrap_paths(args) + attr = getattr(self._wrapped, attr_name) + value = attr(*args, **kwargs) + return rewrap_path(value) + + return wrapper + + +def _forward_magic(cls, attr): + sentinel = object() + @wraps(attr) + def wrapper(self, other=sentinel): + if other is sentinel: + return attr(self._wrapped) + if isinstance(other, cls): + other = other._wrapped + value = attr(self._wrapped, other) + return rewrap_path(value) + return wrapper + + +def thread_wrapper_factory(cls, meth_name): + @async_wraps(cls, pathlib.Path, meth_name) + async def wrapper(self, *args, **kwargs): + args = unwrap_paths(args) + meth = getattr(self._wrapped, meth_name) + func = partial(meth, *args, **kwargs) + value = await trio.run_in_worker_thread(func) + return rewrap_path(value) + + return wrapper + + +class AsyncAutoWrapperType(type): + def __init__(cls, name, bases, attrs): + super().__init__(name, bases, attrs) + + cls._forward = [] + type(cls).generate_forwards(cls, attrs) + type(cls).generate_wraps(cls, attrs) + type(cls).generate_magic(cls, attrs) + + def generate_forwards(cls, attrs): + # forward functions of _forwards + for attr_name, attr in cls._forwards.__dict__.items(): + if attr_name.startswith('_') or attr_name in attrs: + continue + + if isinstance(attr, property): + cls._forward.append(attr_name) + elif isinstance(attr, types.FunctionType): + wrapper = _forward_factory(cls, attr_name, attr) + setattr(cls, attr_name, wrapper) + else: + raise TypeError(attr_name, type(attr)) + + def generate_wraps(cls, attrs): + # generate wrappers for functions of _wraps + for attr_name, attr in cls._wraps.__dict__.items(): + if attr_name.startswith('_') or attr_name in attrs: + continue + + if isinstance(attr, classmethod): + setattr(cls, attr_name, attr) + elif isinstance(attr, types.FunctionType): + wrapper = thread_wrapper_factory(cls, attr_name) + setattr(cls, attr_name, wrapper) + else: + raise TypeError(attr_name, type(attr)) + + def generate_magic(cls, attrs): + # generate wrappers for magic + for attr_name in cls._forward_magic: + attr = getattr(cls._forwards, attr_name) + wrapper = _forward_magic(cls, attr) + setattr(cls, attr_name, wrapper) + + +class Path(metaclass=AsyncAutoWrapperType): + """A :class:`~pathlib.Path` wrapper that executes blocking Path methods in + :meth:`trio.run_in_worker_thread`. + + """ + + _wraps = pathlib.Path + _forwards = pathlib.PurePath + _forward_magic = [ + '__str__', '__bytes__', '__truediv__', '__rtruediv__', + '__eq__', '__lt__', '__le__', '__gt__', '__ge__' + ] + + def __init__(self, *args): + args = unwrap_paths(args) + + self._wrapped = pathlib.Path(*args) + + def __getattr__(self, name): + if name in self._forward: + value = getattr(self._wrapped, name) + return rewrap_path(value) + raise AttributeError(name) + + def __dir__(self): + return super().__dir__() + self._forward + + def __repr__(self): + return 'trio.Path({})'.format(str(self)) + + def __fspath__(self): + try: + return self._wrapped.__fspath__() + # python3.5 compat + except AttributeError: # pragma: no cover + return str(self) + + async def open(self, *args, **kwargs): + func = partial(self._wrapped.open, *args, **kwargs) + value = await trio.run_in_worker_thread(func) + return trio.wrap_file(value) + + +# The value of Path.absolute.__doc__ makes a reference to +# :meth:~pathlib.Path.absolute, which does not exist. Removing this makes more +# sense than inventing our own special docstring for this. +del Path.absolute.__doc__ + + +# python3.5 compat +if hasattr(os, 'PathLike'): + os.PathLike.register(Path) diff --git a/trio/_util.py b/trio/_util.py index 0986e02440..7b3692b697 100644 --- a/trio/_util.py +++ b/trio/_util.py @@ -188,3 +188,22 @@ async def __aenter__(self): async def __aexit__(self, *args): return self.sync.__exit__() + + +def async_wraps(cls, wrapped_cls, attr_name): + """Similar to wraps, but for async wrappers of non-async functions. + + """ + def decorator(func): + func.__name__ = attr_name + func.__qualname__ = '.'.join((cls.__qualname__, + attr_name)) + + func.__doc__ = """Like :meth:`~{}.{}.{}`, but async. + + """.format(wrapped_cls.__module__, + wrapped_cls.__qualname__, + attr_name) + + return func + return decorator diff --git a/trio/tests/test_file_io.py b/trio/tests/test_file_io.py new file mode 100644 index 0000000000..3401229984 --- /dev/null +++ b/trio/tests/test_file_io.py @@ -0,0 +1,194 @@ +import io +import tempfile + +import pytest +from unittest import mock +from unittest.mock import sentinel + +import trio +from trio import _core +from trio._file_io import AsyncIOWrapper, _FILE_SYNC_ATTRS, _FILE_ASYNC_METHODS + + +@pytest.fixture +def path(tmpdir): + return tmpdir.join('test').__fspath__() + + +@pytest.fixture +def wrapped(): + return mock.Mock(spec_set=io.StringIO) + + +@pytest.fixture +def async_file(wrapped): + return trio.wrap_file(wrapped) + + +def test_wrap_invalid(): + with pytest.raises(TypeError): + trio.wrap_file(str()) + + +def test_wrap_non_iobase(): + class FakeFile: + def close(self): # pragma: no cover + pass + + def write(self): # pragma: no cover + pass + + wrapped = FakeFile() + assert not isinstance(wrapped, io.IOBase) + + async_file = trio.wrap_file(wrapped) + assert isinstance(async_file, AsyncIOWrapper) + + del FakeFile.write + + with pytest.raises(TypeError): + trio.wrap_file(FakeFile()) + + +def test_wrapped_property(async_file, wrapped): + assert async_file.wrapped is wrapped + + +def test_dir_matches_wrapped(async_file, wrapped): + attrs = _FILE_SYNC_ATTRS.union(_FILE_ASYNC_METHODS) + + # all supported attrs in wrapped should be available in async_file + assert all(attr in dir(async_file) for attr in attrs if attr in dir(wrapped)) + # all supported attrs not in wrapped should not be available in async_file + assert not any(attr in dir(async_file) for attr in attrs if attr not in dir(wrapped)) + + +def test_unsupported_not_forwarded(): + class FakeFile(io.RawIOBase): + def unsupported_attr(self): # pragma: no cover + pass + + async_file = trio.wrap_file(FakeFile()) + + assert hasattr(async_file.wrapped, 'unsupported_attr') + + with pytest.raises(AttributeError): + getattr(async_file, 'unsupported_attr') + + +def test_sync_attrs_forwarded(async_file, wrapped): + for attr_name in _FILE_SYNC_ATTRS: + if attr_name not in dir(async_file): + continue + + assert getattr(async_file, attr_name) is getattr(wrapped, attr_name) + + +def test_sync_attrs_match_wrapper(async_file, wrapped): + for attr_name in _FILE_SYNC_ATTRS: + if attr_name in dir(async_file): + continue + + with pytest.raises(AttributeError): + getattr(async_file, attr_name) + + with pytest.raises(AttributeError): + getattr(wrapped, attr_name) + + +def test_async_methods_generated_once(async_file): + for meth_name in _FILE_ASYNC_METHODS: + if meth_name not in dir(async_file): + continue + + assert getattr(async_file, meth_name) is getattr(async_file, meth_name) + + +def test_async_methods_signature(async_file): + # use read as a representative of all async methods + assert async_file.read.__name__ == 'read' + assert async_file.read.__qualname__ == 'AsyncIOWrapper.read' + + assert 'io.StringIO.read' in async_file.read.__doc__ + + +async def test_async_methods_wrap(async_file, wrapped): + for meth_name in _FILE_ASYNC_METHODS: + if meth_name not in dir(async_file): + continue + + meth = getattr(async_file, meth_name) + wrapped_meth = getattr(wrapped, meth_name) + + value = await meth(sentinel.argument, keyword=sentinel.keyword) + + wrapped_meth.assert_called_once_with(sentinel.argument, keyword=sentinel.keyword) + assert value == wrapped_meth() + + wrapped.reset_mock() + + +async def test_async_methods_match_wrapper(async_file, wrapped): + for meth_name in _FILE_ASYNC_METHODS: + if meth_name in dir(async_file): + continue + + with pytest.raises(AttributeError): + getattr(async_file, meth_name) + + with pytest.raises(AttributeError): + getattr(wrapped, meth_name) + + +async def test_open(path): + f = await trio.open_file(path, 'w') + + assert isinstance(f, AsyncIOWrapper) + + await f.close() + + +async def test_open_context_manager(path): + async with await trio.open_file(path, 'w') as f: + assert isinstance(f, AsyncIOWrapper) + assert not f.closed + + assert f.closed + + +async def test_async_iter(): + async_file = trio.wrap_file(io.StringIO('test\nfoo\nbar')) + expected = list(async_file.wrapped) + result = [] + async_file.wrapped.seek(0) + + async for line in async_file: + result.append(line) + + assert result == expected + + +async def test_close_cancelled(path): + with _core.open_cancel_scope() as cscope: + f = await trio.open_file(path, 'w') + cscope.cancel() + + with pytest.raises(_core.Cancelled): + await f.write('a') + + with pytest.raises(_core.Cancelled): + await f.close() + + assert f.closed + + +async def test_detach_rewraps_asynciobase(): + raw = io.BytesIO() + buffered = io.BufferedReader(raw) + + async_file = trio.wrap_file(buffered) + + detached = await async_file.detach() + + assert isinstance(detached, AsyncIOWrapper) + assert detached.wrapped is raw diff --git a/trio/tests/test_path.py b/trio/tests/test_path.py new file mode 100644 index 0000000000..346c5ace34 --- /dev/null +++ b/trio/tests/test_path.py @@ -0,0 +1,190 @@ +import os +import pathlib + +import pytest + +import trio +from trio._path import AsyncAutoWrapperType as Type +from trio._file_io import AsyncIOWrapper + + +@pytest.fixture +def path(tmpdir): + p = str(tmpdir.join('test')) + return trio.Path(p) + + +def method_pair(path, method_name): + path = pathlib.Path(path) + async_path = trio.Path(path) + return getattr(path, method_name), getattr(async_path, method_name) + + +async def test_open_is_async_context_manager(path): + async with await path.open('w') as f: + assert isinstance(f, AsyncIOWrapper) + + assert f.closed + + +async def test_magic(): + path = trio.Path('test') + + assert str(path) == 'test' + assert bytes(path) == b'test' + + +cls_pairs = [(trio.Path, pathlib.Path), (pathlib.Path, trio.Path), (trio.Path, trio.Path)] + + +@pytest.mark.parametrize('cls_a,cls_b', cls_pairs) +async def test_cmp_magic(cls_a, cls_b): + a, b = cls_a(''), cls_b('') + assert a == b + assert not a != b + + a, b = cls_a('a'), cls_b('b') + assert a < b + assert b > a + + # this is intentionally testing equivalence with none, due to the + # other=sentinel logic in _forward_magic + assert not a == None + assert not b == None + + +# upstream python3.5 bug: we should also test (pathlib.Path, trio.Path), but +# __*div__ does not properly raise NotImplementedError like the other comparison +# magic, so trio.Path's implementation does not get dispatched +cls_pairs = [(trio.Path, pathlib.Path), (trio.Path, trio.Path), (trio.Path, str), (str, trio.Path)] + + +@pytest.mark.parametrize('cls_a,cls_b', cls_pairs) +async def test_div_magic(cls_a, cls_b): + a, b = cls_a('a'), cls_b('b') + + result = a / b + assert isinstance(result, trio.Path) + assert str(result) == os.path.join('a', 'b') + + +async def test_forwarded_properties(path): + # use `name` as a representative of forwarded properties + + assert 'name' in dir(path) + assert path.name == 'test' + + +async def test_async_method_signature(path): + # use `resolve` as a representative of wrapped methods + + assert path.resolve.__name__ == 'resolve' + assert path.resolve.__qualname__ == 'Path.resolve' + + assert 'pathlib.Path.resolve' in path.resolve.__doc__ + + +@pytest.mark.parametrize('method_name', ['is_dir', 'is_file']) +async def test_compare_async_stat_methods(method_name): + + method, async_method = method_pair('.', method_name) + + result = method() + async_result = await async_method() + + assert result == async_result + + +async def test_invalid_name_not_wrapped(path): + with pytest.raises(AttributeError): + getattr(path, 'invalid_fake_attr') + + +@pytest.mark.parametrize('method_name', ['absolute', 'resolve']) +async def test_async_methods_rewrap(method_name): + + method, async_method = method_pair('.', method_name) + + result = method() + async_result = await async_method() + + assert isinstance(async_result, trio.Path) + assert str(result) == str(async_result) + + +async def test_forward_methods_rewrap(path, tmpdir): + with_name = path.with_name('foo') + with_suffix = path.with_suffix('.py') + + assert isinstance(with_name, trio.Path) + assert with_name == tmpdir.join('foo') + assert isinstance(with_suffix, trio.Path) + assert with_suffix == tmpdir.join('test.py') + + +async def test_forward_properties_rewrap(path): + assert isinstance(path.parent, trio.Path) + + +async def test_forward_methods_without_rewrap(path, tmpdir): + path = await path.parent.resolve() + + assert path.as_uri().startswith('file:///') + + +async def test_repr(): + path = trio.Path('.') + + assert repr(path) == 'trio.Path(.)' + + +class MockWrapped: + unsupported = 'unsupported' + _private = 'private' + + +class MockWrapper: + _forwards = MockWrapped + _wraps = MockWrapped + + +async def test_type_forwards_unsupported(): + with pytest.raises(TypeError): + Type.generate_forwards(MockWrapper, {}) + + +async def test_type_wraps_unsupported(): + with pytest.raises(TypeError): + Type.generate_wraps(MockWrapper, {}) + + +async def test_type_forwards_private(): + Type.generate_forwards(MockWrapper, {'unsupported': None}) + + assert not hasattr(MockWrapper, '_private') + + +async def test_type_wraps_private(): + Type.generate_wraps(MockWrapper, {'unsupported': None}) + + assert not hasattr(MockWrapper, '_private') + + +@pytest.mark.parametrize('meth', [trio.Path.__init__, trio.Path.joinpath]) +async def test_path_wraps_path(path, meth): + wrapped = await path.absolute() + result = meth(path, wrapped) + if result is None: + result = path + + assert wrapped == result + + +async def test_path_nonpath(): + with pytest.raises(TypeError): + trio.Path(1) + + +async def test_open_file_can_open_path(path): + async with await trio.open_file(path, 'w') as f: + assert f.name == path.__fspath__()