"""
Json rpc de/serializer
"""
__author__ = 'VMware, Inc.'
__copyright__ = 'Copyright (c) 2015 VMware, Inc. All rights reserved.'
import decimal
import logging
import six
import base64
try:
import simplejson as json
except ImportError:
import json
from vmware.vapi.core import (
MethodResult, ExecutionContext, ApplicationContext, SecurityContext)
from vmware.vapi.data.type import Type
from vmware.vapi.data.value import (
data_value_factory, StructValue, ErrorValue, ListValue,
OptionalValue, VoidValue, StringValue, BooleanValue, IntegerValue,
DoubleValue, BlobValue, SecretValue)
from vmware.vapi.lib.jsonlib import canonicalize_double
logger = logging.getLogger(__name__)
# Vapi json rpc version
VAPI_JSONRPC_VERSION = '2.0'
# Vapi json rpc method names
VAPI_INVOKE = 'invoke'
vapi_to_json_map = {
Type.BLOB: 'BINARY',
Type.SECRET: 'SECRET',
Type.ERROR: 'ERROR',
Type.STRUCTURE: 'STRUCTURE',
Type.OPTIONAL: 'OPTIONAL',
}
json_to_vapi_map = dict((v, k) for k, v in six.iteritems(vapi_to_json_map))
[docs]class VAPIJsonEncoder(json.JSONEncoder):
"""
Custom JSON encoder that converts vAPI runtime types directly
into JSON string representation.
"""
# Even though __init__ is called below, pylint throws a warning
# that base class __init__ is not called (W0231)
def __init__(self, *args, **kwargs): # pylint: disable=W0231
self._dispatch_map = {
MethodResult: self.visit_method_result,
ExecutionContext: self.visit_execution_context,
dict: self.visit_dict,
list: self.visit_list,
decimal.Decimal: canonicalize_double,
StructValue: self.visit_struct_value,
ErrorValue: self.visit_struct_value,
ListValue: self.visit_list,
OptionalValue: self.visit_optional_value,
DoubleValue: self.visit_double_value,
IntegerValue: self.visit_primitive_value,
BooleanValue: self.visit_primitive_value,
StringValue: self.visit_primitive_value,
VoidValue: self.visit_primitive_value,
BlobValue: self.visit_blob_value_with_type_name,
SecretValue: self.visit_primitive_value_with_type_name,
}
json.JSONEncoder.__init__(self, *args, **kwargs)
#
# Even though the recommended way to subclass JSONEncoder is by overriding
# "default", encode is being used as it will preserve the computed JSON
# string literal for decimals. If we use default, the value returned by
# canonicalize_double will be wrapped inside double quotes in the final
# JSON message.
#
[docs] def encode(self, value):
"""
Encode a given vAPI runtime object
:type value: :class:`object`
:param value: vAPI runtime object
:rtype: :class:`str`
:return: JSON string
"""
return self._dispatch_map.get(type(value), self.visit_default)(value)
[docs] def visit_struct_value(self, value):
"""
Visit a StructValue object
:type value: :class:`vmware.vapi.data.value.StructValue`
:param value: Struct value object
:rtype: :class:`str`
:return: JSON string
"""
type_name = vapi_to_json_map.get(value.type)
items = ['"%s":%s' % (k, self.encode(v))
for k, v in value.get_fields()]
return '{"%s":{"%s":{%s}}}' % (
type_name,
value.name,
','.join(items))
[docs] def visit_list(self, value):
"""
Visit a ListValue object
:type value: :class:`vmware.vapi.data.value.ListValue`
:param value: List value object
:rtype: :class:`str`
:return: JSON string
"""
string = ','.join([self.encode(item)
for item in value])
return '[%s]' % string
[docs] def visit_optional_value(self, value):
"""
Visit a OptionalValue object
:type value: :class:`vmware.vapi.data.value.OptionalValue`
:param value: Optional value object
:rtype: :class:`str`
:return: JSON string
"""
type_name = vapi_to_json_map.get(value.type)
if value.is_set():
return '{"%s":%s}' % (type_name, self.encode(value.value))
else:
return '{"%s":null}' % type_name
@staticmethod
[docs] def visit_double_value(value):
"""
Visit a DoubleValue object
:type value: :class:`vmware.vapi.data.value.DoubleValue`
:param value: Double value object
:rtype: :class:`str`
:return: JSON string
"""
return canonicalize_double(value.value)
[docs] def visit_primitive_value(self, value):
"""
Visit one of StringValue, IntegerValue, BooleanValue or VoidValue
:type value: :class:`vmware.vapi.data.value.StringValue` (or)
:class:`vmware.vapi.data.value.IntegerValue` (or)
:class:`vmware.vapi.data.value.BooleanValue` (or)
:class:`vmware.vapi.data.value.VoidValue` (or)
:param value: StringValue, IntegerValue, BooleanValue or VoidValue object
:rtype: :class:`str`
:return: JSON string
"""
return json.JSONEncoder.encode(self, value.value)
[docs] def visit_primitive_value_with_type_name(self, value):
"""
Visit SecretValue
:type value: :class:`vmware.vapi.data.value.SecretValue`
:param value: SecretValue object
:rtype: :class:`str`
:return: JSON string
"""
type_name = vapi_to_json_map.get(value.type)
return '{"%s":%s}' % (type_name, self.encode(value.value))
[docs] def visit_blob_value_with_type_name(self, value):
"""
Visit BlobValue
:type value: :class:`vmware.vapi.data.value.BlobValue`
:param value: BlobValue object
:rtype: :class:`str`
:return: JSON string
"""
type_name = vapi_to_json_map.get(value.type)
if six.PY3 and isinstance(value.value, str):
# For Python3 we need to convert str to byte string for b64encode
data_value = base64.b64encode(six.b(value.value))
else:
data_value = base64.b64encode(value.value)
return '{"%s":%s}' % (type_name, self.encode(data_value))
[docs] def visit_method_result(self, value):
"""
Visit a MethodResult object
:type value: :class:`vmware.vapi.core.MethodResult`
:param value: MethodResult object
:rtype: :class:`str`
:return: JSON string
"""
if value.error is not None:
return '{"%s":%s}' % ('error', self.encode(value.error))
elif value.output is not None:
return '{"%s":%s}' % ('output', self.encode(value.output))
[docs] def visit_execution_context(self, value):
"""
Visit an ExecutionContext object
:type value: :class:`vmware.vapi.core.ExecutionContext`
:param value: ExecutionContext object
:rtype: :class:`str`
:return: JSON string
"""
if value.security_context:
return '{"%s":%s,"%s":%s}' % (
'appCtx',
self.encode(value.application_context),
'securityCtx',
self.encode(value.security_context))
else:
return '{"%s":%s}' % ('appCtx',
self.encode(value.application_context))
[docs] def visit_dict(self, value):
"""
Visit a dictionary. Application context and Security Context
in vAPI is a free form object, so it can contain a dictionary.
:type value: :class:`dict`
:param value: Dictionary value
:rtype: :class:`str`
:return: JSON string
"""
items = ['%s:%s' % (self.encode(k), self.encode(v))
for k, v in six.iteritems(value)]
return '{%s}' % ','.join(items)
[docs] def visit_default(self, value):
"""
This is the default visit method if the type of the input value
does not match any type in the keys present in dispatch map.
:type value: :class:`object`
:param value: Python object
:rtype: :class:`str`
:return: JSON string
"""
return json.JSONEncoder.encode(self, value)
[docs]class JsonRpcDictToVapi(object):
""" Json rpc dict to vapi type """
def __init__(self):
""" Json rpc dict to vapi type init """
pass
@staticmethod
[docs] def data_value(value):
"""
get data value from new jsonrpc dict
# TODO: Structure names and its fields are converted from
# u'' format to str format. This will break if we allow non
# ASCII characters in the IDL
:type value: :class:`dict`
:param value: json data value
:rtype: subclass of :class:`vmware.vapi.data.value.DataValue`
:return: subclass of data value
"""
result = None
if value is None:
result = data_value_factory(Type.VOID)
elif isinstance(value, dict):
# Optional, Secret, Blob, Error, Structure
# types are inside object
if len(value) > 1:
raise vapi_jsonrpc_error_invalid_params()
(type_name, json_value) = value.popitem()
typ = json_to_vapi_map[type_name]
if typ == Type.SECRET:
result = data_value_factory(typ, json_value)
elif typ == Type.BLOB:
try:
result = data_value_factory(typ, base64.b64decode(json_value))
except Exception as err:
if six.PY2 and isinstance(err, TypeError):
raise vapi_jsonrpc_error_internal_error()
elif six.PY3:
import binascii
if isinstance(err, binascii.Error):
raise vapi_jsonrpc_error_internal_error()
raise
elif typ in [Type.ERROR, Type.STRUCTURE]:
if len(json_value) > 1:
raise vapi_jsonrpc_error_invalid_params()
struct_name, fields = json_value.popitem()
result = data_value_factory(typ, str(struct_name))
for field_name, field_value in six.iteritems(fields):
field_data_value = JsonRpcDictToVapi.data_value(
field_value)
result.set_field(str(field_name), field_data_value)
elif typ == Type.OPTIONAL:
result = data_value_factory(typ)
if json_value is not None:
result.value = JsonRpcDictToVapi.data_value(json_value)
elif isinstance(value, list):
# ListValue is json list
list_data_values = [JsonRpcDictToVapi.data_value(val)
for val in value]
result = data_value_factory(Type.LIST, list_data_values)
elif isinstance(value, bool):
result = data_value_factory(Type.BOOLEAN, value)
elif isinstance(value, six.string_types):
result = data_value_factory(Type.STRING, value)
elif isinstance(value, six.integer_types):
result = data_value_factory(Type.INTEGER, value)
else:
# For Double
result = data_value_factory(Type.DOUBLE, value)
return result
@staticmethod
[docs] def error_value(value):
"""
get error value from jsonrpc dict
:type msg: :class:`dict`
:param msg: json error value
:rtype: :class:`vmware.vapi.data.value.ErrorValue`
:return: error value
"""
return JsonRpcDictToVapi.data_value(value)
@staticmethod
[docs] def method_result(result):
"""
get method result from jsonrpc dict
:type result: :class:`dict`
:param result: json method result
:rtype: :class:`vmware.vapi.core.MethodResult`
:return: method result
"""
output = None
if 'output' in result:
output = JsonRpcDictToVapi.data_value(result['output'])
error = None
if 'error' in result:
error = JsonRpcDictToVapi.error_value(result['error'])
return MethodResult(output=output, error=error)
@staticmethod
[docs] def security_ctx(ctx):
"""
get security context from jsonrpc dict
:type ctx: :class:`dict`
:param ctx: json security context
:rtype: :class:`vmware.vapi.core.SecurityContext`
:return: json user session
"""
if ctx is not None:
return SecurityContext(ctx)
@staticmethod
[docs] def app_ctx(ctx):
"""
get application context from jsonrpc dict
:type ctx: :class:`dict`
:param ctx: json application context
:rtype: :class:`str`
:return: operation identifier
"""
return ApplicationContext(ctx)
@staticmethod
[docs] def execution_context(ctx):
"""
get execution context from jsonrpc dict
:type ctx: :class:`dict`
:param ctx: json execution context
:rtype: :class:`vmware.vapi.core.ExecutionContext`
:return: execution context
"""
app_ctx = JsonRpcDictToVapi.app_ctx(ctx.get('appCtx'))
security_ctx = JsonRpcDictToVapi.security_ctx(ctx.get('securityCtx'))
execution_context = ExecutionContext(app_ctx, security_ctx)
return execution_context
[docs]class JsonRpcError(Exception):
""" json rpc error base class """
def __init__(self, error):
"""
json rpc error base class constructor
:type error: :class:`dict`
:param error: json rpc error
"""
Exception.__init__(self)
self.error = error
[docs]class JsonRpcRequest(object): # pylint: disable=R0922
""" Json rpc request base class """
def __init__(self, version, method, params=None, notification=True, request_id=None):
"""
Json rpc request base class constructor
:type version: :class:`str`
:kwarg version: json rpc request version
:type method: :class:`str`
:kwarg method: json rpc method
:type params: :class:`dict` or :class:`list` or None
:kwarg params: json rpc method params
:type notification: :class:`bool`
:kwarg notification: True for notification
:type request_id: :class:`long` or :class:`int` or :class:`str` or None
:kwarg request_id: json rpc request id. Do not set for notification
"""
self.version = version
if not isinstance(method, six.string_types):
raise vapi_jsonrpc_error_invalid_request()
self.method = method
self.params = params
self.notification = notification
self.id = request_id
[docs] def serialize(self):
"""
Serialize a json rpc request
:rtype: :class:`str`
:return: json rpc request str
"""
raise NotImplementedError
[docs] def validate_response(self, response):
"""
Validate a json rpc response.
Check for version / id mismatch with request
:type response: :class:`JsonRpcResponse`
:param response: json rpc response object to validate
"""
raise NotImplementedError
[docs]class JsonRpcResponse(object): # pylint: disable=R0922
""" Json rpc response base class """
def __init__(self, version, response_id, result, error=None):
"""
Json rpc response base class constructor
:type version: :class:`str`
:kwarg version: json rpc response version
:type response_id: :class:`long` or :class:`int` or :class:`str` or None
:kwarg response_id: json rpc response id
:type result: :class:`dict`
:kwarg result: json rpc response dict
:type error: :class:`JsonRpcError`
:kwarg error: json rpc error
"""
self.version = version
self.id = response_id
# result or error, but not both
if result is not None and error is not None:
logger.error('JSON RPC response has both result and error')
raise vapi_jsonrpc_error_invalid_params()
elif result is None and error is None:
logger.error('JSON RPC response has neither result nor error')
raise vapi_jsonrpc_error_invalid_params()
else:
self.result, self.error = result, error
[docs] def serialize(self):
"""
Serialize a json rpc response
:rtype: :class:`str`
:return: json rpc response str
"""
raise NotImplementedError
[docs]class JsonRpc20Error(JsonRpcError):
""" json rpc 2.0 error """
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
INTERNAL_ERROR = -32603
PARSE_ERROR = -32700
# TRANSPORT_ERROR is defined in xmlrpc error code
# http://xmlrpc-epi.sourceforge.net/specs/rfc.fault_codes.php
# but not JSON RPC 2.0
# Need this for server connection error
TRANSPORT_ERROR = -32300
SERVER_ERROR_RANGE_MIN = -32768
SERVER_ERROR_RANGE_MAX = -32000
DEFAULT_MESSAGES = {
INVALID_REQUEST: 'Invalid Request.',
METHOD_NOT_FOUND: 'Method not found.',
INVALID_PARAMS: 'Invalid params.',
INTERNAL_ERROR: 'Internal error.',
PARSE_ERROR: 'Parse error.',
TRANSPORT_ERROR: 'Transport error.',
}
def __init__(self, code, message=None, data=None):
"""
Json rpc 2.0 error
:type code: :class:`int`
:kwarg code: error type that occurred
:type message: :class:`str`
:kwarg message: a short description of the error
:type data: Primitive type / :class:`dict` / None
:kwarg data: more info about the error
"""
try:
code = int(code)
except ValueError:
logger.error('JSON RPC error code type: Expecting "int". Got "%s"',
str(type(code)))
raise
self.code = code
if message is None and (code >= self.SERVER_ERROR_RANGE_MIN or
code <= self.SERVER_ERROR_RANGE_MAX):
message = self.DEFAULT_MESSAGES.get(code, 'Server error.')
self.json_message = message
error = {'code': code, 'message': message}
if data:
error['data'] = str(data)
self.data = data
JsonRpcError.__init__(self, error)
def __str__(self):
return "%d: %s: %s" % (self.code, str(self.json_message), str(self.data))
[docs]def vapi_jsonrpc_error_parse_error(data=None):
"""
vapi json rpc parse error
:type data: :class:`dict`
:param data: json rpc error object
:rtype: :class:`JsonRpcError`
:return: json rpc error object
"""
return JsonRpc20Error(code=JsonRpc20Error.PARSE_ERROR, data=data)
[docs]def vapi_jsonrpc_error_invalid_request(data=None):
"""
vapi json rpc invalid request error
:type data: :class:`dict`
:param data: json rpc error object
:rtype: :class:`JsonRpcError`
:return: json rpc error object
"""
return JsonRpc20Error(code=JsonRpc20Error.INVALID_REQUEST, data=data)
[docs]def vapi_jsonrpc_error_method_not_found(data=None):
"""
vapi json rpc method not found error
:type data: :class:`dict`
:param data: json rpc error object
:rtype: :class:`JsonRpcError`
:return: json rpc error object
"""
return JsonRpc20Error(code=JsonRpc20Error.METHOD_NOT_FOUND, data=data)
[docs]def vapi_jsonrpc_error_invalid_params(data=None):
"""
vapi json rpc invalid params error
:type data: :class:`dict`
:param data: json rpc error object
:rtype: :class:`JsonRpcError`
:return: json rpc error object
"""
return JsonRpc20Error(code=JsonRpc20Error.INVALID_PARAMS, data=data)
[docs]def vapi_jsonrpc_error_internal_error(data=None):
"""
vapi json rpc internal error
:type data: :class:`dict`
:param data: json rpc error object
:rtype: :class:`JsonRpcError`
:return: json rpc error object
"""
return JsonRpc20Error(code=JsonRpc20Error.INTERNAL_ERROR, data=data)
[docs]def vapi_jsonrpc_error_transport_error(data=None):
"""
vapi json rpc transport error
:type data: :class:`dict`
:param data: json rpc error object
:rtype: :class:`JsonRpcError`
:return: json rpc error object
"""
return JsonRpc20Error(code=JsonRpc20Error.TRANSPORT_ERROR, data=data)
[docs]class JsonRpc20Request(JsonRpcRequest):
""" Json rpc 2.0 request """
def __init__(self, **kwargs):
"""
Json rpc 2.0 request init
:type jsonrpc: :class:`str` or None
:kwarg jsonrpc: json rpc request version
:type method: :class:`str`
:kwarg method: json rpc method
:type params: :class:`dict` or :class:`list` or None
:kwarg params: json rpc method params
:type id: :class:`long` or :class:`int` or :class:`str` or None
:kwarg id: json rpc request id. Do not set for notification
"""
version = kwargs.get('jsonrpc')
notification, id_ = True, None
if 'id' in kwargs:
notification, id_ = False, kwargs.get('id')
if (id_ and not (isinstance(id_, six.string_types)
or isinstance(id_, six.integer_types))):
logger.error(
'JSON RPC request id must be None/str/int. Get "%s"',
str(type(id_)))
raise vapi_jsonrpc_error_invalid_request()
method = kwargs.get('method')
params = kwargs.get('params')
if not (params is None or isinstance(params, dict) or
isinstance(params, list)):
raise vapi_jsonrpc_error_invalid_request()
JsonRpcRequest.__init__(self, version=version,
method=method,
params=params,
notification=notification,
request_id=id_)
[docs] def serialize(self):
"""
Serialize a json rpc 2.0 request
:rtype: :class:`str`
:return: json rpc request str
"""
result_dict = {'jsonrpc': self.version,
'method': self.method,
'params': self.params,
'id': self.id}
return json.dumps(result_dict,
check_circular=False,
separators=(',', ':'),
cls=VAPIJsonEncoder)
[docs] def serialize_notification(self):
"""
Serialize a json rpc 2.0 notification
:rtype: :class:`str`
:return: json rpc notification str
"""
result_dict = {'jsonrpc': self.version,
'method': self.method,
'params': self.params}
return json.dumps(result_dict,
check_circular=False,
separators=(',', ':'),
cls=VAPIJsonEncoder)
[docs] def validate_response(self, response):
"""
Validate a json rpc 2.0 response.
Check for version / id mismatch with request
:type response: :class:`JsonRpcResponse`
:param response: json rpc response object to validate
"""
if self.notification:
logger.error('JSON RPC notification does not have response')
raise vapi_jsonrpc_error_invalid_params()
# Check jsonrpc version
# NYI: backward compatible?
request_version = self.version
if response.version != request_version:
logger.error('JSON RPC incompatible version: "%s". Expecting "%s"',
str(response.version), str(request_version))
raise vapi_jsonrpc_error_invalid_params()
# Note: response_id MUST be Null for PARSE_ERROR / INVALID_REQUEST
# NYI: Relax this requirement?
if (response.error and
(response.error.code == JsonRpc20Error.PARSE_ERROR or
response.error.code == JsonRpc20Error.INVALID_REQUEST)):
id_ = None
else:
id_ = self.id
# Check request / response id
if id_ != response.id:
logger.error('JSON RPC response id mismatch: "%s". Expecting "%s"',
str(response.id), str(id_))
raise vapi_jsonrpc_error_invalid_params()
[docs]class JsonRpc20Response(JsonRpcResponse):
""" Json rpc 2.0 response """
def __init__(self, **kwargs):
"""
Json rpc 2.0 response init
Either result or error must be set, but not both
:type jsonrpc: :class:`str` or None
:kwarg jsonrpc: json rpc response version
:type result: :class:`dict`
:kwarg result: json rpc response dict
:type error: :class:`JsonRpcError` or dict
:kwarg error: json rpc response error
:type id: :class:`long` or :class:`int` or :class:`str`
:kwarg id: json rpc response id
"""
version = kwargs.get('jsonrpc')
id_ = kwargs.get('id')
# result or error
result = kwargs.get('result')
error = kwargs.get('error')
if error is not None:
if isinstance(error, JsonRpcError):
pass
elif isinstance(error, dict):
code = error.get('code')
message = error.get('message')
error = JsonRpc20Error(code=code, message=message, data=error)
else:
logger.error(
'JSON RPC params type error. Expecting "%s". Got "%s"',
'JsonRpcError / dict', str(type(error)))
raise vapi_jsonrpc_error_invalid_params()
JsonRpcResponse.__init__(self, version=version, response_id=id_,
result=result, error=error)
[docs] def serialize(self):
"""
Serialize a json rpc 2.0 response
:rtype: :class:`str`
:return: json rpc response str
"""
result_dict = {'jsonrpc': self.version,
'id': self.id}
if self.result is not None:
result_dict['result'] = self.result
else:
# result is None, error MUST not be None
result_dict['error'] = self.error.error
if (self.error.code == JsonRpc20Error.PARSE_ERROR or
self.error.code == JsonRpc20Error.INVALID_REQUEST):
result_dict['id'] = None
output = json.dumps(result_dict,
check_circular=False,
separators=(',', ':'),
cls=VAPIJsonEncoder)
output = output.encode('utf-8')
return output
[docs]def jsonrpc_request_factory(**kwargs):
"""
Json rpc request factory
For json 2.0: set jsonrpc to '2.0'
For json 1.1: set version to '1.1'
For json 1.0: set neither jsonrpc / version
The parameters accepted depends on json version
See corresponding json rpc request class for init parameters
:type jsonrpc: :class:`str` or None
:kwarg jsonrpc: json rpc request version (2.0)
:type version: :class:`str` or None
:kwarg version: json rpc request version (1.1)
:type method: :class:`str`
:kwarg method: json rpc method
:type params: :class:`dict` or :class:`list` or None
:kwarg params: json rpc method params
:type id: :class:`long` or :class:`int` or :class:`str` or None
:kwarg id: json rpc request id. Do not set for notification
:rtype: :class:`JsonRpcRequest`
:return: json rpc request object
"""
jsonrpc_version = kwargs.get('jsonrpc')
if jsonrpc_version and jsonrpc_version == '2.0':
# json rpc 2.0
request = JsonRpc20Request(**kwargs)
else:
jsonrpc_version = kwargs.get('version')
if jsonrpc_version and jsonrpc_version == '1.1':
# NYI: json rpc 1.1
# request = JsonRpc11Request(kwargs)
request = None
else:
# NYI: json rpc 1.0
# request = JsonRpc10Request(kwargs)
request = None
return request
[docs]def jsonrpc_response_factory(**kwargs):
"""
Json rpc response factory
For json 2.0: set jsonrpc to '2.0'
For json 1.1: set version to '1.1'
For json 1.0: set neither jsonrpc / version
The parameters accepted depends on json version
See corresponding json rpc response class for init parameters
:type jsonrpc: :class:`str` or None
:kwarg jsonrpc: json rpc response version (2.0)
:type version: :class:`str` or None
:kwarg version: json rpc response version (1.1)
:type method: :class:`str`
:kwarg method: json rpc method
:type params: :class:`dict` or :class:`list` or None
:kwarg params: json rpc method params
:type id: :class:`long` or :class:`int` or :class:`str`
:kwarg id: json rpc response id
:rtype: :class:`JsonRpcResponse`
:return: json rpc response object
"""
jsonrpc_version = kwargs.get('jsonrpc')
if jsonrpc_version and jsonrpc_version == '2.0':
# json rpc 2.0
response = JsonRpc20Response(**kwargs)
else:
jsonrpc_version = kwargs.get('version')
if jsonrpc_version and jsonrpc_version == '1.1':
# NYI: json rpc 1.1
# response = JsonRpc11Response(kwargs)
response = None
else:
# NYI: json rpc 1.0
# response = JsonRpc10Response(kwargs)
response = None
return response
[docs]def vapi_jsonrpc_request_factory(**kwargs):
"""
Json rpc request factory
:type method: :class:`str`
:kwarg method: json rpc method
:type params: :class:`dict` or :class:`list` or None
:kwarg params: json rpc method params
:type id: :class:`long` or :class:`int` or :class:`str` or None
:kwarg id: json rpc request id. Do not set for notification
:rtype: :class:`JsonRpcRequest`
:return: json rpc request object
"""
jsonrpc_version = kwargs.get('jsonrpc', VAPI_JSONRPC_VERSION)
# vapi json rpc supports version 2.0 only
assert(jsonrpc_version == VAPI_JSONRPC_VERSION)
kwargs.setdefault('jsonrpc', jsonrpc_version)
return JsonRpc20Request(**kwargs)
[docs]def vapi_jsonrpc_notification_factory(**kwargs):
"""
vapi json rpc notification factory
:type method: :class:`str`
:kwarg method: json rpc method
:type params: :class:`dict` or :class:`list` or None
:kwarg params: json rpc method params
:rtype: :class:`JsonRpcRequest`
:return: json rpc request object
"""
kwargs.pop('id', None)
return vapi_jsonrpc_request_factory(**kwargs)
[docs]def vapi_jsonrpc_response_factory(request, **kwargs):
"""
vapi json rpc response factory
:type request: :class:`JsonRpcRequest`
:kwarg request: json rpc request object
:type kwargs: :class:`dict`
:kwarg kwargs: See JsonRpc20Response for constructor parameters
:rtype: :class:`JsonRpcResponse`
:return: json rpc response object
"""
if request is not None and request.notification:
# Notification do not have response
logger.error('JSON RPC notificaition should not have id')
raise vapi_jsonrpc_error_invalid_params()
jsonrpc_version = kwargs.get('jsonrpc', VAPI_JSONRPC_VERSION)
# vapi json rpc supports version 2.0 only
assert(jsonrpc_version == VAPI_JSONRPC_VERSION)
kwargs.setdefault('jsonrpc', jsonrpc_version)
kwargs.setdefault('id', None if request is None else request.id)
return JsonRpc20Response(**kwargs)
[docs]def to_strkey_dict(dictionary):
"""
Convert a unicode key dict into str key dict
:type dictionary: :class:`dict`
:param dictionary: unicode dictionary
:rtype: :class:`dict`
:return: string key dict
"""
result = {}
for key, val in six.iteritems(dictionary):
result[str(key)] = val
return result
[docs]def deserialize_request(request_str):
"""
Deserialize a json rpc request
:type request_str: :class:`str` or :class:`bytes:` or :class:`file`
:param request_str: json rpc request str or a file like object
:rtype: :class:`JsonRpcRequest`
:return: json rpc request
"""
# NYI: Vapi object directly from json string? Not easy with standard json
# deserializer
try:
# In 2.x, we get str from underlying layer and in 3.x,
# we get bytes
if isinstance(request_str, six.string_types):
obj = json.loads(request_str, parse_float=decimal.Decimal)
elif isinstance(request_str, six.binary_type):
obj = json.loads(request_str.decode('utf-8'),
parse_float=decimal.Decimal)
else:
obj = json.load(request_str, parse_float=decimal.Decimal)
except ValueError as err:
logger.error('Deserialize request error %s', str(err))
raise vapi_jsonrpc_error_parse_error(data=str(err))
try:
request = jsonrpc_request_factory(**to_strkey_dict(obj))
except JsonRpcError as err:
raise
except Exception as err:
logger.error('Invalid request %s', str(err))
raise vapi_jsonrpc_error_invalid_request(data=str(err))
if request is None:
logger.error('Invalid request')
raise vapi_jsonrpc_error_invalid_request()
return request
[docs]def deserialize_response(response_str):
"""
Deserialize a json rpc response
:type response_str: :class:`str` or :class:`bytes`
:param response_str: json rpc response str
:rtype: :class:`JsonRpcResponse`
:return: json rpc response
"""
# NYI: Vapi object directly from json string? Not easy with standard json
# deserializer
try:
# In 2.x, we get str from underlying layer and in 3.x,
# we get bytes
if isinstance(response_str, six.binary_type):
response_str = response_str.decode('utf-8')
obj = json.loads(response_str, parse_float=decimal.Decimal)
except ValueError as err:
logger.error('Deserialize response error %s', str(err))
raise vapi_jsonrpc_error_parse_error(data=str(err))
try:
response = jsonrpc_response_factory(**to_strkey_dict(obj))
except JsonRpcError as err:
raise
except Exception as err:
logger.error('Invalid response %s', str(err))
raise vapi_jsonrpc_error_invalid_request(data=str(err))
if response is None:
logger.error('Invalid response')
raise vapi_jsonrpc_error_invalid_params()
return response