We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
resp3 parser async parser incorrectly parses maps
Issue discovered in pr #2947

the evaulation order of `key:val` in `{key():val() for foo in bar}` dict comprehension is undefined in Python.  It mostly is left-right,
but some versions, notably recent 3.7 versions, don't behave that way.  It is best to use explicit ordering when reading from  a stream.

This had already been fixed for __non-async__ parsing, but in pr #2947 caused some scheduling behaviour which triggered this to happen. A fix for __async__ has been added there.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `redis/_parsers/resp3.py`
Content:
```
1 from logging import getLogger
2 from typing import Any, Union
3 
4 from ..exceptions import ConnectionError, InvalidResponse, ResponseError
5 from ..typing import EncodableT
6 from .base import _AsyncRESPBase, _RESPBase
7 from .socket import SERVER_CLOSED_CONNECTION_ERROR
8 
9 
10 class _RESP3Parser(_RESPBase):
11     """RESP3 protocol implementation"""
12 
13     def __init__(self, socket_read_size):
14         super().__init__(socket_read_size)
15         self.push_handler_func = self.handle_push_response
16 
17     def handle_push_response(self, response):
18         logger = getLogger("push_response")
19         logger.info("Push response: " + str(response))
20         return response
21 
22     def read_response(self, disable_decoding=False, push_request=False):
23         pos = self._buffer.get_pos() if self._buffer else None
24         try:
25             result = self._read_response(
26                 disable_decoding=disable_decoding, push_request=push_request
27             )
28         except BaseException:
29             if self._buffer:
30                 self._buffer.rewind(pos)
31             raise
32         else:
33             self._buffer.purge()
34             return result
35 
36     def _read_response(self, disable_decoding=False, push_request=False):
37         raw = self._buffer.readline()
38         if not raw:
39             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
40 
41         byte, response = raw[:1], raw[1:]
42 
43         # server returned an error
44         if byte in (b"-", b"!"):
45             if byte == b"!":
46                 response = self._buffer.read(int(response))
47             response = response.decode("utf-8", errors="replace")
48             error = self.parse_error(response)
49             # if the error is a ConnectionError, raise immediately so the user
50             # is notified
51             if isinstance(error, ConnectionError):
52                 raise error
53             # otherwise, we're dealing with a ResponseError that might belong
54             # inside a pipeline response. the connection's read_response()
55             # and/or the pipeline's execute() will raise this error if
56             # necessary, so just return the exception instance here.
57             return error
58         # single value
59         elif byte == b"+":
60             pass
61         # null value
62         elif byte == b"_":
63             return None
64         # int and big int values
65         elif byte in (b":", b"("):
66             return int(response)
67         # double value
68         elif byte == b",":
69             return float(response)
70         # bool value
71         elif byte == b"#":
72             return response == b"t"
73         # bulk response
74         elif byte == b"$":
75             response = self._buffer.read(int(response))
76         # verbatim string response
77         elif byte == b"=":
78             response = self._buffer.read(int(response))[4:]
79         # array response
80         elif byte == b"*":
81             response = [
82                 self._read_response(disable_decoding=disable_decoding)
83                 for _ in range(int(response))
84             ]
85         # set response
86         elif byte == b"~":
87             # redis can return unhashable types (like dict) in a set,
88             # so we need to first convert to a list, and then try to convert it to a set
89             response = [
90                 self._read_response(disable_decoding=disable_decoding)
91                 for _ in range(int(response))
92             ]
93             try:
94                 response = set(response)
95             except TypeError:
96                 pass
97         # map response
98         elif byte == b"%":
99             # we use this approach and not dict comprehension here
100             # because this dict comprehension fails in python 3.7
101             resp_dict = {}
102             for _ in range(int(response)):
103                 key = self._read_response(disable_decoding=disable_decoding)
104                 resp_dict[key] = self._read_response(
105                     disable_decoding=disable_decoding, push_request=push_request
106                 )
107             response = resp_dict
108         # push response
109         elif byte == b">":
110             response = [
111                 self._read_response(
112                     disable_decoding=disable_decoding, push_request=push_request
113                 )
114                 for _ in range(int(response))
115             ]
116             res = self.push_handler_func(response)
117             if not push_request:
118                 return self._read_response(
119                     disable_decoding=disable_decoding, push_request=push_request
120                 )
121             else:
122                 return res
123         else:
124             raise InvalidResponse(f"Protocol Error: {raw!r}")
125 
126         if isinstance(response, bytes) and disable_decoding is False:
127             response = self.encoder.decode(response)
128         return response
129 
130     def set_push_handler(self, push_handler_func):
131         self.push_handler_func = push_handler_func
132 
133 
134 class _AsyncRESP3Parser(_AsyncRESPBase):
135     def __init__(self, socket_read_size):
136         super().__init__(socket_read_size)
137         self.push_handler_func = self.handle_push_response
138 
139     def handle_push_response(self, response):
140         logger = getLogger("push_response")
141         logger.info("Push response: " + str(response))
142         return response
143 
144     async def read_response(
145         self, disable_decoding: bool = False, push_request: bool = False
146     ):
147         if self._chunks:
148             # augment parsing buffer with previously read data
149             self._buffer += b"".join(self._chunks)
150             self._chunks.clear()
151         self._pos = 0
152         response = await self._read_response(
153             disable_decoding=disable_decoding, push_request=push_request
154         )
155         # Successfully parsing a response allows us to clear our parsing buffer
156         self._clear()
157         return response
158 
159     async def _read_response(
160         self, disable_decoding: bool = False, push_request: bool = False
161     ) -> Union[EncodableT, ResponseError, None]:
162         if not self._stream or not self.encoder:
163             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
164         raw = await self._readline()
165         response: Any
166         byte, response = raw[:1], raw[1:]
167 
168         # if byte not in (b"-", b"+", b":", b"$", b"*"):
169         #     raise InvalidResponse(f"Protocol Error: {raw!r}")
170 
171         # server returned an error
172         if byte in (b"-", b"!"):
173             if byte == b"!":
174                 response = await self._read(int(response))
175             response = response.decode("utf-8", errors="replace")
176             error = self.parse_error(response)
177             # if the error is a ConnectionError, raise immediately so the user
178             # is notified
179             if isinstance(error, ConnectionError):
180                 self._clear()  # Successful parse
181                 raise error
182             # otherwise, we're dealing with a ResponseError that might belong
183             # inside a pipeline response. the connection's read_response()
184             # and/or the pipeline's execute() will raise this error if
185             # necessary, so just return the exception instance here.
186             return error
187         # single value
188         elif byte == b"+":
189             pass
190         # null value
191         elif byte == b"_":
192             return None
193         # int and big int values
194         elif byte in (b":", b"("):
195             return int(response)
196         # double value
197         elif byte == b",":
198             return float(response)
199         # bool value
200         elif byte == b"#":
201             return response == b"t"
202         # bulk response
203         elif byte == b"$":
204             response = await self._read(int(response))
205         # verbatim string response
206         elif byte == b"=":
207             response = (await self._read(int(response)))[4:]
208         # array response
209         elif byte == b"*":
210             response = [
211                 (await self._read_response(disable_decoding=disable_decoding))
212                 for _ in range(int(response))
213             ]
214         # set response
215         elif byte == b"~":
216             # redis can return unhashable types (like dict) in a set,
217             # so we need to first convert to a list, and then try to convert it to a set
218             response = [
219                 (await self._read_response(disable_decoding=disable_decoding))
220                 for _ in range(int(response))
221             ]
222             try:
223                 response = set(response)
224             except TypeError:
225                 pass
226         # map response
227         elif byte == b"%":
228             response = {
229                 (await self._read_response(disable_decoding=disable_decoding)): (
230                     await self._read_response(disable_decoding=disable_decoding)
231                 )
232                 for _ in range(int(response))
233             }
234         # push response
235         elif byte == b">":
236             response = [
237                 (
238                     await self._read_response(
239                         disable_decoding=disable_decoding, push_request=push_request
240                     )
241                 )
242                 for _ in range(int(response))
243             ]
244             res = self.push_handler_func(response)
245             if not push_request:
246                 return await self._read_response(
247                     disable_decoding=disable_decoding, push_request=push_request
248                 )
249             else:
250                 return res
251         else:
252             raise InvalidResponse(f"Protocol Error: {raw!r}")
253 
254         if isinstance(response, bytes) and disable_decoding is False:
255             response = self.encoder.decode(response)
256         return response
257 
258     def set_push_handler(self, push_handler_func):
259         self.push_handler_func = push_handler_func
260
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/redis/_parsers/resp3.py b/redis/_parsers/resp3.py
--- a/redis/_parsers/resp3.py
+++ b/redis/_parsers/resp3.py
@@ -96,8 +96,9 @@
                 pass
         # map response
         elif byte == b"%":
-            # we use this approach and not dict comprehension here
-            # because this dict comprehension fails in python 3.7
+            # We cannot use a dict-comprehension to parse stream.
+            # Evaluation order of key:val expression in dict comprehension only
+            # became defined to be left-right in version 3.8
             resp_dict = {}
             for _ in range(int(response)):
                 key = self._read_response(disable_decoding=disable_decoding)
@@ -225,12 +226,16 @@
                 pass
         # map response
         elif byte == b"%":
-            response = {
-                (await self._read_response(disable_decoding=disable_decoding)): (
-                    await self._read_response(disable_decoding=disable_decoding)
+            # We cannot use a dict-comprehension to parse stream.
+            # Evaluation order of key:val expression in dict comprehension only
+            # became defined to be left-right in version 3.8
+            resp_dict = {}
+            for _ in range(int(response)):
+                key = await self._read_response(disable_decoding=disable_decoding)
+                resp_dict[key] = await self._read_response(
+                    disable_decoding=disable_decoding, push_request=push_request
                 )
-                for _ in range(int(response))
-            }
+            response = resp_dict
         # push response
         elif byte == b">":
             response = [

{"golden_diff": "diff --git a/redis/_parsers/resp3.py b/redis/_parsers/resp3.py\n--- a/redis/_parsers/resp3.py\n+++ b/redis/_parsers/resp3.py\n@@ -96,8 +96,9 @@\n                 pass\n         # map response\n         elif byte == b\"%\":\n-            # we use this approach and not dict comprehension here\n-            # because this dict comprehension fails in python 3.7\n+            # We cannot use a dict-comprehension to parse stream.\n+            # Evaluation order of key:val expression in dict comprehension only\n+            # became defined to be left-right in version 3.8\n             resp_dict = {}\n             for _ in range(int(response)):\n                 key = self._read_response(disable_decoding=disable_decoding)\n@@ -225,12 +226,16 @@\n                 pass\n         # map response\n         elif byte == b\"%\":\n-            response = {\n-                (await self._read_response(disable_decoding=disable_decoding)): (\n-                    await self._read_response(disable_decoding=disable_decoding)\n+            # We cannot use a dict-comprehension to parse stream.\n+            # Evaluation order of key:val expression in dict comprehension only\n+            # became defined to be left-right in version 3.8\n+            resp_dict = {}\n+            for _ in range(int(response)):\n+                key = await self._read_response(disable_decoding=disable_decoding)\n+                resp_dict[key] = await self._read_response(\n+                    disable_decoding=disable_decoding, push_request=push_request\n                 )\n-                for _ in range(int(response))\n-            }\n+            response = resp_dict\n         # push response\n         elif byte == b\">\":\n             response = [\n", "issue": "resp3 parser async parser incorrectly parses maps\nIssue discovered in pr #2947\r\n\r\nthe evaulation order of `key:val` in `{key():val() for foo in bar}` dict comprehension is undefined in Python.  It mostly is left-right,\r\nbut some versions, notably recent 3.7 versions, don't behave that way.  It is best to use explicit ordering when reading from  a stream.\r\n\r\nThis had already been fixed for __non-async__ parsing, but in pr #2947 caused some scheduling behaviour which triggered this to happen. A fix for __async__ has been added there.\r\n\n", "before_files": [{"content": "from logging import getLogger\nfrom typing import Any, Union\n\nfrom ..exceptions import ConnectionError, InvalidResponse, ResponseError\nfrom ..typing import EncodableT\nfrom .base import _AsyncRESPBase, _RESPBase\nfrom .socket import SERVER_CLOSED_CONNECTION_ERROR\n\n\nclass _RESP3Parser(_RESPBase):\n    \"\"\"RESP3 protocol implementation\"\"\"\n\n    def __init__(self, socket_read_size):\n        super().__init__(socket_read_size)\n        self.push_handler_func = self.handle_push_response\n\n    def handle_push_response(self, response):\n        logger = getLogger(\"push_response\")\n        logger.info(\"Push response: \" + str(response))\n        return response\n\n    def read_response(self, disable_decoding=False, push_request=False):\n        pos = self._buffer.get_pos() if self._buffer else None\n        try:\n            result = self._read_response(\n                disable_decoding=disable_decoding, push_request=push_request\n            )\n        except BaseException:\n            if self._buffer:\n                self._buffer.rewind(pos)\n            raise\n        else:\n            self._buffer.purge()\n            return result\n\n    def _read_response(self, disable_decoding=False, push_request=False):\n        raw = self._buffer.readline()\n        if not raw:\n            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)\n\n        byte, response = raw[:1], raw[1:]\n\n        # server returned an error\n        if byte in (b\"-\", b\"!\"):\n            if byte == b\"!\":\n                response = self._buffer.read(int(response))\n            response = response.decode(\"utf-8\", errors=\"replace\")\n            error = self.parse_error(response)\n            # if the error is a ConnectionError, raise immediately so the user\n            # is notified\n            if isinstance(error, ConnectionError):\n                raise error\n            # otherwise, we're dealing with a ResponseError that might belong\n            # inside a pipeline response. the connection's read_response()\n            # and/or the pipeline's execute() will raise this error if\n            # necessary, so just return the exception instance here.\n            return error\n        # single value\n        elif byte == b\"+\":\n            pass\n        # null value\n        elif byte == b\"_\":\n            return None\n        # int and big int values\n        elif byte in (b\":\", b\"(\"):\n            return int(response)\n        # double value\n        elif byte == b\",\":\n            return float(response)\n        # bool value\n        elif byte == b\"#\":\n            return response == b\"t\"\n        # bulk response\n        elif byte == b\"$\":\n            response = self._buffer.read(int(response))\n        # verbatim string response\n        elif byte == b\"=\":\n            response = self._buffer.read(int(response))[4:]\n        # array response\n        elif byte == b\"*\":\n            response = [\n                self._read_response(disable_decoding=disable_decoding)\n                for _ in range(int(response))\n            ]\n        # set response\n        elif byte == b\"~\":\n            # redis can return unhashable types (like dict) in a set,\n            # so we need to first convert to a list, and then try to convert it to a set\n            response = [\n                self._read_response(disable_decoding=disable_decoding)\n                for _ in range(int(response))\n            ]\n            try:\n                response = set(response)\n            except TypeError:\n                pass\n        # map response\n        elif byte == b\"%\":\n            # we use this approach and not dict comprehension here\n            # because this dict comprehension fails in python 3.7\n            resp_dict = {}\n            for _ in range(int(response)):\n                key = self._read_response(disable_decoding=disable_decoding)\n                resp_dict[key] = self._read_response(\n                    disable_decoding=disable_decoding, push_request=push_request\n                )\n            response = resp_dict\n        # push response\n        elif byte == b\">\":\n            response = [\n                self._read_response(\n                    disable_decoding=disable_decoding, push_request=push_request\n                )\n                for _ in range(int(response))\n            ]\n            res = self.push_handler_func(response)\n            if not push_request:\n                return self._read_response(\n                    disable_decoding=disable_decoding, push_request=push_request\n                )\n            else:\n                return res\n        else:\n            raise InvalidResponse(f\"Protocol Error: {raw!r}\")\n\n        if isinstance(response, bytes) and disable_decoding is False:\n            response = self.encoder.decode(response)\n        return response\n\n    def set_push_handler(self, push_handler_func):\n        self.push_handler_func = push_handler_func\n\n\nclass _AsyncRESP3Parser(_AsyncRESPBase):\n    def __init__(self, socket_read_size):\n        super().__init__(socket_read_size)\n        self.push_handler_func = self.handle_push_response\n\n    def handle_push_response(self, response):\n        logger = getLogger(\"push_response\")\n        logger.info(\"Push response: \" + str(response))\n        return response\n\n    async def read_response(\n        self, disable_decoding: bool = False, push_request: bool = False\n    ):\n        if self._chunks:\n            # augment parsing buffer with previously read data\n            self._buffer += b\"\".join(self._chunks)\n            self._chunks.clear()\n        self._pos = 0\n        response = await self._read_response(\n            disable_decoding=disable_decoding, push_request=push_request\n        )\n        # Successfully parsing a response allows us to clear our parsing buffer\n        self._clear()\n        return response\n\n    async def _read_response(\n        self, disable_decoding: bool = False, push_request: bool = False\n    ) -> Union[EncodableT, ResponseError, None]:\n        if not self._stream or not self.encoder:\n            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)\n        raw = await self._readline()\n        response: Any\n        byte, response = raw[:1], raw[1:]\n\n        # if byte not in (b\"-\", b\"+\", b\":\", b\"$\", b\"*\"):\n        #     raise InvalidResponse(f\"Protocol Error: {raw!r}\")\n\n        # server returned an error\n        if byte in (b\"-\", b\"!\"):\n            if byte == b\"!\":\n                response = await self._read(int(response))\n            response = response.decode(\"utf-8\", errors=\"replace\")\n            error = self.parse_error(response)\n            # if the error is a ConnectionError, raise immediately so the user\n            # is notified\n            if isinstance(error, ConnectionError):\n                self._clear()  # Successful parse\n                raise error\n            # otherwise, we're dealing with a ResponseError that might belong\n            # inside a pipeline response. the connection's read_response()\n            # and/or the pipeline's execute() will raise this error if\n            # necessary, so just return the exception instance here.\n            return error\n        # single value\n        elif byte == b\"+\":\n            pass\n        # null value\n        elif byte == b\"_\":\n            return None\n        # int and big int values\n        elif byte in (b\":\", b\"(\"):\n            return int(response)\n        # double value\n        elif byte == b\",\":\n            return float(response)\n        # bool value\n        elif byte == b\"#\":\n            return response == b\"t\"\n        # bulk response\n        elif byte == b\"$\":\n            response = await self._read(int(response))\n        # verbatim string response\n        elif byte == b\"=\":\n            response = (await self._read(int(response)))[4:]\n        # array response\n        elif byte == b\"*\":\n            response = [\n                (await self._read_response(disable_decoding=disable_decoding))\n                for _ in range(int(response))\n            ]\n        # set response\n        elif byte == b\"~\":\n            # redis can return unhashable types (like dict) in a set,\n            # so we need to first convert to a list, and then try to convert it to a set\n            response = [\n                (await self._read_response(disable_decoding=disable_decoding))\n                for _ in range(int(response))\n            ]\n            try:\n                response = set(response)\n            except TypeError:\n                pass\n        # map response\n        elif byte == b\"%\":\n            response = {\n                (await self._read_response(disable_decoding=disable_decoding)): (\n                    await self._read_response(disable_decoding=disable_decoding)\n                )\n                for _ in range(int(response))\n            }\n        # push response\n        elif byte == b\">\":\n            response = [\n                (\n                    await self._read_response(\n                        disable_decoding=disable_decoding, push_request=push_request\n                    )\n                )\n                for _ in range(int(response))\n            ]\n            res = self.push_handler_func(response)\n            if not push_request:\n                return await self._read_response(\n                    disable_decoding=disable_decoding, push_request=push_request\n                )\n            else:\n                return res\n        else:\n            raise InvalidResponse(f\"Protocol Error: {raw!r}\")\n\n        if isinstance(response, bytes) and disable_decoding is False:\n            response = self.encoder.decode(response)\n        return response\n\n    def set_push_handler(self, push_handler_func):\n        self.push_handler_func = push_handler_func\n", "path": "redis/_parsers/resp3.py"}], "after_files": [{"content": "from logging import getLogger\nfrom typing import Any, Union\n\nfrom ..exceptions import ConnectionError, InvalidResponse, ResponseError\nfrom ..typing import EncodableT\nfrom .base import _AsyncRESPBase, _RESPBase\nfrom .socket import SERVER_CLOSED_CONNECTION_ERROR\n\n\nclass _RESP3Parser(_RESPBase):\n    \"\"\"RESP3 protocol implementation\"\"\"\n\n    def __init__(self, socket_read_size):\n        super().__init__(socket_read_size)\n        self.push_handler_func = self.handle_push_response\n\n    def handle_push_response(self, response):\n        logger = getLogger(\"push_response\")\n        logger.info(\"Push response: \" + str(response))\n        return response\n\n    def read_response(self, disable_decoding=False, push_request=False):\n        pos = self._buffer.get_pos() if self._buffer else None\n        try:\n            result = self._read_response(\n                disable_decoding=disable_decoding, push_request=push_request\n            )\n        except BaseException:\n            if self._buffer:\n                self._buffer.rewind(pos)\n            raise\n        else:\n            self._buffer.purge()\n            return result\n\n    def _read_response(self, disable_decoding=False, push_request=False):\n        raw = self._buffer.readline()\n        if not raw:\n            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)\n\n        byte, response = raw[:1], raw[1:]\n\n        # server returned an error\n        if byte in (b\"-\", b\"!\"):\n            if byte == b\"!\":\n                response = self._buffer.read(int(response))\n            response = response.decode(\"utf-8\", errors=\"replace\")\n            error = self.parse_error(response)\n            # if the error is a ConnectionError, raise immediately so the user\n            # is notified\n            if isinstance(error, ConnectionError):\n                raise error\n            # otherwise, we're dealing with a ResponseError that might belong\n            # inside a pipeline response. the connection's read_response()\n            # and/or the pipeline's execute() will raise this error if\n            # necessary, so just return the exception instance here.\n            return error\n        # single value\n        elif byte == b\"+\":\n            pass\n        # null value\n        elif byte == b\"_\":\n            return None\n        # int and big int values\n        elif byte in (b\":\", b\"(\"):\n            return int(response)\n        # double value\n        elif byte == b\",\":\n            return float(response)\n        # bool value\n        elif byte == b\"#\":\n            return response == b\"t\"\n        # bulk response\n        elif byte == b\"$\":\n            response = self._buffer.read(int(response))\n        # verbatim string response\n        elif byte == b\"=\":\n            response = self._buffer.read(int(response))[4:]\n        # array response\n        elif byte == b\"*\":\n            response = [\n                self._read_response(disable_decoding=disable_decoding)\n                for _ in range(int(response))\n            ]\n        # set response\n        elif byte == b\"~\":\n            # redis can return unhashable types (like dict) in a set,\n            # so we need to first convert to a list, and then try to convert it to a set\n            response = [\n                self._read_response(disable_decoding=disable_decoding)\n                for _ in range(int(response))\n            ]\n            try:\n                response = set(response)\n            except TypeError:\n                pass\n        # map response\n        elif byte == b\"%\":\n            # We cannot use a dict-comprehension to parse stream.\n            # Evaluation order of key:val expression in dict comprehension only\n            # became defined to be left-right in version 3.8\n            resp_dict = {}\n            for _ in range(int(response)):\n                key = self._read_response(disable_decoding=disable_decoding)\n                resp_dict[key] = self._read_response(\n                    disable_decoding=disable_decoding, push_request=push_request\n                )\n            response = resp_dict\n        # push response\n        elif byte == b\">\":\n            response = [\n                self._read_response(\n                    disable_decoding=disable_decoding, push_request=push_request\n                )\n                for _ in range(int(response))\n            ]\n            res = self.push_handler_func(response)\n            if not push_request:\n                return self._read_response(\n                    disable_decoding=disable_decoding, push_request=push_request\n                )\n            else:\n                return res\n        else:\n            raise InvalidResponse(f\"Protocol Error: {raw!r}\")\n\n        if isinstance(response, bytes) and disable_decoding is False:\n            response = self.encoder.decode(response)\n        return response\n\n    def set_push_handler(self, push_handler_func):\n        self.push_handler_func = push_handler_func\n\n\nclass _AsyncRESP3Parser(_AsyncRESPBase):\n    def __init__(self, socket_read_size):\n        super().__init__(socket_read_size)\n        self.push_handler_func = self.handle_push_response\n\n    def handle_push_response(self, response):\n        logger = getLogger(\"push_response\")\n        logger.info(\"Push response: \" + str(response))\n        return response\n\n    async def read_response(\n        self, disable_decoding: bool = False, push_request: bool = False\n    ):\n        if self._chunks:\n            # augment parsing buffer with previously read data\n            self._buffer += b\"\".join(self._chunks)\n            self._chunks.clear()\n        self._pos = 0\n        response = await self._read_response(\n            disable_decoding=disable_decoding, push_request=push_request\n        )\n        # Successfully parsing a response allows us to clear our parsing buffer\n        self._clear()\n        return response\n\n    async def _read_response(\n        self, disable_decoding: bool = False, push_request: bool = False\n    ) -> Union[EncodableT, ResponseError, None]:\n        if not self._stream or not self.encoder:\n            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)\n        raw = await self._readline()\n        response: Any\n        byte, response = raw[:1], raw[1:]\n\n        # if byte not in (b\"-\", b\"+\", b\":\", b\"$\", b\"*\"):\n        #     raise InvalidResponse(f\"Protocol Error: {raw!r}\")\n\n        # server returned an error\n        if byte in (b\"-\", b\"!\"):\n            if byte == b\"!\":\n                response = await self._read(int(response))\n            response = response.decode(\"utf-8\", errors=\"replace\")\n            error = self.parse_error(response)\n            # if the error is a ConnectionError, raise immediately so the user\n            # is notified\n            if isinstance(error, ConnectionError):\n                self._clear()  # Successful parse\n                raise error\n            # otherwise, we're dealing with a ResponseError that might belong\n            # inside a pipeline response. the connection's read_response()\n            # and/or the pipeline's execute() will raise this error if\n            # necessary, so just return the exception instance here.\n            return error\n        # single value\n        elif byte == b\"+\":\n            pass\n        # null value\n        elif byte == b\"_\":\n            return None\n        # int and big int values\n        elif byte in (b\":\", b\"(\"):\n            return int(response)\n        # double value\n        elif byte == b\",\":\n            return float(response)\n        # bool value\n        elif byte == b\"#\":\n            return response == b\"t\"\n        # bulk response\n        elif byte == b\"$\":\n            response = await self._read(int(response))\n        # verbatim string response\n        elif byte == b\"=\":\n            response = (await self._read(int(response)))[4:]\n        # array response\n        elif byte == b\"*\":\n            response = [\n                (await self._read_response(disable_decoding=disable_decoding))\n                for _ in range(int(response))\n            ]\n        # set response\n        elif byte == b\"~\":\n            # redis can return unhashable types (like dict) in a set,\n            # so we need to first convert to a list, and then try to convert it to a set\n            response = [\n                (await self._read_response(disable_decoding=disable_decoding))\n                for _ in range(int(response))\n            ]\n            try:\n                response = set(response)\n            except TypeError:\n                pass\n        # map response\n        elif byte == b\"%\":\n            # We cannot use a dict-comprehension to parse stream.\n            # Evaluation order of key:val expression in dict comprehension only\n            # became defined to be left-right in version 3.8\n            resp_dict = {}\n            for _ in range(int(response)):\n                key = await self._read_response(disable_decoding=disable_decoding)\n                resp_dict[key] = await self._read_response(\n                    disable_decoding=disable_decoding, push_request=push_request\n                )\n            response = resp_dict\n        # push response\n        elif byte == b\">\":\n            response = [\n                (\n                    await self._read_response(\n                        disable_decoding=disable_decoding, push_request=push_request\n                    )\n                )\n                for _ in range(int(response))\n            ]\n            res = self.push_handler_func(response)\n            if not push_request:\n                return await self._read_response(\n                    disable_decoding=disable_decoding, push_request=push_request\n                )\n            else:\n                return res\n        else:\n            raise InvalidResponse(f\"Protocol Error: {raw!r}\")\n\n        if isinstance(response, bytes) and disable_decoding is False:\n            response = self.encoder.decode(response)\n        return response\n\n    def set_push_handler(self, push_handler_func):\n        self.push_handler_func = push_handler_func\n", "path": "redis/_parsers/resp3.py"}]}