-
-
Notifications
You must be signed in to change notification settings - Fork 44
/
_adapters.py
109 lines (84 loc) · 3.11 KB
/
_adapters.py
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
from contextlib import suppress
from . import abc
class SpecLoaderAdapter:
"""
Adapt a package spec to adapt the underlying loader.
"""
def __init__(self, spec, adapter=lambda spec: spec.loader):
self.spec = spec
self.loader = adapter(spec)
def __getattr__(self, name):
return getattr(self.spec, name)
class TraversableResourcesLoader:
"""
Adapt a loader to provide TraversableResources.
"""
def __init__(self, spec):
self.spec = spec
def get_resource_reader(self, name):
return CompatibilityFiles(self.spec)._native()
class CompatibilityFiles:
"""
Adapter for an existing or non-existant resource reader
to provide a compability .files().
"""
class Path(abc.Traversable):
def __init__(self, spec, reader, name=None):
if not spec and not name:
raise ValueError('Need at least a module spec or a resource name')
self._spec = spec
self._reader = reader
self._name = name
def iterdir(self):
if not self._spec or not self._reader:
return iter(())
return iter(
[
CompatibilityFiles.Path(self._spec, self._reader, path)
for path in self._reader.contents()
]
)
def is_dir(self):
raise FileNotFoundError("Can't access resource {}".format(self.name))
is_file = exists = is_dir # type: ignore
def joinpath(self, other):
if not self._spec:
raise FileNotFoundError(
'Orphan resource {} has no sub-resources'.format(other)
)
if not self._reader or other not in list(self._reader.contents()):
raise FileNotFoundError(
'Module {} does not have resource {}'.format(self._spec.name, other)
)
return CompatibilityFiles.Path(self._spec, self._reader, other)
@property
def name(self):
return self._name if self._name is not None else self._spec.name
def open(self):
if self._name is None:
raise FileNotFoundError(
"Can't open module {} as a resource".format(self._spec.name)
)
return self._reader.open_resource(self.name)
def __init__(self, spec):
self.spec = spec
@property
def _reader(self):
with suppress(AttributeError):
return self.spec.loader.get_resource_reader(self.spec.name)
def _native(self):
"""
Return the native reader if it supports files().
"""
reader = self._reader
return reader if hasattr(reader, 'files') else self
def __getattr__(self, attr):
return getattr(self._reader, attr)
def files(self):
return CompatibilityFiles.Path(self.spec, self._reader)
def wrap_spec(package):
"""
Construct a package spec with traversable compatibility
on the spec/loader/reader.
"""
return SpecLoaderAdapter(package.__spec__, TraversableResourcesLoader)