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 newline : `str` or `None`, optional 

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

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

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

43 newline representation. 

44 """ 

45 

46 def __init__( 

47 self, 

48 mode: str, 

49 log: logging.Logger, 

50 uri: DavResourcePath, 

51 stat: DavFileMetadata, 

52 *, 

53 newline: AnyStr | None = None, 

54 ) -> None: 

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

56 self._uri: DavResourcePath = uri 

57 self._stat: DavFileMetadata = stat 

58 self._current_position = 0 

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

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

61 self._closed = CloseStatus.OPEN 

62 

63 def close(self) -> None: 

64 self._closed = CloseStatus.CLOSED 

65 self._cache = None 

66 

67 @property 

68 def closed(self) -> bool: 

69 return self._closed == CloseStatus.CLOSED 

70 

71 def fileno(self) -> int: 

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

73 

74 def flush(self) -> None: 

75 modes = set(self._mode) 

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

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

78 

79 @property 

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

81 return False 

82 

83 def readable(self) -> bool: 

84 return True 

85 

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

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

88 

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

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

91 

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

93 match whence: 

94 case io.SEEK_SET: 

95 if offset < 0: 

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

97 self._current_position = offset 

98 case io.SEEK_CUR: 

99 self._current_position += offset 

100 case io.SEEK_END: 

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

102 case _: 

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

104 

105 if self._current_position < 0: 

106 self._current_position = 0 

107 

108 return self._current_position 

109 

110 def seekable(self) -> bool: 

111 return True 

112 

113 def tell(self) -> int: 

114 return self._current_position 

115 

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

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

118 

119 def writable(self) -> bool: 

120 return False 

121 

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

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

124 

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

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

127 

128 @property 

129 def _eof(self) -> bool: 

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

131 

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

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

134 memory buffer. 

135 """ 

136 if self._cache is None: 

137 self._cache = io.BytesIO() 

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

139 

140 return self._cache 

141 

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

143 if self._eof or size == 0: 

144 return b"" 

145 

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

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

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

149 # for retrieving small chunks. 

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

151 self._download_to_cache() 

152 

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

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

155 # cache. 

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

157 cache = self._download_to_cache() 

158 self._current_position = self._stat.size 

159 return cache.getvalue() 

160 

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

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

163 if self._cache is not None: 

164 start = self._current_position 

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

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

167 

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

169 # I/O buffer to reduce memory allocations. 

170 if self._buffer is None: 

171 self._buffer = io.BytesIO() 

172 

173 start = self._current_position 

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

175 self._buffer.seek(0) 

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

177 count = self._buffer.tell() 

178 self._current_position += count 

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

180 

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

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

183 of bytes read. 

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)