_path.py at ec37d7a ⋅ python-trio/trio

# type: ignore

from functools import wraps, partial
import os
import types
import pathlib

import trio
from trio._util import async_wraps, SubclassingDeprecatedIn_v0_15_0


# 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):
        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 iter_wrapper_factory(cls, meth_name):
    @async_wraps(cls, cls._wraps, meth_name)
    async def wrapper(self, *args, **kwargs):
        meth = getattr(self._wrapped, meth_name)
        func = partial(meth, *args, **kwargs)
        # Make sure that the full iteration is performed in the thread
        # by converting the generator produced by pathlib into a list
        items = await trio.to_thread.run_sync(lambda: list(func()))
        return (rewrap_path(item) for item in items)

    return wrapper


def thread_wrapper_factory(cls, meth_name):
    @async_wraps(cls, cls._wraps, meth_name)
    async def wrapper(self, *args, **kwargs):
        meth = getattr(self._wrapped, meth_name)
        func = partial(meth, *args, **kwargs)
        value = await trio.to_thread.run_sync(func)
        return rewrap_path(value)

    return wrapper


def classmethod_wrapper_factory(cls, meth_name):
    @classmethod
    @async_wraps(cls, cls._wraps, meth_name)
    async def wrapper(cls, *args, **kwargs):
        meth = getattr(cls._wraps, meth_name)
        func = partial(meth, *args, **kwargs)
        value = await trio.to_thread.run_sync(func)
        return rewrap_path(value)

    return wrapper


class AsyncAutoWrapperType(SubclassingDeprecatedIn_v0_15_0):
    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)
        type(cls).generate_iter(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():
            # .z. exclude cls._wrap_iter
            if attr_name.startswith("_") or attr_name in attrs:
                continue
            if isinstance(attr, classmethod):
                wrapper = classmethod_wrapper_factory(cls, attr_name)
                setattr(cls, attr_name, wrapper)
            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)

    def generate_iter(cls, attrs):
        # generate wrappers for methods that return iterators
        for attr_name, attr in cls._wraps.__dict__.items():
            if attr_name in cls._wrap_iter:
                wrapper = iter_wrapper_factory(cls, attr_name)
                setattr(cls, attr_name, wrapper)


