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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
|
"""Manage other filesystems as a folder hierarchy.
"""
from __future__ import absolute_import, print_function, unicode_literals
import typing
from . import errors
from .base import FS
from .memoryfs import MemoryFS
from .mode import validate_open_mode, validate_openbin_mode
from .path import abspath, forcedir, normpath
if typing.TYPE_CHECKING:
from typing import (
IO,
Any,
BinaryIO,
Collection,
Iterator,
List,
MutableSequence,
Optional,
Text,
Tuple,
Union,
)
from .enums import ResourceType
from .info import Info, RawInfo
from .permissions import Permissions
from .subfs import SubFS
M = typing.TypeVar("M", bound="MountFS")
class MountError(Exception):
"""Thrown when mounts conflict."""
class MountFS(FS):
"""A virtual filesystem that maps directories on to other file-systems."""
_meta = {
"virtual": True,
"read_only": False,
"unicode_paths": True,
"case_insensitive": False,
"invalid_path_chars": "\0",
}
def __init__(self, auto_close=True):
# type: (bool) -> None
"""Create a new `MountFS` instance.
Arguments:
auto_close (bool): If `True` (the default), the child
filesystems will be closed when `MountFS` is closed.
"""
super(MountFS, self).__init__()
self.auto_close = auto_close
self.default_fs = MemoryFS() # type: FS
self.mounts = [] # type: MutableSequence[Tuple[Text, FS]]
def __repr__(self):
# type: () -> str
return "MountFS(auto_close={!r})".format(self.auto_close)
def __str__(self):
# type: () -> str
return "<mountfs>"
def _delegate(self, path):
# type: (Text) -> Tuple[FS, Text]
"""Get the delegate FS for a given path.
Arguments:
path (str): A path.
Returns:
(FS, str): a tuple of ``(<fs>, <path>)`` for a mounted filesystem,
or ``(None, None)`` if no filesystem is mounted on the
given ``path``.
"""
_path = forcedir(abspath(normpath(path)))
is_mounted = _path.startswith
for mount_path, fs in self.mounts:
if is_mounted(mount_path):
return fs, _path[len(mount_path) :].rstrip("/")
return self.default_fs, path
def mount(self, path, fs):
# type: (Text, Union[FS, Text]) -> None
"""Mounts a host FS object on a given path.
Arguments:
path (str): A path within the MountFS.
fs (FS or str): A filesystem (instance or URL) to mount.
"""
if isinstance(fs, str):
from .opener import open_fs
fs = open_fs(fs)
if not isinstance(fs, FS):
raise TypeError("fs argument must be an FS object or a FS URL")
if fs is self:
raise ValueError("Unable to mount self")
_path = forcedir(abspath(normpath(path)))
for mount_path, _ in self.mounts:
if _path.startswith(mount_path):
raise MountError("mount point overlaps existing mount")
self.mounts.append((_path, fs))
self.default_fs.makedirs(_path, recreate=True)
def close(self):
# type: () -> None
# Explicitly closes children if requested
if self.auto_close:
for _path, fs in self.mounts:
fs.close()
del self.mounts[:]
self.default_fs.close()
super(MountFS, self).close()
def desc(self, path):
# type: (Text) -> Text
if not self.exists(path):
raise errors.ResourceNotFound(path)
fs, delegate_path = self._delegate(path)
if fs is self.default_fs:
fs = self
return "{path} on {fs}".format(fs=fs, path=delegate_path)
def getinfo(self, path, namespaces=None):
# type: (Text, Optional[Collection[Text]]) -> Info
self.check()
fs, _path = self._delegate(path)
return fs.getinfo(_path, namespaces=namespaces)
def listdir(self, path):
# type: (Text) -> List[Text]
self.check()
fs, _path = self._delegate(path)
return fs.listdir(_path)
def makedir(self, path, permissions=None, recreate=False):
# type: (Text, Optional[Permissions], bool) -> SubFS[FS]
self.check()
fs, _path = self._delegate(path)
return fs.makedir(_path, permissions=permissions, recreate=recreate)
def openbin(self, path, mode="r", buffering=-1, **kwargs):
# type: (Text, Text, int, **Any) -> BinaryIO
validate_openbin_mode(mode)
self.check()
fs, _path = self._delegate(path)
return fs.openbin(_path, mode=mode, buffering=-1, **kwargs)
def remove(self, path):
# type: (Text) -> None
self.check()
fs, _path = self._delegate(path)
return fs.remove(_path)
def removedir(self, path):
# type: (Text) -> None
self.check()
path = normpath(path)
if path in ("", "/"):
raise errors.RemoveRootError(path)
fs, _path = self._delegate(path)
return fs.removedir(_path)
def readbytes(self, path):
# type: (Text) -> bytes
self.check()
fs, _path = self._delegate(path)
return fs.readbytes(_path)
def download(self, path, file, chunk_size=None, **options):
# type: (Text, BinaryIO, Optional[int], **Any) -> None
fs, _path = self._delegate(path)
return fs.download(_path, file, chunk_size=chunk_size, **options)
def readtext(
self,
path, # type: Text
encoding=None, # type: Optional[Text]
errors=None, # type: Optional[Text]
newline="", # type: Text
):
# type: (...) -> Text
self.check()
fs, _path = self._delegate(path)
return fs.readtext(_path, encoding=encoding, errors=errors, newline=newline)
def getsize(self, path):
# type: (Text) -> int
self.check()
fs, _path = self._delegate(path)
return fs.getsize(_path)
def getsyspath(self, path):
# type: (Text) -> Text
self.check()
fs, _path = self._delegate(path)
return fs.getsyspath(_path)
def gettype(self, path):
# type: (Text) -> ResourceType
self.check()
fs, _path = self._delegate(path)
return fs.gettype(_path)
def geturl(self, path, purpose="download"):
# type: (Text, Text) -> Text
self.check()
fs, _path = self._delegate(path)
return fs.geturl(_path, purpose=purpose)
def hasurl(self, path, purpose="download"):
# type: (Text, Text) -> bool
self.check()
fs, _path = self._delegate(path)
return fs.hasurl(_path, purpose=purpose)
def isdir(self, path):
# type: (Text) -> bool
self.check()
fs, _path = self._delegate(path)
return fs.isdir(_path)
def isfile(self, path):
# type: (Text) -> bool
self.check()
fs, _path = self._delegate(path)
return fs.isfile(_path)
def scandir(
self,
path, # type: Text
namespaces=None, # type: Optional[Collection[Text]]
page=None, # type: Optional[Tuple[int, int]]
):
# type: (...) -> Iterator[Info]
self.check()
fs, _path = self._delegate(path)
return fs.scandir(_path, namespaces=namespaces, page=page)
def setinfo(self, path, info):
# type: (Text, RawInfo) -> None
self.check()
fs, _path = self._delegate(path)
return fs.setinfo(_path, info)
def validatepath(self, path):
# type: (Text) -> Text
self.check()
fs, _path = self._delegate(path)
fs.validatepath(_path)
path = abspath(normpath(path))
return path
def open(
self,
path, # type: Text
mode="r", # type: Text
buffering=-1, # type: int
encoding=None, # type: Optional[Text]
errors=None, # type: Optional[Text]
newline="", # type: Text
**options # type: Any
):
# type: (...) -> IO
validate_open_mode(mode)
self.check()
fs, _path = self._delegate(path)
return fs.open(
_path,
mode=mode,
buffering=buffering,
encoding=encoding,
errors=errors,
newline=newline,
**options
)
def upload(self, path, file, chunk_size=None, **options):
# type: (Text, BinaryIO, Optional[int], **Any) -> None
self.check()
fs, _path = self._delegate(path)
return fs.upload(_path, file, chunk_size=chunk_size, **options)
def writebytes(self, path, contents):
# type: (Text, bytes) -> None
self.check()
fs, _path = self._delegate(path)
return fs.writebytes(_path, contents)
def writetext(
self,
path, # type: Text
contents, # type: Text
encoding="utf-8", # type: Text
errors=None, # type: Optional[Text]
newline="", # type: Text
):
# type: (...) -> None
fs, _path = self._delegate(path)
return fs.writetext(
_path, contents, encoding=encoding, errors=errors, newline=newline
)
|
