Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
message = 'Invalid params'
class JSONRPCInternalError(FixedErrorMessageMixin, InvalidRequestError):
"""Unspecified error, not in the called function."""
jsonrpc_error_code = -32603
message = 'Internal error'
class JSONRPCServerError(FixedErrorMessageMixin, InvalidRequestError):
"""Unspecified error, this message originates from the called function."""
jsonrpc_error_code = -32000
message = ''
class JSONRPCError(FixedErrorMessageMixin, RPCError):
"""Reconstructs (to some extend) the server-side exception.
The client creates this exception by providing it with the ``error``
attribute of the JSON error response object returned by the server.
:param dict error: This dict contains the error specification:
* code (int): the numeric error code.
* message (str): the error description.
* data (any): if present, the data attribute of the error
"""
def __init__(
self, error: Union['JSONRPCErrorResponse', Dict[str, Any]]
) -> None:
if isinstance(error, JSONRPCErrorResponse):
super(JSONRPCError, self).__init__(error.error)
class MSGPACKRPCInvalidParamsError(FixedErrorMessageMixin, InvalidRequestError):
msgpackrpc_error_code = -32602
message = "Invalid params"
class MSGPACKRPCInternalError(FixedErrorMessageMixin, InvalidRequestError):
msgpackrpc_error_code = -32603
message = "Internal error"
class MSGPACKRPCServerError(FixedErrorMessageMixin, InvalidRequestError):
msgpackrpc_error_code = -32000
message = ""
class MSGPACKRPCError(FixedErrorMessageMixin, RPCError):
"""Reconstructs (to some extend) the server-side exception.
The client creates this exception by providing it with the ``error``
attribute of the MSGPACK error response object returned by the server.
:param error: This tuple contains the error specification: the numeric error
code and the error description.
"""
def __init__(
self, error: Union["MSGPACKRPCErrorResponse", Tuple[int, str]]
) -> None:
if isinstance(error, MSGPACKRPCErrorResponse):
super().__init__(error.error)
self._msgpackrpc_error_code = error._msgpackrpc_error_code
else:
def handle_client(message):
try:
request = protocol.parse_request(message[-1])
except RPCError as e:
log.exception(e)
response = e.error_respond()
else:
response = dispatcher.dispatch(request)
log.debug('Response okay: %r', response)
# send reply
message[-1] = response.serialize()
log.debug('Replying %s to %r', message[-1], message[0])
socket.send_multipart(message)
"""
if isinstance(data, bytes):
data = data.decode()
try:
req = json.loads(data)
except Exception as e:
raise JSONRPCParseError()
if isinstance(req, list):
# batch request
requests = JSONRPCBatchRequest()
for subreq in req:
try:
requests.append(self._parse_subrequest(subreq))
except RPCError as e:
requests.append(e)
except Exception as e:
requests.append(JSONRPCInvalidRequestError(request_id=subreq.get("id")))
if not requests:
raise JSONRPCInvalidRequestError()
return requests
else:
return self._parse_subrequest(req)