class Path(metaclass=AsyncAutoWrapperType):
    """A :class:`pathlib.Path` wrapper that executes blocking methods in
    :meth:`trio.to_thread.run_sync`.

    """

    _wraps = pathlib.Path
    _forwards = pathlib.PurePath
    _forward_magic = [
        "__str__",
        "__bytes__",
        "__truediv__",
        "__rtruediv__",
        "__eq__",
        "__lt__",
        "__le__",
        "__gt__",
        "__ge__",
        "__hash__",
    ]
    _wrap_iter = ["glob", "rglob", "iterdir"]

    def __init__(self, *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(repr(str(self)))

    def __fspath__(self):
        return os.fspath(self._wrapped)

    @wraps(pathlib.Path.open)
    async def open(self, *args, **kwargs):
        """Open the file pointed to by the path, like the :func:`trio.open_file`
        function does.

        """

        func = partial(self._wrapped.open, *args, **kwargs)
        value = await trio.to_thread.run_sync(func)
        return trio.wrap_file(value)


Path.iterdir.__doc__ = """
    Like :meth:`pathlib.Path.iterdir`, but async.

    This is an async method that returns a synchronous iterator, so you
    use it like::

       for subpath in await mypath.iterdir():
           ...

    Note that it actually loads the whole directory list into memory
    immediately, during the initial call. (See `issue #501
    <https://github.com/python-trio/trio/issues/501>`__ for discussion.)

"""

# 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__

os.PathLike.register(Path)

Read our documentation on viewing source code .

Navigation	Overlay
`t` Navigate files	`h` Toggle hits
`y` Change url to tip of branch	`m` Toggle misses
`b / v` Jump to prev/next hit line	`p` Toggle partial
`z / x` Jump to prev/next missed or partial line	`1..9` Toggle flags
`shift + o` Open current page in GitHub	`a` Toggle all on
`/ or ?` Show keyboard shortcuts dialog	`c` Toggle context lines or commits

1		# type: ignore
2
3	22	from functools import wraps, partial
4	22	import os
5	22	import types
6	22	import pathlib
7
8	22	import trio
9	22	from trio._util import async_wraps, SubclassingDeprecatedIn_v0_15_0
10
11
12		# re-wrap return value from methods that return new instances of pathlib.Path
13	22	def rewrap_path(value):
14	22	if isinstance(value, pathlib.Path):
15	22	value = Path(value)
16	22	return value
17
18
19	22	def _forward_factory(cls, attr_name, attr):
20	22	@wraps(attr)
21	9	def wrapper(self, args, *kwargs):
22	22	attr = getattr(self._wrapped, attr_name)
23	22	value = attr(args, *kwargs)
24	22	return rewrap_path(value)
25
26	22	return wrapper
27
28
29	22	def _forward_magic(cls, attr):
30	22	sentinel = object()
31
32	22	@wraps(attr)
33	22	def wrapper(self, other=sentinel):
34	22	if other is sentinel:
35	22	return attr(self._wrapped)
36	22	if isinstance(other, cls):
37	22	other = other._wrapped
38	22	value = attr(self._wrapped, other)
39	22	return rewrap_path(value)
40
41	22	return wrapper
42
43
44	22	def iter_wrapper_factory(cls, meth_name):
45	22	@async_wraps(cls, cls._wraps, meth_name)
46	9	async def wrapper(self, args, *kwargs):
47	22	meth = getattr(self._wrapped, meth_name)
48	22	func = partial(meth, args, *kwargs)
49		# Make sure that the full iteration is performed in the thread
50		# by converting the generator produced by pathlib into a list
51	22	items = await trio.to_thread.run_sync(lambda: list(func()))
52	22	return (rewrap_path(item) for item in items)
53
54	22	return wrapper
55
56
57	22	def thread_wrapper_factory(cls, meth_name):
58	22	@async_wraps(cls, cls._wraps, meth_name)
59	9	async def wrapper(self, args, *kwargs):
60	22	meth = getattr(self._wrapped, meth_name)
61	22	func = partial(meth, args, *kwargs)
62	22	value = await trio.to_thread.run_sync(func)
63	22	return rewrap_path(value)
64
65	22	return wrapper
66
67
68	22	def classmethod_wrapper_factory(cls, meth_name):
69	22	@classmethod
70	22	@async_wraps(cls, cls._wraps, meth_name)
71	9	async def wrapper(cls, args, *kwargs):
72	22	meth = getattr(cls._wraps, meth_name)
73	22	func = partial(meth, args, *kwargs)
74	22	value = await trio.to_thread.run_sync(func)
75	22	return rewrap_path(value)
76
77	22	return wrapper
78
79
80	22	class AsyncAutoWrapperType(SubclassingDeprecatedIn_v0_15_0):
81	22	def __init__(cls, name, bases, attrs):
82	22	super().__init__(name, bases, attrs)
83
84	22	cls._forward = []
85	22	type(cls).generate_forwards(cls, attrs)
86	22	type(cls).generate_wraps(cls, attrs)
87	22	type(cls).generate_magic(cls, attrs)
88	22	type(cls).generate_iter(cls, attrs)
89
90	22	def generate_forwards(cls, attrs):
91		# forward functions of _forwards
92	22	for attr_name, attr in cls._forwards.__dict__.items():
93	22	if attr_name.startswith("_") or attr_name in attrs:
94	22	continue
95
96	22	if isinstance(attr, property):
97	22	cls._forward.append(attr_name)
98	22	elif isinstance(attr, types.FunctionType):
99	22	wrapper = _forward_factory(cls, attr_name, attr)
100	22	setattr(cls, attr_name, wrapper)
101		else:
102	22	raise TypeError(attr_name, type(attr))
103
104	22	def generate_wraps(cls, attrs):
105		# generate wrappers for functions of _wraps
106	22	for attr_name, attr in cls._wraps.__dict__.items():
107		# .z. exclude cls._wrap_iter
108	22	if attr_name.startswith("_") or attr_name in attrs:
109	22	continue
110	22	if isinstance(attr, classmethod):
111	22	wrapper = classmethod_wrapper_factory(cls, attr_name)
112	22	setattr(cls, attr_name, wrapper)
113	22	elif isinstance(attr, types.FunctionType):
114	22	wrapper = thread_wrapper_factory(cls, attr_name)
115	22	setattr(cls, attr_name, wrapper)
116		else:
117	22	raise TypeError(attr_name, type(attr))
118
119	22	def generate_magic(cls, attrs):
120		# generate wrappers for magic
121	22	for attr_name in cls._forward_magic:
122	22	attr = getattr(cls._forwards, attr_name)
123	22	wrapper = _forward_magic(cls, attr)
124	22	setattr(cls, attr_name, wrapper)
125
126	22	def generate_iter(cls, attrs):
127		# generate wrappers for methods that return iterators
128	22	for attr_name, attr in cls._wraps.__dict__.items():
129	22	if attr_name in cls._wrap_iter:
130	22	wrapper = iter_wrapper_factory(cls, attr_name)
131	22	setattr(cls, attr_name, wrapper)
132
133
134	22	class Path(metaclass=AsyncAutoWrapperType):
135		"""A :class:`pathlib.Path` wrapper that executes blocking methods in
136		:meth:`trio.to_thread.run_sync`.
137
138		"""
139
140	22	_wraps = pathlib.Path
141	22	_forwards = pathlib.PurePath
142	22	_forward_magic = [
143		"__str__",
144		"__bytes__",
145		"__truediv__",
146		"__rtruediv__",
147		"__eq__",
148		"__lt__",
149		"__le__",
150		"__gt__",
151		"__ge__",
152		"__hash__",
153		]
154	22	_wrap_iter = ["glob", "rglob", "iterdir"]
155
156	22	def __init__(self, *args):
157	22	self._wrapped = pathlib.Path(*args)
158
159	22	def __getattr__(self, name):
160	22	if name in self._forward:
161	22	value = getattr(self._wrapped, name)
162	22	return rewrap_path(value)
163	22	raise AttributeError(name)
164
165	22	def __dir__(self):
166	22	return super().__dir__() + self._forward
167
168	22	def __repr__(self):
169	22	return "trio.Path({})".format(repr(str(self)))
170
171	22	def __fspath__(self):
172	22	return os.fspath(self._wrapped)
173
174	22	@wraps(pathlib.Path.open)
175	9	async def open(self, args, *kwargs):
176		"""Open the file pointed to by the path, like the :func:`trio.open_file`
177		function does.
178
179		"""
180
181	22	func = partial(self._wrapped.open, args, *kwargs)
182	22	value = await trio.to_thread.run_sync(func)
183	22	return trio.wrap_file(value)
184
185
186	22	Path.iterdir.__doc__ = """
187		Like :meth:`pathlib.Path.iterdir`, but async.
188
189		This is an async method that returns a synchronous iterator, so you
190		use it like::
191
192		for subpath in await mypath.iterdir():
193		...
194
195		Note that it actually loads the whole directory list into memory
196		immediately, during the initial call. (See `issue #501
197		<https://github.com/python-trio/trio/issues/501>`__ for discussion.)
198
199		"""
200
201		# The value of Path.absolute.__doc__ makes a reference to
202		# :meth:~pathlib.Path.absolute, which does not exist. Removing this makes more
203		# sense than inventing our own special docstring for this.
204	22	del Path.absolute.__doc__
205
206	22	os.PathLike.register(Path)