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

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 ResourceInfo, ResourcePath, ResourcePathExpression 

33from .file import _path_to_info 

34 

35log = logging.getLogger(__name__) 

36 

37 

38class PackageResourcePath(ResourcePath): 

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

40 

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

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

43 resource name. 

44 """ 

45 

46 quotePaths = False 

47 

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

49 """Obtain the object representing the resource. 

50 

51 Returns 

52 ------- 

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

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

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

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

57 """ 

58 # Need the path without the leading /. 

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

60 try: 

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

62 except ModuleNotFoundError: 

63 return None 

64 return ref 

65 

66 def isdir(self) -> bool: 

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

68 if self.dirLike is None: 

69 ref = self._get_ref() 

70 if ref is not None: 

71 self.dirLike = ref.is_dir() 

72 else: 

73 return False 

74 return self.dirLike 

75 

76 def exists(self) -> bool: 

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

78 ref = self._get_ref() 

79 if ref is None: 

80 return False 

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

82 

83 def get_info(self) -> ResourceInfo: 

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

85 ref = self._get_ref() 

86 if ref is None or not (ref.is_file() or ref.is_dir()): 

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

88 

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

90 

91 if info is None: 

92 # Edge case such as file in Zip. 

93 return ResourceInfo( 

94 uri=str(self), 

95 is_file=True, 

96 size=0, 

97 last_modified=None, 

98 checksums={}, 

99 ) 

100 return info 

101 

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

103 ref = self._get_ref() 

104 if not ref: 

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

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

107 return fh.read(size) 

108 

109 @contextlib.contextmanager 

110 def as_local( 

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

112 ) -> Iterator[ResourcePath]: 

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

114 

115 Parameters 

116 ---------- 

117 multithreaded : `bool`, optional 

118 Unused. 

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

120 Unused. 

121 

122 Yields 

123 ------ 

124 local : `ResourcePath` 

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

126 system. 

127 multithreaded : `bool`, optional 

128 Unused. 

129 

130 Notes 

131 ----- 

132 The context manager will automatically delete any local temporary 

133 file. 

134 

135 Examples 

136 -------- 

137 Should be used as a context manager: 

138 

139 .. code-block:: py 

140 

141 with uri.as_local() as local: 

142 ospath = local.ospath 

143 """ 

144 ref = self._get_ref() 

145 if ref is None: 

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

147 if ref.is_dir(): 

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

149 

150 with resources.as_file(ref) as file: 

151 yield ResourcePath(file) 

152 

153 @contextlib.contextmanager 

154 def open( 

155 self, 

156 mode: str = "r", 

157 *, 

158 encoding: str | None = None, 

159 prefer_file_temporary: bool = False, 

160 ) -> Iterator[ResourceHandleProtocol]: 

161 # Docstring inherited. 

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

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

164 ref = self._get_ref() 

165 if ref is None: 

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

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

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

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

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(): 

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.")