Coverage for python/lsst/resources/packageresource.py: 99%

1# This file is part of lsst-resources. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11 

12from __future__ import annotations 

13 

14__all__ = ("PackageResourcePath",) 

15 

16import contextlib 

17import logging 

18import re 

19from collections.abc import Iterator 

20from importlib import resources 

21from typing import TYPE_CHECKING 

22 

23if TYPE_CHECKING: 

24 try: 

25 import fsspec 

26 from fsspec.spec import AbstractFileSystem 

27 except ImportError: 

28 fsspec = None 

29 AbstractFileSystem = type 

30 

31from ._resourceHandles._baseResourceHandle import ResourceHandleProtocol 

32from ._resourcePath import ResourcePath, ResourcePathExpression 

33 

34log = logging.getLogger(__name__) 

35 

36 

37class PackageResourcePath(ResourcePath): 

38 """URI referring to a Python package resource. 

39 

40 These URIs look like: ``resource://lsst.daf.butler/configs/file.yaml`` 

41 where the network location is the Python package and the path is the 

42 resource name. 

43 """ 

44 

45 def _get_ref(self) -> resources.abc.Traversable | None: 

46 """Obtain the object representing the resource. 

47 

48 Returns 

49 ------- 

50 path : `resources.abc.Traversable` or `None` 

51 The reference to the resource path, or `None` if the module 

52 associated with the resources is not accessible. This can happen 

53 if Python can't import the Python package defining the resource. 

54 """ 

55 # Need the path without the leading /. 

56 path = self.path.lstrip("/") 

57 try: 

58 ref = resources.files(self.netloc).joinpath(path) 

59 except ModuleNotFoundError: 

60 return None 

61 return ref 

62 

63 def isdir(self) -> bool: 

64 """Return True if this URI is a directory, else False.""" 

65 if self.dirLike is None: 

66 ref = self._get_ref() 

67 if ref is not None: 

68 self.dirLike = ref.is_dir() 

69 else: 

70 return False 

71 return self.dirLike 

72 

73 def exists(self) -> bool: 

74 """Check that the python resource exists.""" 

75 ref = self._get_ref() 

76 if ref is None: 

77 return False 

78 return ref.is_file() or ref.is_dir() 

79 

80 def read(self, size: int = -1) -> bytes: 

81 ref = self._get_ref() 

82 if not ref: 

83 raise FileNotFoundError(f"Unable to locate resource {self}.") 

84 with ref.open("rb") as fh: 

85 return fh.read(size) 

86 

87 @contextlib.contextmanager 

88 def as_local( 

89 self, multithreaded: bool = True, tmpdir: ResourcePathExpression | None = None 

90 ) -> Iterator[ResourcePath]: 

91 """Return the location of the Python resource as local file. 

92 

93 Parameters 

94 ---------- 

95 multithreaded : `bool`, optional 

96 Unused. 

97 tmpdir : `ResourcePathExpression` or `None`, optional 

98 Unused. 

99 

100 Yields 

101 ------ 

102 local : `ResourcePath` 

103 This might be the original resource or a copy on the local file 

104 system. 

105 multithreaded : `bool`, optional 

106 Unused. 

107 

108 Notes 

109 ----- 

110 The context manager will automatically delete any local temporary 

111 file. 

112 

113 Examples 

114 -------- 

115 Should be used as a context manager: 

116 

117 .. code-block:: py 

118 

119 with uri.as_local() as local: 

120 ospath = local.ospath 

121 """ 

122 ref = self._get_ref() 

123 if ref is None: 

124 raise FileNotFoundError(f"Resource {self} could not be located.") 

125 if ref.is_dir(): 

126 raise IsADirectoryError(f"Directory-like URI {self} cannot be fetched as local.") 

127 

128 with resources.as_file(ref) as file: 

129 yield ResourcePath(file) 

130 

131 @contextlib.contextmanager 

132 def open( 

133 self, 

134 mode: str = "r", 

135 *, 

136 encoding: str | None = None, 

137 prefer_file_temporary: bool = False, 

138 ) -> Iterator[ResourceHandleProtocol]: 

139 # Docstring inherited. 

140 if "r" not in mode or "+" in mode: 

141 raise RuntimeError(f"Package resource URI {self} is read-only.") 

142 ref = self._get_ref() 

143 if ref is None: 

144 raise FileNotFoundError(f"Could not open resource {self}.") 

145 # mypy uses the literal value of mode to work out the parameters 

146 # and return value but mode here is a variable. 

147 with ref.open(mode, encoding=encoding) as buffer: # type: ignore[call-overload] 

148 yield buffer 

149 

150 def walk( 

151 self, file_filter: str | re.Pattern | None = None 

152 ) -> Iterator[list | tuple[ResourcePath, list[str], list[str]]]: 

153 # Docstring inherited. 

154 if not self.isdir(): 

155 raise ValueError(f"Can not walk a non-directory URI: {self}") 

156 

157 if isinstance(file_filter, str): 

158 file_filter = re.compile(file_filter) 

159 

160 ref = self._get_ref() 

161 if ref is None: 

162 raise ValueError(f"Unable to find resource {self}.") 

163 

164 files: list[str] = [] 

165 dirs: list[str] = [] 

166 for item in ref.iterdir(): 

167 if item.is_dir(): 

168 dirs.append(item.name) 

169 elif item.is_file(): 169 ↛ 166line 169 didn't jump to line 166 because the condition on line 169 was always true

170 files.append(item.name) 

171 # If the item wasn't covered by one of the cases above that 

172 # means it was deleted concurrently with this walk or is 

173 # not a plain file/directory/symlink 

174 

175 if file_filter is not None: 

176 files = [f for f in files if file_filter.search(f)] 

177 

178 if not dirs and not files: 

179 return 

180 else: 

181 yield type(self)(self, forceAbsolute=False, forceDirectory=True), dirs, files 

182 

183 for dir in dirs: 

184 new_uri = self.join(dir, forceDirectory=True) 

185 yield from new_uri.walk(file_filter) 

186 

187 def to_fsspec(self) -> tuple[AbstractFileSystem, str]: 

188 """Return an abstract file system and path that can be used by fsspec. 

189 

190 Python package resources are effectively local files in most cases 

191 but can be found inside ZIP files. To support this we would have 

192 to change this API to a context manager (using 

193 ``importlib.resources.as_file``) or find an API where fsspec knows 

194 about python package resource. 

195 

196 Returns 

197 ------- 

198 fs : `fsspec.spec.AbstractFileSystem` 

199 A file system object suitable for use with the returned path. 

200 path : `str` 

201 A path that can be opened by the file system object. 

202 """ 

203 raise NotImplementedError("fsspec can not be used with python package resources.")