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)

Read our documentation on viewing source code .

Loading