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

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 Generator, Iterator 

20from importlib import resources 

21from typing import IO, TYPE_CHECKING 

22 

23if TYPE_CHECKING: 

24 from fsspec.spec import AbstractFileSystem 

25 

26from ._resourceHandles._baseResourceHandle import ResourceHandleProtocol 

27from ._resourcePath import ResourceInfo, ResourcePath, ResourcePathExpression 

28from .file import _path_to_info 

29 

30log = logging.getLogger(__name__) 

31 

32 

33class PackageResourcePath(ResourcePath): 

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

35 

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

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

38 resource name. 

39 """ 

40 

41 quotePaths = False 

42 

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

44 """Obtain the object representing the resource. 

45 

46 Returns 

47 ------- 

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

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

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

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

52 """ 

53 # Need the path without the leading /. 

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

55 try: 

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

57 except ModuleNotFoundError: 

58 return None 

59 return ref 

60 

61 def isdir(self) -> bool: 

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

63 if self.dirLike is None: 

64 ref = self._get_ref() 

65 if ref is not None: 

66 self.dirLike = ref.is_dir() 

67 else: 

68 return False 

69 return self.dirLike 

70 

71 def exists(self) -> bool: 

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

73 ref = self._get_ref() 

74 if ref is None: 

75 return False 

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

77 

78 def get_info(self) -> ResourceInfo: 

79 """Return metadata about the resource without reading its contents.""" 

80 ref = self._get_ref() 

81 if ref is None or not (ref.is_file() or ref.is_dir()): 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true

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

83 

84 info = _path_to_info(str(self), ref) 

85 

86 if info is None: 86 ↛ 88line 86 didn't jump to line 88 because the condition on line 86 was never true

87 # Edge case such as file in Zip. 

88 return ResourceInfo( 

89 uri=str(self), 

90 is_file=True, 

91 size=0, 

92 last_modified=None, 

93 checksums={}, 

94 ) 

95 return info 

96 

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

98 ref = self._get_ref() 

99 if not ref: 

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

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

102 return fh.read(size) 

103 

104 @contextlib.contextmanager 

105 def as_local( 

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

107 ) -> Generator[ResourcePath]: 

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

109 

110 Parameters 

111 ---------- 

112 multithreaded : `bool`, optional 

113 Unused. 

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

115 Unused. 

116 

117 Yields 

118 ------ 

119 local : `ResourcePath` 

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

121 system. 

122 multithreaded : `bool`, optional 

123 Unused. 

124 

125 Notes 

126 ----- 

127 The context manager will automatically delete any local temporary 

128 file. 

129 

130 Examples 

131 -------- 

132 Should be used as a context manager: 

133 

134 .. code-block:: py 

135 

136 with uri.as_local() as local: 

137 ospath = local.ospath 

138 """ 

139 ref = self._get_ref() 

140 if ref is None: 

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

142 if ref.is_dir(): 

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

144 

145 with resources.as_file(ref) as file: 

146 yield ResourcePath(file) 

147 

148 @contextlib.contextmanager 

149 def open( 

150 self, 

151 mode: str = "r", 

152 *, 

153 encoding: str | None = None, 

154 prefer_file_temporary: bool = False, 

155 ) -> Generator[ResourceHandleProtocol]: 

156 # Docstring inherited. 

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

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

159 ref = self._get_ref() 

160 if ref is None: 

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

162 # Traversable.open is overloaded on the literal mode, so dispatch 

163 # explicitly rather than passing the mode variable through. 

164 handle: IO[bytes] | IO[str] 

165 if "b" in mode: 

166 handle = ref.open("rb") 

167 else: 

168 handle = ref.open("r", encoding=encoding) 

169 with handle as buffer: 

170 yield buffer 

171 

172 def walk( 

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

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

175 # Docstring inherited. 

176 if not self.isdir(): 

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

178 

179 if isinstance(file_filter, str): 

180 file_filter = re.compile(file_filter) 

181 

182 ref = self._get_ref() 

183 if ref is None: 

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

185 

186 files: list[str] = [] 

187 dirs: list[str] = [] 

188 for item in ref.iterdir(): 

189 if item.is_dir(): 

190 dirs.append(item.name) 

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

192 files.append(item.name) 

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

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

195 # not a plain file/directory/symlink 

196 

197 if file_filter is not None: 

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

199 

200 if not dirs and not files: 

201 return 

202 else: 

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

204 

205 for dir in dirs: 

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

207 yield from new_uri.walk(file_filter) 

208 

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

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

211 

212 Python package resources are effectively local files in most cases 

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

214 to change this API to a context manager (using 

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

216 about python package resource. 

217 

218 Returns 

219 ------- 

220 fs : `fsspec.spec.AbstractFileSystem` 

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

222 path : `str` 

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

224 """ 

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