Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pkg_resources/_vendor/importlib_resources/readers.py: 27%

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