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
|
"""Miscellaneous tools for operating on filesystems.
"""
from __future__ import print_function, unicode_literals
import typing
from . import errors
from .errors import DirectoryNotEmpty, ResourceNotFound
from .path import abspath, dirname, normpath, recursepath
if typing.TYPE_CHECKING:
from typing import IO, List, Optional, Text, Union
from .base import FS
def remove_empty(fs, path):
# type: (FS, Text) -> None
"""Remove all empty parents.
Arguments:
fs (FS): A filesystem instance.
path (str): Path to a directory on the filesystem.
"""
path = abspath(normpath(path))
try:
while path not in ("", "/"):
fs.removedir(path)
path = dirname(path)
except DirectoryNotEmpty:
pass
def copy_file_data(src_file, dst_file, chunk_size=None):
# type: (IO, IO, Optional[int]) -> None
"""Copy data from one file object to another.
Arguments:
src_file (io.IOBase): File open for reading.
dst_file (io.IOBase): File open for writing.
chunk_size (int): Number of bytes to copy at
a time (or `None` to use sensible default).
"""
_chunk_size = 1024 * 1024 if chunk_size is None else chunk_size
read = src_file.read
write = dst_file.write
# The 'or None' is so that it works with binary and text files
for chunk in iter(
lambda: read(_chunk_size) or None, None
): # type: Optional[Union[bytes, str]]
write(chunk)
def get_intermediate_dirs(fs, dir_path):
# type: (FS, Text) -> List[Text]
"""Get a list of non-existing intermediate directories.
Arguments:
fs (FS): A filesystem instance.
dir_path (str): A path to a new directory on the filesystem.
Returns:
list: A list of non-existing paths.
Raises:
~fs.errors.DirectoryExpected: If a path component
references a file and not a directory.
"""
intermediates = []
with fs.lock():
for path in recursepath(abspath(dir_path), reverse=True):
try:
resource = fs.getinfo(path)
except ResourceNotFound:
intermediates.append(abspath(path))
else:
if resource.is_dir:
break
raise errors.DirectoryExpected(dir_path)
return intermediates[::-1][:-1]
def is_thread_safe(*filesystems):
# type: (FS) -> bool
"""Check if all filesystems are thread-safe.
Arguments:
filesystems (FS): Filesystems instances to check.
Returns:
bool: if all filesystems are thread safe.
"""
return all(fs.getmeta().get("thread_safe", False) for fs in filesystems)
|
