Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
try:
req = msgpack.unpackb(data, raw=False)
except Exception:
raise MSGPACKRPCParseError()
if not isinstance(req, list):
raise MSGPACKRPCInvalidRequestError()
if len(req) < 2:
raise MSGPACKRPCInvalidRequestError()
if req[0] == 0:
# MSGPACK request
request_id = req[1]
if not isinstance(request_id, int):
raise MSGPACKRPCInvalidRequestError()
if len(req) == 4:
return self._parse_request(req)
else:
raise MSGPACKRPCInvalidRequestError(request_id=request_id)
elif req[0] == 2:
# MSGPACK notification
if len(req) == 3:
return self._parse_notification(req)
else:
raise MSGPACKRPCInvalidRequestError()
else:
raise MSGPACKRPCInvalidRequestError()
def _parse_request(self, req):
if not isinstance(req[2], six.string_types):
raise MSGPACKRPCInvalidRequestError(request_id=req[1])
request = MSGPACKRPCRequest()
request.one_way = False
request.method = req[2]
request.unique_id = req[1]
params = req[3]
# params should not be None according to the spec; if there are
# no params, an empty array must be used
if isinstance(params, list):
request.args = params
else:
raise MSGPACKRPCInvalidParamsError(request_id=req[1])
return request
def _parse_notification(self, req):
if not isinstance(req[1], six.string_types):
raise MSGPACKRPCInvalidRequestError()
request = MSGPACKRPCRequest()
request.one_way = True
request.method = req[1]
params = req[2]
# params should not be None according to the spec; if there are
# no params, an empty array must be used
if isinstance(params, list):
request.args = params
else:
raise MSGPACKRPCInvalidParamsError(request_id=req[1])
return request
def _get_code_and_message(error):
assert isinstance(error, (Exception, six.string_types))
if isinstance(error, Exception):
if hasattr(error, "msgpackrpc_error_code"):
code = error.msgpackrpc_error_code
msg = str(error)
elif isinstance(error, InvalidRequestError):
code = MSGPACKRPCInvalidRequestError.msgpackrpc_error_code
msg = MSGPACKRPCInvalidRequestError.message
elif isinstance(error, MethodNotFoundError):
code = MSGPACKRPCMethodNotFoundError.msgpackrpc_error_code
msg = MSGPACKRPCMethodNotFoundError.message
else:
# allow exception message to propagate
code = MSGPACKRPCServerError.msgpackrpc_error_code
msg = str(error)
else:
code = -32000
msg = error
return code, msg
def _get_code_and_message(error):
assert isinstance(error, (Exception, six.string_types))
if isinstance(error, Exception):
if hasattr(error, "msgpackrpc_error_code"):
code = error.msgpackrpc_error_code
msg = str(error)
elif isinstance(error, InvalidRequestError):
code = MSGPACKRPCInvalidRequestError.msgpackrpc_error_code
msg = MSGPACKRPCInvalidRequestError.message
elif isinstance(error, MethodNotFoundError):
code = MSGPACKRPCMethodNotFoundError.msgpackrpc_error_code
msg = MSGPACKRPCMethodNotFoundError.message
else:
# allow exception message to propagate
code = MSGPACKRPCServerError.msgpackrpc_error_code
msg = str(error)
else:
code = -32000
msg = error
return code, msg
Called by the client when constructing a request.
MSGPACK-RPC allows only the ``args`` argument to be set; keyword
arguments are not supported.
:param str method: The method name to invoke.
:param list args: The positional arguments to call the method with.
:param dict kwargs: The keyword arguments to call the method with; must
be ``None`` as the protocol does not support keyword arguments.
:param bool one_way: The request is an update, i.e. it does not expect a reply.
:return: A new request instance
:rtype: :py:class:`MSGPACKRPCRequest`
:raises InvalidRequestError: when ``kwargs`` is defined.
"""
if kwargs:
raise MSGPACKRPCInvalidRequestError("Does not support kwargs")
request = self.request_factory()
request.one_way = one_way
if not one_way:
request.unique_id = self._get_unique_id()
request.method = method
request.args = list(args) if args is not None else []
request.kwargs = None
return request