Spaces:
Running
Running
| import collections | |
| import zipfile | |
| import pathlib | |
| from . import abc | |
| def remove_duplicates(items): | |
| return iter(collections.OrderedDict.fromkeys(items)) | |
| class FileReader(abc.TraversableResources): | |
| def __init__(self, loader): | |
| self.path = pathlib.Path(loader.path).parent | |
| def resource_path(self, resource): | |
| """ | |
| Return the file system path to prevent | |
| `resources.path()` from creating a temporary | |
| copy. | |
| """ | |
| return str(self.path.joinpath(resource)) | |
| def files(self): | |
| return self.path | |
| class ZipReader(abc.TraversableResources): | |
| def __init__(self, loader, module): | |
| _, _, name = module.rpartition('.') | |
| self.prefix = loader.prefix.replace('\\', '/') + name + '/' | |
| self.archive = loader.archive | |
| def open_resource(self, resource): | |
| try: | |
| return super().open_resource(resource) | |
| except KeyError as exc: | |
| raise FileNotFoundError(exc.args[0]) | |
| def is_resource(self, path): | |
| # workaround for `zipfile.Path.is_file` returning true | |
| # for non-existent paths. | |
| target = self.files().joinpath(path) | |
| return target.is_file() and target.exists() | |
| def files(self): | |
| return zipfile.Path(self.archive, self.prefix) | |
| class MultiplexedPath(abc.Traversable): | |
| """ | |
| Given a series of Traversable objects, implement a merged | |
| version of the interface across all objects. Useful for | |
| namespace packages which may be multihomed at a single | |
| name. | |
| """ | |
| def __init__(self, *paths): | |
| self._paths = list(map(pathlib.Path, remove_duplicates(paths))) | |
| if not self._paths: | |
| message = 'MultiplexedPath must contain at least one path' | |
| raise FileNotFoundError(message) | |
| if not all(path.is_dir() for path in self._paths): | |
| raise NotADirectoryError('MultiplexedPath only supports directories') | |
| def iterdir(self): | |
| visited = [] | |
| for path in self._paths: | |
| for file in path.iterdir(): | |
| if file.name in visited: | |
| continue | |
| visited.append(file.name) | |
| yield file | |
| def read_bytes(self): | |
| raise FileNotFoundError(f'{self} is not a file') | |
| def read_text(self, *args, **kwargs): | |
| raise FileNotFoundError(f'{self} is not a file') | |
| def is_dir(self): | |
| return True | |
| def is_file(self): | |
| return False | |
| def joinpath(self, child): | |
| # first try to find child in current paths | |
| for file in self.iterdir(): | |
| if file.name == child: | |
| return file | |
| # if it does not exist, construct it with the first path | |
| return self._paths[0] / child | |
| __truediv__ = joinpath | |
| def open(self, *args, **kwargs): | |
| raise FileNotFoundError(f'{self} is not a file') | |
| def name(self): | |
| return self._paths[0].name | |
| def __repr__(self): | |
| paths = ', '.join(f"'{path}'" for path in self._paths) | |
| return f'MultiplexedPath({paths})' | |
| class NamespaceReader(abc.TraversableResources): | |
| def __init__(self, namespace_path): | |
| if 'NamespacePath' not in str(namespace_path): | |
| raise ValueError('Invalid path') | |
| self.path = MultiplexedPath(*list(namespace_path)) | |
| def resource_path(self, resource): | |
| """ | |
| Return the file system path to prevent | |
| `resources.path()` from creating a temporary | |
| copy. | |
| """ | |
| return str(self.path.joinpath(resource)) | |
| def files(self): | |
| return self.path | |