Coverage for python/lsst/resources/_resourceHandles/_davResourceHandle.py: 24%

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__ = ("DavReadResourceHandle",) 

15 

16import io 

17import logging 

18from collections.abc import Callable, Iterable 

19from typing import TYPE_CHECKING, AnyStr 

20 

21from ..davutils import DavClient, redact_url 

22from ._baseResourceHandle import BaseResourceHandle, CloseStatus 

23 

24if TYPE_CHECKING: 

25 from ..dav import DavResourcePath 

26 

27 

28class DavReadResourceHandle(BaseResourceHandle[bytes]): 

29 """WebDAV-based specialization of `.BaseResourceHandle`. 

30 

31 Parameters 

32 ---------- 

33 mode : `str` 

34 Handle modes as described in the python `io` module. 

35 log : `~logging.Logger` 

36 Logger to used when writing messages. 

37 uri : `lsst.resources.dav.DavResourcePath` 

38 URI of remote resource. 

39 file_size : `int` 

40 Size of the remote file, in bytes. 

41 newline : `str` or `None`, optional 

42 When doing multiline operations, break the stream on given character. 

43 Defaults to newline. If a file is opened in binary mode, this argument 

44 is not used, as binary files will only split lines on the binary 

45 newline representation. 

46 """ 

47 

48 def __init__( 

49 self, 

50 mode: str, 

51 log: logging.Logger, 

52 uri: DavResourcePath, 

53 file_size: int, 

54 *, 

55 newline: AnyStr | None = None, 

56 ) -> None: 

57 super().__init__(mode, log, uri, newline=newline) 

58 self._client: DavClient = uri._client 

59 self._filesize: int = file_size 

60 self._closed = CloseStatus.OPEN 

61 self._current_position = 0 

62 self._cache: DavReadAheadCache = DavReadAheadCache( 

63 client=self._client, 

64 url=uri._internal_url, 

65 filesize=self._filesize, 

66 blocksize=self._client._config.block_size, 

67 log=log, 

68 ) 

69 self._log.debug("initializing read handle for %s [%#x]", self._uri, id(self)) 

70 

71 def close(self) -> None: 

72 if self._closed != CloseStatus.CLOSED: 

73 self._log.debug("closing read handle for %s [%#x]", self._uri, id(self)) 

74 self._cache.close() 

75 self._closed = CloseStatus.CLOSED 

76 

77 @property 

78 def closed(self) -> bool: 

79 return self._closed == CloseStatus.CLOSED 

80 

81 def fileno(self) -> int: 

82 raise io.UnsupportedOperation("DavReadResourceHandle does not have a file number") 

83 

84 def flush(self) -> None: 

85 modes = set(self._mode) 

86 if {"w", "x", "a", "+"} & modes: 

87 raise io.UnsupportedOperation("DavReadResourceHandles are read only") 

88 

89 @property 

90 def isatty(self) -> bool | Callable[[], bool]: 

91 return False 

92 

93 def readable(self) -> bool: 

94 return True 

95 

96 def readline(self, size: int = -1) -> bytes: 

97 raise io.UnsupportedOperation("DavReadResourceHandles Do not support line by line reading") 

98 

99 def readlines(self, hint: int = -1) -> Iterable[bytes]: 

100 raise io.UnsupportedOperation("DavReadResourceHandles Do not support line by line reading") 

101 

102 def seek(self, offset: int, whence: int = io.SEEK_SET) -> int: 

103 self._log.debug( 

104 "handle seek for %s: offset=%d, whence=%d, current_position=%d", 

105 self._uri, 

106 offset, 

107 whence, 

108 self._current_position, 

109 ) 

110 

111 match whence: 

112 case io.SEEK_SET: 

113 if offset < 0: 

114 raise ValueError(f"negative seek value {offset}") 

115 self._current_position = offset 

116 case io.SEEK_CUR: 

117 self._current_position += offset 

118 case io.SEEK_END: 

119 self._current_position = self._filesize + offset 

120 case _: 

121 raise ValueError(f"unexpected value {whence} for whence in seek()") 

122 

123 if self._current_position < 0: 

124 self._current_position = 0 

125 

126 return self._current_position 

127 

128 def seekable(self) -> bool: 

129 return True 

130 

131 def tell(self) -> int: 

132 return self._current_position 

133 

134 def truncate(self, size: int | None = None) -> int: 

135 raise io.UnsupportedOperation("DavReadResourceHandles Do not support truncation") 

136 

137 def writable(self) -> bool: 

138 return False 

139 

140 def write(self, b: bytes, /) -> int: 

141 raise io.UnsupportedOperation("DavReadResourceHandles are read only") 

142 

143 def writelines(self, b: Iterable[bytes], /) -> None: 

144 raise io.UnsupportedOperation("DavReadResourceHandles are read only") 

145 

146 @property 

147 def _eof(self) -> bool: 

148 return self._current_position >= self._filesize 

149 

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

