1import collections
2import pathlib
3import operator
4
5from . import abc
6
7from ._itertools import unique_everseen
8from ._compat import ZipPath
9
10
11def remove_duplicates(items):
12 return iter(collections.OrderedDict.fromkeys(items))
13
14
15class FileReader(abc.TraversableResources):
16 def __init__(self, loader):
17 self.path = pathlib.Path(loader.path).parent
18
19 def resource_path(self, resource):
20 """
21 Return the file system path to prevent
22 `resources.path()` from creating a temporary
23 copy.
24 """
25 return str(self.path.joinpath(resource))
26
27 def files(self):
28 return self.path
29
30
31class ZipReader(abc.TraversableResources):
32 def __init__(self, loader, module):
33 _, _, name = module.rpartition('.')
34 self.prefix = loader.prefix.replace('\\', '/') + name + '/'
35 self.archive = loader.archive
36
37 def open_resource(self, resource):
38 try:
39 return super().open_resource(resource)
40 except KeyError as exc:
41 raise FileNotFoundError(exc.args[0])
42
43 def is_resource(self, path):
44 # workaround for `zipfile.Path.is_file` returning true
45 # for non-existent paths.
46 target = self.files().joinpath(path)
47 return target.is_file() and target.exists()
48
49 def files(self):
50 return ZipPath(self.archive, self.prefix)
51
52
53class MultiplexedPath(abc.Traversable):
54 """
55 Given a series of Traversable objects, implement a merged
56 version of the interface across all objects. Useful for
57 namespace packages which may be multihomed at a single
58 name.
59 """
60
61 def __init__(self, *paths):
62 self._paths = list(map(pathlib.Path, remove_duplicates(paths)))
63 if not self._paths:
64 message = 'MultiplexedPath must contain at least one path'
65 raise FileNotFoundError(message)
66 if not all(path.is_dir() for path in self._paths):
67 raise NotADirectoryError('MultiplexedPath only supports directories')
68
69 def iterdir(self):
70 files = (file for path in self._paths for file in path.iterdir())
71 return unique_everseen(files, key=operator.attrgetter('name'))
72
73 def read_bytes(self):
74 raise FileNotFoundError(f'{self} is not a file')
75
76 def read_text(self, *args, **kwargs):
77 raise FileNotFoundError(f'{self} is not a file')
78
79 def is_dir(self):
80 return True
81
82 def is_file(self):
83 return False
84
85 def joinpath(self, *descendants):
86 try:
87 return super().joinpath(*descendants)
88 except abc.TraversalError:
89 # One of the paths did not resolve (a directory does not exist).
90 # Just return something that will not exist.
91 return self._paths[0].joinpath(*descendants)
92
93 def open(self, *args, **kwargs):
94 raise FileNotFoundError(f'{self} is not a file')
95
96 @property
97 def name(self):
98 return self._paths[0].name
99
100 def __repr__(self):
101 paths = ', '.join(f"'{path}'" for path in self._paths)
102 return f'MultiplexedPath({paths})'
103
104
105class NamespaceReader(abc.TraversableResources):
106 def __init__(self, namespace_path):
107 if 'NamespacePath' not in str(namespace_path):
108 raise ValueError('Invalid path')
109 self.path = MultiplexedPath(*list(namespace_path))
110
111 def resource_path(self, resource):
112 """
113 Return the file system path to prevent
114 `resources.path()` from creating a temporary
115 copy.
116 """
117 return str(self.path.joinpath(resource))
118
119 def files(self):
120 return self.path