Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/_parsers/resp3.py: 12%

1from logging import getLogger 

2from typing import Any, Union 

3 

4from ..exceptions import ConnectionError, InvalidResponse, ResponseError 

5from ..typing import EncodableT 

6from ..utils import SENTINEL 

7from .base import ( 

8 AsyncPushNotificationsParser, 

9 PushNotificationsParser, 

10 _AsyncRESPBase, 

11 _RESPBase, 

12) 

13from .socket import SERVER_CLOSED_CONNECTION_ERROR 

14 

15 

16class _RESP3Parser(_RESPBase, PushNotificationsParser): 

17 """RESP3 protocol implementation""" 

18 

19 def __init__(self, socket_read_size): 

20 super().__init__(socket_read_size) 

21 self.pubsub_push_handler_func = self.handle_pubsub_push_response 

22 self.node_moving_push_handler_func = None 

23 self.maintenance_push_handler_func = None 

24 self.oss_cluster_maint_push_handler_func = None 

25 self.invalidation_push_handler_func = None 

26 

27 def handle_pubsub_push_response(self, response): 

28 logger = getLogger("push_response") 

29 logger.debug("Push response: %s", response) 

30 return response 

31 

32 def read_response( 

33 self, 

34 disable_decoding=False, 

35 push_request=False, 

36 timeout: Union[float, object] = SENTINEL, 

37 ): 

38 pos = self._buffer.get_pos() if self._buffer is not None else None 

39 try: 

40 result = self._read_response( 

41 disable_decoding=disable_decoding, 

42 push_request=push_request, 

43 timeout=timeout, 

44 ) 

45 except BaseException: 

46 if self._buffer is not None: 

47 self._buffer.rewind(pos) 

48 raise 

49 else: 

50 if self._buffer is not None: 

51 try: 

52 self._buffer.purge() 

53 except AttributeError: 

54 # Buffer may have been set to None by another thread after 

55 # the check above; result is still valid so we don't raise 

56 pass 

57 return result 

58 

59 def _read_response( 

60 self, 

61 disable_decoding=False, 

62 push_request=False, 

63 timeout: Union[float, object] = SENTINEL, 

64 ): 

65 raw = self._buffer.readline(timeout=timeout) 

66 if not raw: 

67 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) 

68 

69 byte, response = raw[:1], raw[1:] 

70 

71 # server returned an error 

72 if byte in (b"-", b"!"): 

73 if byte == b"!": 

74 response = self._buffer.read(int(response), timeout=timeout) 

75 response = response.decode("utf-8", errors="replace") 

76 error = self.parse_error(response) 

77 # if the error is a ConnectionError, raise immediately so the user 

78 # is notified 

79 if isinstance(error, ConnectionError): 

80 raise error 

81 # otherwise, we're dealing with a ResponseError that might belong 

82 # inside a pipeline response. the connection's read_response() 

83 # and/or the pipeline's execute() will raise this error if 

84 # necessary, so just return the exception instance here. 

85 return error 

86 # single value 

87 elif byte == b"+": 

88 pass 

89 # null value 

90 elif byte == b"_": 

91 return None 

92 # int and big int values 

93 elif byte in (b":", b"("): 

94 return int(response) 

95 # double value 

96 elif byte == b",": 

97 return float(response) 

98 # bool value 

99 elif byte == b"#": 

100 return response == b"t" 

101 # bulk response 

102 elif byte == b"$": 

103 response = self._buffer.read(int(response), timeout=timeout) 

104 # verbatim string response 

105 elif byte == b"=": 

106 response = self._buffer.read(int(response), timeout=timeout)[4:] 

107 # array response 

108 elif byte == b"*": 

109 response = [ 

110 self._read_response(disable_decoding=disable_decoding, timeout=timeout) 

111 for _ in range(int(response)) 

112 ] 

113 # set response 

114 elif byte == b"~": 

115 # redis can return unhashable types (like dict) in a set, 

116 # so we return sets as list, all the time, for predictability 

117 response = [ 

118 self._read_response(disable_decoding=disable_decoding, timeout=timeout) 

119 for _ in range(int(response)) 

120 ] 

121 # map response 

122 elif byte == b"%": 

123 # We cannot use a dict-comprehension to parse stream. 

124 # Evaluation order of key:val expression in dict comprehension only 

125 # became defined to be left-right in version 3.8 

126 resp_dict = {} 

127 for _ in range(int(response)): 

128 key = self._read_response( 

129 disable_decoding=disable_decoding, timeout=timeout 

130 ) 

131 resp_dict[key] = self._read_response( 

132 disable_decoding=disable_decoding, 

133 push_request=push_request, 

134 timeout=timeout, 

135 ) 

136 response = resp_dict 

137 # push response 

138 elif byte == b">": 

139 response = [ 

140 self._read_response( 

141 disable_decoding=disable_decoding, 

142 push_request=push_request, 

143 timeout=timeout, 

144 ) 

145 for _ in range(int(response)) 

146 ] 

147 response = self.handle_push_response(response) 

148 

149 # if this is a push request return the push response 

150 if push_request: 

151 return response 

152 