151 self._log.debug( 

152 "handle read for %s: filesize=%d, current_position=%d, size=%d", 

153 self._uri, 

154 self._filesize, 

155 self._current_position, 

156 size, 

157 ) 

158 

159 if self.closed: 

160 raise ValueError("I/O operation on closed file") 

161 

162 if size == 0 or self._eof: 

163 return b"" 

164 

165 if size < 0: 

166 # Read up to the end of the file 

167 size = self._filesize - self._current_position 

168 

169 output = self._cache.fetch(start=self._current_position, end=self._current_position + size) 

170 self._current_position += len(output) 

171 

172 self._log.debug("returning %d bytes from handle read for %s", len(output), self._uri) 

173 

174 return output 

175 

176 def readinto(self, output: bytearray) -> int: 

177 """Read up to `len(output)` bytes into `output` and return the number 

178 of bytes read. 

179 

180 Parameters 

181 ---------- 

182 output : `bytearray` 

183 Byte array to write output into. 

184 """ 

185 if self._eof or len(output) == 0: 

186 return 0 

187 

188 data = self.read(len(output)) 

189 output[:] = data 

190 return len(data) 

191 

192 

193class DavReadAheadCache: 

194 """Helper read-ahead cache for fetching chunks of a DavResourceHandle. 

195 

196 Parameters 

197 ---------- 

198 client : `lsst.resources.davutils.DavClient` 

199 The webDAV client to interact with the server to download data. 

200 url : `str` 

201 URL of the resource to download data from. 

202 filesize : `int` 

203 Size in bytes of the remote file. 

204 blocksize : `int` 

205 Size in bytes of the block for this resource. This is the size we use 

206 to retrieve data from this resource. 

207 log : `logging.Logger` 

208 Logger object to emit log records. 

209 

210 Notes 

211 ----- 

212 Behavior of this cache is inspired from fsspec's ReadAheadCache class. 

213 https://github.com/fsspec/filesystem_spec/blob/master/fsspec/caching.py 

214 """ 

215 

216 def __init__( 

217 self, client: DavClient, url: str, filesize: int, blocksize: int, log: logging.Logger 

218 ) -> None: 

219 self._client: DavClient = client 

220 self._url: str = url 

221 self._backend_url: str = url 

222 self._filesize: int = filesize 

223 self._blocksize: int = blocksize 

224 self._cache = b"" 

225 self._start: int = 0 

226 self._end: int = 0 

227 self._log: logging.Logger = log 

228 

229 def geturl(self) -> str: 

230 return redact_url(self._url) 

231 

232 def fetch(self, start: int, end: int) -> bytes: 

233 """Fetch a chunk of the file and store it in memory. 

234 

235 Parameters 

236 ---------- 

237 start : `int` 

238 Position of the first byte of the chunk. 

239 end : `int` 

240 Position of the last byte of the chunk. 

241 

242 Returns 

243 ------- 

244 output: `bytes` 

245 A chunk of up to end-start bytes. The returned chunk is 

246 served directly from the in-memory buffer without fetching new 

247 data from the remote file if it is already cached. Otherwise, 

248 a new chunk is retrieved from the origin file server and cached 

249 in memory. The size of the chunk can be the configured block 

250 size for this particular kind of resource path or the remaining 

251 bytes in the file. 

252 """ 

253 self._log.debug( 

254 "DavReadAheadCache.fetch: %s start=%d end=%d [total: %d]", 

255 self.geturl(), 

256 start, 

257 end, 

258 end - start, 

259 ) 

260 start = max(0, start) 

261 end = min(end, self._filesize) 

262 if start >= self._filesize or start >= end: 

263 return b"" 

264 

265 if start >= self._start and end <= self._end: 

266 # The requested chunk is entirely cached 

267 return self._cache[start - self._start : end - self._start] 

268 

269 # The requested chunk is not fully in cache. Repopulate the cache 

270 # with a number of blocks large enough to satisfy the requested chunk. 

271 blocks_to_fetch = 1 + ((end - start) // self._blocksize) 

272 bytes_to_fetch = self._blocksize * blocks_to_fetch 

273 end_range = min(self._filesize, start + bytes_to_fetch) 

274 start_range = max(0, end_range - bytes_to_fetch) 

275 

276 self._log.debug( 

277 "populating handle cache for %s with %d blocks [%d - %d, total bytes: %d]", 

278 self.geturl(), 

279 blocks_to_fetch, 

280 start_range, 

281 end_range, 

282 end_range - start_range, 

283 ) 

284 

285 # Read the requested range. 

286 self._backend_url, self._cache = self._client.read_range( 

287 self._backend_url, start=start_range, end=end_range - 1 

288 ) 

289 self._start = start_range 

290 self._end = self._start + len(self._cache) 

291 return self._cache[start - self._start : end - self._start] 

292 

293 def close(self) -> None: 

294 # Notify the remote server that we are not issueing more partial 

295 # read requests for this file handle. 

296 self._client._close(self._backend_url)