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
|
"""Tools for managing OS errors.
"""
from __future__ import print_function, unicode_literals
import sys
import typing
import errno
import platform
from contextlib import contextmanager
from . import errors
if typing.TYPE_CHECKING:
from typing import Iterator, Optional, Text, Type, Union
from types import TracebackType
try:
from collections.abc import Mapping
except ImportError:
from collections import Mapping # noqa: E811
_WINDOWS_PLATFORM = platform.system() == "Windows"
class _ConvertOSErrors(object):
"""Context manager to convert OSErrors in to FS Errors."""
FILE_ERRORS = {
64: errors.RemoteConnectionError, # ENONET
errno.EACCES: errors.PermissionDenied,
errno.ENOENT: errors.ResourceNotFound,
errno.EFAULT: errors.ResourceNotFound,
errno.ESRCH: errors.ResourceNotFound,
errno.ENOTEMPTY: errors.DirectoryNotEmpty,
errno.EEXIST: errors.FileExists,
183: errors.DirectoryExists,
# errno.ENOTDIR: errors.DirectoryExpected,
errno.ENOTDIR: errors.ResourceNotFound,
errno.EISDIR: errors.FileExpected,
errno.EINVAL: errors.FileExpected,
errno.ENOSPC: errors.InsufficientStorage,
errno.EPERM: errors.PermissionDenied,
errno.ENETDOWN: errors.RemoteConnectionError,
errno.ECONNRESET: errors.RemoteConnectionError,
errno.ENAMETOOLONG: errors.PathError,
errno.EOPNOTSUPP: errors.Unsupported,
errno.ENOSYS: errors.Unsupported,
}
DIR_ERRORS = FILE_ERRORS.copy()
DIR_ERRORS[errno.ENOTDIR] = errors.DirectoryExpected
DIR_ERRORS[errno.EEXIST] = errors.DirectoryExists
DIR_ERRORS[errno.EINVAL] = errors.DirectoryExpected
if _WINDOWS_PLATFORM: # pragma: no cover
DIR_ERRORS[13] = errors.DirectoryExpected
DIR_ERRORS[267] = errors.DirectoryExpected
FILE_ERRORS[13] = errors.FileExpected
def __init__(self, opname, path, directory=False):
# type: (Text, Text, bool) -> None
self._opname = opname
self._path = path
self._directory = directory
def __enter__(self):
# type: () -> _ConvertOSErrors
return self
def __exit__(
self,
exc_type, # type: Optional[Type[BaseException]]
exc_value, # type: Optional[BaseException]
traceback, # type: Optional[TracebackType]
):
# type: (...) -> None
os_errors = self.DIR_ERRORS if self._directory else self.FILE_ERRORS
if exc_type and isinstance(exc_value, EnvironmentError):
_errno = exc_value.errno
fserror = os_errors.get(_errno, errors.OperationFailed)
if _errno == errno.EACCES and sys.platform == "win32":
if getattr(exc_value, "args", None) == 32: # pragma: no cover
fserror = errors.ResourceLocked
raise fserror(self._path, exc=exc_value).with_traceback(traceback)
# Stops linter complaining about invalid class name
convert_os_errors = _ConvertOSErrors
@contextmanager
def unwrap_errors(path_replace):
# type: (Union[Text, Mapping[Text, Text]]) -> Iterator[None]
"""Get a context to map OS errors to their `fs.errors` counterpart.
The context will re-write the paths in resource exceptions to be
in the same context as the wrapped filesystem.
The only parameter may be the path from the parent, if only one path
is to be unwrapped. Or it may be a dictionary that maps wrapped
paths on to unwrapped paths.
"""
try:
yield
except errors.ResourceError as e:
if hasattr(e, "path"):
if isinstance(path_replace, Mapping):
e.path = path_replace.get(e.path, e.path)
else:
e.path = path_replace
raise
|