153 # `timeout` has to be forwarded here too: it is the caller's bound 

154 # for this read, and dropping it would silently fall back to the 

155 # connection's socket_timeout for the rest of the response just 

156 # because a push message happened to interleave. The hiredis parser 

157 # already forwards it on the same continuation. 

158 return self._read_response( 

159 disable_decoding=disable_decoding, 

160 push_request=push_request, 

161 timeout=timeout, 

162 ) 

163 else: 

164 raise InvalidResponse(f"Protocol Error: {raw!r}") 

165 

166 if isinstance(response, bytes) and disable_decoding is False: 

167 response = self.encoder.decode(response) 

168 

169 return response 

170 

171 

172class _AsyncRESP3Parser(_AsyncRESPBase, AsyncPushNotificationsParser): 

173 def __init__(self, socket_read_size): 

174 super().__init__(socket_read_size) 

175 self.pubsub_push_handler_func = self.handle_pubsub_push_response 

176 self.invalidation_push_handler_func = None 

177 

178 async def handle_pubsub_push_response(self, response): 

179 logger = getLogger("push_response") 

180 logger.debug("Push response: %s", response) 

181 return response 

182 

183 async def read_response( 

184 self, disable_decoding: bool = False, push_request: bool = False 

185 ): 

186 if not self._connected: 

187 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) 

188 if self._chunks: 

189 # augment parsing buffer with previously read data 

190 self._buffer += b"".join(self._chunks) 

191 self._chunks.clear() 

192 self._pos = 0 

193 response = await self._read_response( 

194 disable_decoding=disable_decoding, push_request=push_request 

195 ) 

196 # Successfully parsing a response allows us to clear our parsing buffer 

197 self._clear() 

198 return response 

199 

200 async def _read_response( 

201 self, disable_decoding: bool = False, push_request: bool = False 

202 ) -> Union[EncodableT, ResponseError, None]: 

203 if not self._stream or not self.encoder: 

204 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) 

205 raw = await self._readline() 

206 response: Any 

207 byte, response = raw[:1], raw[1:] 

208 

209 # if byte not in (b"-", b"+", b":", b"$", b"*"): 

210 # raise InvalidResponse(f"Protocol Error: {raw!r}") 

211 

212 # server returned an error 

213 if byte in (b"-", b"!"): 

214 if byte == b"!": 

215 response = await self._read(int(response)) 

216 response = response.decode("utf-8", errors="replace") 

217 error = self.parse_error(response) 

218 # if the error is a ConnectionError, raise immediately so the user 

219 # is notified 

220 if isinstance(error, ConnectionError): 

221 self._clear() # Successful parse 

222 raise error 

223 # otherwise, we're dealing with a ResponseError that might belong 

224 # inside a pipeline response. the connection's read_response() 

225 # and/or the pipeline's execute() will raise this error if 

226 # necessary, so just return the exception instance here. 

227 return error 

228 # single value 

229 elif byte == b"+": 

230 pass 

231 # null value 

232 elif byte == b"_": 

233 return None 

234 # int and big int values 

235 elif byte in (b":", b"("): 

236 return int(response) 

237 # double value 

238 elif byte == b",": 

239 return float(response) 

240 # bool value 

241 elif byte == b"#": 

242 return response == b"t" 

243 # bulk response 

244 elif byte == b"$": 

245 response = await self._read(int(response)) 

246 # verbatim string response 

247 elif byte == b"=": 

248 response = (await self._read(int(response)))[4:] 

249 # array response 

250 elif byte == b"*": 

251 response = [ 

252 (await self._read_response(disable_decoding=disable_decoding)) 

253 for _ in range(int(response)) 

254 ] 

255 # set response 

256 elif byte == b"~": 

257 # redis can return unhashable types (like dict) in a set, 

258 # so we always convert to a list, to have predictable return types 

259 response = [ 

260 (await self._read_response(disable_decoding=disable_decoding)) 

261 for _ in range(int(response)) 

262 ] 

263 # map response 

264 elif byte == b"%": 

265 # We cannot use a dict-comprehension to parse stream. 

266 # Evaluation order of key:val expression in dict comprehension only 

267 # became defined to be left-right in version 3.8 

268 resp_dict = {} 

269 for _ in range(int(response)): 

270 key = await self._read_response(disable_decoding=disable_decoding) 

271 resp_dict[key] = await self._read_response( 

272 disable_decoding=disable_decoding, push_request=push_request 

273 ) 

274 response = resp_dict 

275 # push response 

276 elif byte == b">": 

277 response = [ 

278 ( 

279 await self._read_response( 

280 disable_decoding=disable_decoding, push_request=push_request 

281 ) 

282 ) 

283 for _ in range(int(response)) 

284 ] 

285 response = await self.handle_push_response(response) 

286 if not push_request: 

287 return await self._read_response( 

288 disable_decoding=disable_decoding, push_request=push_request 

289 ) 

290 else: 

291 return response 

292 else: 

293 raise InvalidResponse(f"Protocol Error: {raw!r}") 

294 

295 if isinstance(response, bytes) and disable_decoding is False: 

296 response = self.encoder.decode(response) 

297 return response