(self)
| 160 | return response |
| 161 | |
| 162 | def _get_response(self): |
| 163 | req_start_time = time.time() |
| 164 | try: |
| 165 | http_response = self.__conn.getresponse() |
| 166 | except socket.timeout: |
| 167 | raise JSONRPCException({ |
| 168 | 'code': -344, |
| 169 | 'message': '%r RPC took longer than %f seconds. Consider ' |
| 170 | 'using larger timeout for calls that take ' |
| 171 | 'longer to return.' % (self._service_name, |
| 172 | self.__conn.timeout)}) from None |
| 173 | if http_response is None: |
| 174 | raise JSONRPCException({ |
| 175 | 'code': -342, 'message': 'missing HTTP response from server'}) |
| 176 | |
| 177 | # Check for no-content HTTP status code, which can be returned when an |
| 178 | # RPC client requests a JSON-RPC 2.0 "notification" with no response. |
| 179 | # Currently this is only possible if clients call the _request() method |
| 180 | # directly to send a raw request. |
| 181 | if http_response.status == HTTPStatus.NO_CONTENT: |
| 182 | if len(http_response.read()) != 0: |
| 183 | raise JSONRPCException({'code': -342, 'message': 'Content received with NO CONTENT status code'}) |
| 184 | return None, http_response.status |
| 185 | |
| 186 | content_type = http_response.getheader('Content-Type') |
| 187 | if content_type != 'application/json': |
| 188 | raise JSONRPCException( |
| 189 | {'code': -342, 'message': f"non-JSON HTTP response with \'{http_response.status} {http_response.reason}\' from server: {http_response.read().decode()}"}, |
| 190 | http_response.status) |
| 191 | |
| 192 | data = http_response.read() |
| 193 | try: |
| 194 | responsedata = data.decode('utf8') |
| 195 | except UnicodeDecodeError as e: |
| 196 | raise JSONRPCException({ |
| 197 | 'code': -342, 'message': f'Cannot decode response in utf8 format, content: {data}, exception: {e}'}) |
| 198 | response = json.loads(responsedata, parse_float=decimal.Decimal) |
| 199 | elapsed = time.time() - req_start_time |
| 200 | if "error" in response and response["error"] is None: |
| 201 | log.debug("<-%s- [%.6f] %s" % (response["id"], elapsed, self._json_dumps(response["result"]))) |
| 202 | else: |
| 203 | log.debug("<-- [%.6f] %s" % (elapsed, responsedata)) |
| 204 | return response, http_response.status |
| 205 | |
| 206 | def __truediv__(self, relative_uri): |
| 207 | return AuthServiceProxy("{}/{}".format(self.__service_url, relative_uri), self._service_name, connection=self.__conn) |
no test coverage detected