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

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 DavFileMetadata 

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 stat : `DavFileMetadata` 

40 Information about this resource. 

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 stat: DavFileMetadata, 

54 *, 

55 newline: AnyStr | None = None, 

56 ) -> None: 

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

58 self._uri: DavResourcePath = uri 

59 self._stat: DavFileMetadata = stat 

60 self._current_position = 0 

61 self._cache: io.BytesIO | None = None 

62 self._buffer: io.BytesIO | None = None 

63 self._closed = CloseStatus.OPEN 

64 

65 def close(self) -> None: 

66 self._closed = CloseStatus.CLOSED 

67 self._cache = None 

68 

69 @property 

70 def closed(self) -> bool: 

71 return self._closed == CloseStatus.CLOSED 

72 

73 def fileno(self) -> int: 

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

75 

76 def flush(self) -> None: 

77 modes = set(self._mode) 

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

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

80 

81 @property 

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

83 return False 

84 

85 def readable(self) -> bool: 

86 return True 

87 

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

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

90 

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

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

93 

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

95 match whence: 

96 case io.SEEK_SET: 

97 if offset < 0: 

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

99 self._current_position = offset 

100 case io.SEEK_CUR: 

101 self._current_position += offset 

102 case io.SEEK_END: 

103 self._current_position = self._stat.size + offset 

104 case _: 

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

106 

107 if self._current_position < 0: 

108 self._current_position = 0 

109 

110 return self._current_position 

111 

112 def seekable(self) -> bool: 

113 return True 

114 

115 def tell(self) -> int: 

116 return self._current_position 

117 

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

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

120 

121 def writable(self) -> bool: 

122 return False 

123 

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

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

126 

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

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

129 

130 @property 

131 def _eof(self) -> bool: 

132 return self._current_position >= self._stat.size 

133 

134 def _download_to_cache(self) -> io.BytesIO: 

135 """Download the entire content of the remote resource to an internal 

136 memory buffer. 

137 """ 

138 if self._cache is None: 

139 self._cache = io.BytesIO() 

140 self._cache.write(self._uri.read()) 

141 

142 return self._cache 

143 

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

145 if self._eof or size == 0: 

146 return b"" 

147 

148 # If this file's size is small than the buffer size configured for 

149 # this URI's client, download the entire file in one request and cache 

150 # its content. This avoids multiple roundtrips to the server 

151 # for retrieving small chunks. 

152 if self._stat.size <= self._uri._client._config.buffer_size: 

153 self._download_to_cache() 

154 

155 # If we are asked to read the whole file content, cache the entire 

156 # file content and return a copy-on-write memory view of our internal 

157 # cache. 

158 if self._current_position == 0 and size == -1: 

159 cache = self._download_to_cache() 

160 self._current_position = self._stat.size 

161 return cache.getvalue() 

162 

163 # This is a partial read. If we have already cached the whole file 

164 # content use the cache to build the return value. 

165 if self._cache is not None: 

166 start = self._current_position 

167 end = self._current_position = self._stat.size if size < 0 else start + size 

168 return self._cache.getvalue()[start:end] 

169 

170 # We need to make a partial read from the server. Reuse our internal 

171 # I/O buffer to reduce memory allocations. 

172 if self._buffer is None: 

173 self._buffer = io.BytesIO() 

174 

175 start = self._current_position 

176 end = self._stat.size if size < 0 else min(start + size, self._stat.size) 

177 self._buffer.seek(0) 

178 self._buffer.write(self._uri.read_range(start=start, end=end - 1)) 

179 count = self._buffer.tell() 

180 self._current_position += count 

181 return self._buffer.getvalue()[0:count] 

182 

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

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

185 of bytes read. 

186 

187 Parameters 

188 ---------- 

189 output : `bytearray` 

190 Byte array to write output into. 

191 """ 

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

193 return 0 

194 

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

196 output[:] = data 

197 return len(data)