Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/_parsers/resp3.py: 12%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from logging import getLogger
2from typing import Any, Union
4from ..exceptions import ConnectionError, InvalidResponse, ResponseError
5from ..typing import EncodableT
6from ..utils import SENTINEL
7from .base import (
8 AsyncPushNotificationsParser,
9 PushNotificationsParser,
10 _AsyncRESPBase,
11 _parse_float,
12 _parse_int,
13 _RESPBase,
14)
15from .socket import SERVER_CLOSED_CONNECTION_ERROR
18class _RESP3Parser(_RESPBase, PushNotificationsParser):
19 """RESP3 protocol implementation"""
21 def __init__(self, socket_read_size):
22 super().__init__(socket_read_size)
23 self.pubsub_push_handler_func = self.handle_pubsub_push_response
24 self.node_moving_push_handler_func = None
25 self.maintenance_push_handler_func = None
26 self.oss_cluster_maint_push_handler_func = None
27 self.invalidation_push_handler_func = None
29 def handle_pubsub_push_response(self, response):
30 logger = getLogger("push_response")
31 logger.debug("Push response: %s", response)
32 return response
34 def read_response(
35 self,
36 disable_decoding=False,
37 push_request=False,
38 timeout: Union[float, object] = SENTINEL,
39 ):
40 pos = self._buffer.get_pos() if self._buffer is not None else None
41 try:
42 result = self._read_response(
43 disable_decoding=disable_decoding,
44 push_request=push_request,
45 timeout=timeout,
46 )
47 except BaseException:
48 if self._buffer is not None:
49 self._buffer.rewind(pos)
50 raise
51 else:
52 if self._buffer is not None:
53 try:
54 self._buffer.purge()
55 except AttributeError:
56 # Buffer may have been set to None by another thread after
57 # the check above; result is still valid so we don't raise
58 pass
59 return result
61 def _read_response(
62 self,
63 disable_decoding=False,
64 push_request=False,
65 timeout: Union[float, object] = SENTINEL,
66 ):
67 raw = self._buffer.readline(timeout=timeout)
68 if not raw:
69 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
71 byte, response = raw[:1], raw[1:]
73 # server returned an error
74 if byte in (b"-", b"!"):
75 if byte == b"!":
76 response = self._buffer.read(_parse_int(response, raw), timeout=timeout)
77 response = response.decode("utf-8", errors="replace")
78 error = self.parse_error(response)
79 # if the error is a ConnectionError, raise immediately so the user
80 # is notified
81 if isinstance(error, ConnectionError):
82 raise error
83 # otherwise, we're dealing with a ResponseError that might belong
84 # inside a pipeline response. the connection's read_response()
85 # and/or the pipeline's execute() will raise this error if
86 # necessary, so just return the exception instance here.
87 return error
88 # single value
89 elif byte == b"+":
90 pass
91 # null value
92 elif byte == b"_":
93 return None
94 # int and big int values
95 elif byte in (b":", b"("):
96 return _parse_int(response, raw)
97 # double value
98 elif byte == b",":
99 return _parse_float(response, raw)
100 # bool value
101 elif byte == b"#":
102 return response == b"t"
103 # bulk response
104 elif byte == b"$":
105 response = self._buffer.read(_parse_int(response, raw), timeout=timeout)
106 # verbatim string response
107 elif byte == b"=":
108 response = self._buffer.read(_parse_int(response, raw), timeout=timeout)[4:]
109 # array response
110 elif byte == b"*":
111 response = [
112 self._read_response(disable_decoding=disable_decoding, timeout=timeout)
113 for _ in range(_parse_int(response, raw))
114 ]
115 # set response
116 elif byte == b"~":
117 # redis can return unhashable types (like dict) in a set,
118 # so we return sets as list, all the time, for predictability
119 response = [
120 self._read_response(disable_decoding=disable_decoding, timeout=timeout)
121 for _ in range(_parse_int(response, raw))
122 ]
123 # map response
124 elif byte == b"%":
125 # We cannot use a dict-comprehension to parse stream.
126 # Evaluation order of key:val expression in dict comprehension only
127 # became defined to be left-right in version 3.8
128 resp_dict = {}
129 for _ in range(_parse_int(response, raw)):
130 key = self._read_response(
131 disable_decoding=disable_decoding, timeout=timeout
132 )
133 resp_dict[key] = self._read_response(
134 disable_decoding=disable_decoding,
135 push_request=push_request,
136 timeout=timeout,
137 )
138 response = resp_dict
139 # push response
140 elif byte == b">":
141 response = [
142 self._read_response(
143 disable_decoding=disable_decoding,
144 push_request=push_request,
145 timeout=timeout,
146 )
147 for _ in range(_parse_int(response, raw))
148 ]
149 response = self.handle_push_response(response)
151 # if this is a push request return the push response
152 if push_request:
153 return response
155 # `timeout` has to be forwarded here too: it is the caller's bound
156 # for this read, and dropping it would silently fall back to the
157 # connection's socket_timeout for the rest of the response just
158 # because a push message happened to interleave. The hiredis parser
159 # already forwards it on the same continuation.
160 return self._read_response(
161 disable_decoding=disable_decoding,
162 push_request=push_request,
163 timeout=timeout,
164 )
165 else:
166 raise InvalidResponse(f"Protocol Error: {raw!r}")
168 if isinstance(response, bytes) and disable_decoding is False:
169 response = self.encoder.decode(response)
171 return response
174class _AsyncRESP3Parser(_AsyncRESPBase, AsyncPushNotificationsParser):
175 def __init__(self, socket_read_size):
176 super().__init__(socket_read_size)
177 self.pubsub_push_handler_func = self.handle_pubsub_push_response
178 self.invalidation_push_handler_func = None
180 async def handle_pubsub_push_response(self, response):
181 logger = getLogger("push_response")
182 logger.debug("Push response: %s", response)
183 return response
185 async def read_response(
186 self, disable_decoding: bool = False, push_request: bool = False
187 ):
188 if not self._connected:
189 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
190 if self._chunks:
191 # augment parsing buffer with previously read data
192 self._buffer += b"".join(self._chunks)
193 self._chunks.clear()
194 self._pos = 0
195 response = await self._read_response(
196 disable_decoding=disable_decoding, push_request=push_request
197 )
198 # Successfully parsing a response allows us to clear our parsing buffer
199 self._clear()
200 return response
202 async def _read_response(
203 self, disable_decoding: bool = False, push_request: bool = False
204 ) -> Union[EncodableT, ResponseError, None]:
205 if not self._stream or not self.encoder:
206 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
207 raw = await self._readline()
208 response: Any
209 byte, response = raw[:1], raw[1:]
211 # if byte not in (b"-", b"+", b":", b"$", b"*"):
212 # raise InvalidResponse(f"Protocol Error: {raw!r}")
214 # server returned an error
215 if byte in (b"-", b"!"):
216 if byte == b"!":
217 response = await self._read(_parse_int(response, raw))
218 response = response.decode("utf-8", errors="replace")
219 error = self.parse_error(response)
220 # if the error is a ConnectionError, raise immediately so the user
221 # is notified
222 if isinstance(error, ConnectionError):
223 self._clear() # Successful parse
224 raise error
225 # otherwise, we're dealing with a ResponseError that might belong
226 # inside a pipeline response. the connection's read_response()
227 # and/or the pipeline's execute() will raise this error if
228 # necessary, so just return the exception instance here.
229 return error
230 # single value
231 elif byte == b"+":
232 pass
233 # null value
234 elif byte == b"_":
235 return None
236 # int and big int values
237 elif byte in (b":", b"("):
238 return _parse_int(response, raw)
239 # double value
240 elif byte == b",":
241 return _parse_float(response, raw)
242 # bool value
243 elif byte == b"#":
244 return response == b"t"
245 # bulk response
246 elif byte == b"$":
247 response = await self._read(_parse_int(response, raw))
248 # verbatim string response
249 elif byte == b"=":
250 response = (await self._read(_parse_int(response, raw)))[4:]
251 # array response
252 elif byte == b"*":
253 response = [
254 (await self._read_response(disable_decoding=disable_decoding))
255 for _ in range(_parse_int(response, raw))
256 ]
257 # set response
258 elif byte == b"~":
259 # redis can return unhashable types (like dict) in a set,
260 # so we always convert to a list, to have predictable return types
261 response = [
262 (await self._read_response(disable_decoding=disable_decoding))
263 for _ in range(_parse_int(response, raw))
264 ]
265 # map response
266 elif byte == b"%":
267 # We cannot use a dict-comprehension to parse stream.
268 # Evaluation order of key:val expression in dict comprehension only
269 # became defined to be left-right in version 3.8
270 resp_dict = {}
271 for _ in range(_parse_int(response, raw)):
272 key = await self._read_response(disable_decoding=disable_decoding)
273 resp_dict[key] = await self._read_response(
274 disable_decoding=disable_decoding, push_request=push_request
275 )
276 response = resp_dict
277 # push response
278 elif byte == b">":
279 response = [
280 (
281 await self._read_response(
282 disable_decoding=disable_decoding, push_request=push_request
283 )
284 )
285 for _ in range(_parse_int(response, raw))
286 ]
287 response = await self.handle_push_response(response)
288 if not push_request:
289 return await self._read_response(
290 disable_decoding=disable_decoding, push_request=push_request
291 )
292 else:
293 return response
294 else:
295 raise InvalidResponse(f"Protocol Error: {raw!r}")
297 if isinstance(response, bytes) and disable_decoding is False:
298 response = self.encoder.decode(response)
299 return response