"""
Type converter to/from vAPI runtime data model to Python native data model
"""
__author__ = 'VMware, Inc.'
__copyright__ = 'Copyright 2012-2014 VMware, Inc. All rights reserved. -- VMware Confidential'
import logging
import six
from vmware.vapi.exception import CoreException
from vmware.vapi.bindings.type import (
BindingTypeVisitor, OptionalType, MAP_KEY_FIELD,
MAP_VALUE_FIELD)
from vmware.vapi.bindings.enum import Enum
from vmware.vapi.bindings.error import (VapiError, UnresolvedError)
from vmware.vapi.bindings.struct import VapiStruct
from vmware.vapi.bindings.datetime_helper import DateTimeConverter
from vmware.vapi.bindings.uri_helper import URIValidator
from vmware.vapi.lib.converter import Converter
from vmware.vapi.data.value import DataValue, StructValue
from vmware.vapi.l10n.runtime import message_factory
logger = logging.getLogger(__name__)
[docs]class PythonToVapiVisitor(BindingTypeVisitor): # pylint: disable=W0223
"""
Visitor to convert from Python native value to vAPI DataValue
"""
def __init__(self, value):
"""
Initialize PythonToVapiVisitor
:type value: :class:`object`
:param value: Native python value
"""
self._in_value = value
self._out_value = None
self._dispatch_map = {
'VoidType': self.visit_void,
'IntegerType': self.visit_primitive,
'DoubleType': self.visit_primitive,
'StringType': self.visit_primitive,
'SecretType': self.visit_primitive,
'BooleanType': self.visit_primitive,
'BlobType': self.visit_primitive,
'OptionalType': self.visit_optional,
'ListType': self.visit_list,
'SetType': self.visit_set,
'MapType': self.visit_map,
'StructType': self.visit_struct,
'ErrorType': self.visit_error,
'ReferenceType': self.visit_reference,
'OpaqueType': self.visit_opaque,
'DynamicStructType': self.visit_dynamic_struct,
'AnyErrorType': self.visit_any_error,
'DateTimeType': self.visit_date_time,
'URIType': self.visit_uri,
'EnumType': self.visit_enum,
'IdType': self.visit_primitive,
}
BindingTypeVisitor.__init__(self)
[docs] def visit(self, typ):
class_name = typ.__class__.__name__
method = self._dispatch_map[class_name]
method(typ)
[docs] def get_out_value(self):
"""
Returns the vAPI DataValue converted from the Python native value
:rtype: :class:`vmware.vapi.data.value.DataValue`
:return: vAPI DataValue
"""
return self._out_value
[docs] def visit_primitive(self, typ):
"""
Visit a primitive type python value
:type typ: :class:`vmware.vapi.binding.type.BindingType`
:param typ: Binding type of the value
"""
self._out_value = typ.definition.new_value(self._in_value)
[docs] def visit_void(self, typ):
"""
Visit a void value (i.e. None)
:type typ: :class:`vmware.vapi.binding.type.VoidType`
:param typ: Binding type of the value
"""
if self._in_value is not None:
msg = message_factory.get_message(
'vapi.bindings.typeconverter.voiddef.expect.null',
type(self._in_value).__name__)
logger.debug(msg)
raise CoreException(msg)
self._out_value = typ.definition.new_value()
[docs] def visit_opaque(self, typ):
"""
Visit an opaque value. Don't do any conversion.
:type typ: :class:`vmware.vapi.binding.type.OpaqueType`
:param typ: Binding type of the value
"""
if not isinstance(self._in_value, DataValue):
msg = message_factory.get_message(
'vapi.bindings.typeconverter.unexpected.python.type',
DataValue.__name__, type(self._in_value).__name__)
logger.debug(msg)
raise CoreException(msg)
self._out_value = self._in_value
[docs] def visit_list(self, typ):
"""
Visit a list value
:type typ: :class:`vmware.vapi.binding.type.ListType`
:param typ: Binding type of the value
"""
in_value = self._in_value
out_value = typ.definition.new_value()
elt_typ = typ.element_type
for elt_value in in_value:
self._in_value = elt_value
self.visit(elt_typ)
out_value.add(self._out_value)
self._in_value = in_value
self._out_value = out_value
[docs] def visit_set(self, typ):
"""
Visit a python set
:type typ: :class:`vmware.vapi.binding.type.SetType`
:param typ: Binding type of the value
"""
if not isinstance(self._in_value, set) and not isinstance(self._in_value, frozenset):
msg = message_factory.get_message(
'vapi.bindings.typeconverter.invalid',
'set', type(self._in_value))
logger.debug(msg)
raise CoreException(msg)
self.visit_list(typ)
[docs] def visit_map(self, typ):
"""
Visit a python dict
:type typ: :class:`vmware.vapi.binding.type.MapType`
:param typ: Binding type of the value
"""
in_value = self._in_value
out_value = typ.definition.new_value()
struct_def = typ.definition.element_type
for k, v in six.iteritems(in_value):
struct_val = struct_def.new_value()
self._in_value = k
self.visit(typ.key_type)
struct_val.set_field(MAP_KEY_FIELD, self._out_value)
self._in_value = v
self.visit(typ.value_type)
struct_val.set_field(MAP_VALUE_FIELD, self._out_value)
out_value.add(struct_val)
self._in_value = in_value
self._out_value = out_value
def _visit_vapi_struct(self, typ):
"""
Visit an instance of VapiStruct class
:type typ: :class:`vmware.vapi.binding.type.StructType`
:param typ: Binding type of the value
:rtype: :class:`vmware.vapi.data.value.DataValue`
:return: vAPI Data value
"""
in_value = self._in_value
out_value = typ.definition.new_value()
field_names = typ.get_field_names()
for field in field_names:
try:
self._in_value = in_value.get_field(field)
# If self._in_value is None and field type is not
# optional, raise an error
field_type = typ.get_field(field)
if (not isinstance(field_type, OptionalType) and
self._in_value is None):
raise AttributeError
except AttributeError:
msg = message_factory.get_message(
'vapi.bindings.typeconverter.struct.missing.field',
field, typ.name)
logger.debug(msg)
raise CoreException(msg)
try:
self.visit(typ.get_field(field))
except Exception as e:
msg = message_factory.get_message(
'vapi.bindings.typeconverter.struct.invalid.field',
field, typ.name)
logger.debug(msg)
raise CoreException(msg, e)
out_value.set_field(field, self._out_value)
self._in_value = in_value
self._out_value = out_value
def _visit_python_dict(self, typ):
"""
Visit an instance of Python native dictionary
:type typ: :class:`vmware.vapi.binding.type.StructType`
:param typ: Binding type of the value
:rtype: :class:`vmware.vapi.data.value.DataValue`
:return: vAPI Data value
"""
in_value = self._in_value
out_value = typ.definition.new_value()
field_names = typ.get_field_names()
for field in field_names:
try:
self._in_value = in_value[field]
except KeyError:
msg = message_factory.get_message(
'vapi.bindings.typeconverter.dict.missing.key',
field)
logger.debug(msg)
raise CoreException(msg)
self.visit(typ.get_field(field))
out_value.set_field(field, self._out_value)
self._in_value = in_value
self._out_value = out_value
[docs] def visit_struct(self, typ):
"""
Visit a struct value
:type typ: :class:`vmware.vapi.binding.type.StructType`
:param typ: Binding type of the value
"""
if isinstance(self._in_value, VapiStruct):
self._visit_vapi_struct(typ)
# Validating the constraints in the struct value
# before sending it over the wire. The validation does
# not happen during initialization of VapiStruct or
# StructValue
self._in_value.validate_struct_value(self._out_value)
elif isinstance(self._in_value, dict):
self._visit_python_dict(typ)
else:
msg = message_factory.get_message(
'vapi.bindings.typeconverter.unexpected.struct.type',
type(self._in_value).__name__)
logger.debug(msg)
raise CoreException(msg)
[docs] def visit_error(self, typ):
"""
Visit an error value
:type typ: :class:`vmware.vapi.binding.type.ErrorType`
:param typ: Binding type of the value
"""
if isinstance(self._in_value, VapiError):
self._visit_vapi_struct(typ)
else:
msg = message_factory.get_message(
'vapi.bindings.typeconverter.unexpected.error.type',
type(self._in_value).__name__)
logger.debug(msg)
raise CoreException(msg)
[docs] def visit_dynamic_struct(self, typ):
"""
Visit any struct value
:type typ: :class:`vmware.vapi.binding.type.DynamicStructType`
:param typ: Binding type of the value
"""
if self._in_value.__class__.__name__ == 'VapiStruct':
self._out_value = self._in_value.get_struct_value()
elif self._in_value.__class__.__name__ == 'VapiError':
self._out_value = self._in_value.get_error_value()
elif isinstance(self._in_value, VapiStruct):
self._visit_vapi_struct(self._in_value.get_binding_type())
elif isinstance(self._in_value, StructValue):
self._out_value = self._in_value
else:
msg = message_factory.get_message(
'vapi.bindings.typeconverter.unexpected.python.type',
StructValue.__name__, type(self._in_value).__name__)
logger.debug(msg)
raise CoreException(msg)
[docs] def visit_any_error(self, typ):
"""
Visit any error value
:type typ: :class:`vmware.vapi.binding.type.AnyErrorType`
:param typ: Binding type of the value
"""
if isinstance(self._in_value, UnresolvedError):
self._out_value = self._in_value.get_error_value()
elif isinstance(self._in_value, VapiError):
self._visit_vapi_struct(self._in_value.get_binding_type())
else:
msg = message_factory.get_message(
'vapi.bindings.typeconverter.unexpected.python.type',
VapiError.__name__, type(self._in_value).__name__)
logger.debug(msg)
raise CoreException(msg)
[docs] def visit_optional(self, typ):
"""
Visit an optional value
:type typ: :class:`vmware.vapi.binding.type.OptionalType`
:param typ: Binding type of the value
"""
if self._in_value is None:
self._out_value = typ.definition.new_value()
else:
self.visit(typ.element_type)
self._out_value = typ.definition.new_value(self._out_value)
[docs] def visit_date_time(self, typ):
"""
Visit a datetime value
:type typ: :class:`vmware.vapi.binding.type.DateTimeType`
:param typ: Binding type of the value
"""
dt_str = DateTimeConverter.convert_from_datetime(self._in_value)
self._out_value = typ.definition.new_value(dt_str)
[docs] def visit_uri(self, typ):
"""
Visit an URI value
:type typ: :class:`vmware.vapi.binding.type.UriType`
:param typ: Binding type of the value
"""
URIValidator.validate(self._in_value)
self._out_value = typ.definition.new_value(self._in_value)
[docs] def visit_reference(self, typ):
"""
Visit a reference type
:type typ: :class:`vmware.vapi.binding.type.ReferenceType`
:param typ: Binding type of the value
"""
self.visit(typ.resolved_type)
[docs] def visit_enum(self, typ):
"""
Visit a enum type python value
:type typ: :class:`vmware.vapi.binding.type.EnumType`
:param typ: Binding type of the value
"""
if not isinstance(self._in_value, Enum):
# TODO: Verify if the value is instance of actual binding
# class
msg = message_factory.get_message(
'vapi.bindings.typeconverter.unexpected.python.type',
Enum, type(self._in_value))
logger.debug(msg)
raise CoreException(msg)
# Convert from vmware.vapi.bindings.Enum instance to string value
enum_string_value = str(self._in_value)
self._out_value = typ.definition.new_value(enum_string_value)
[docs]class VapiToPythonVisitor(BindingTypeVisitor): # pylint: disable=W0223
"""
Visitor to convert from vAPI DataValue to Python native value
"""
def __init__(self, value, resolver):
"""
Initialize VapiToPythonVisitor
:type value: :class:`vmware.vapi.data.value.DataValue`
:param value: vAPI DataValue to be converted
:type resolver: :class:`vmware.vapi.bindings.common.NameToTypeResolver` or
``None``
:param resolver: Type resolver
"""
self._in_value = value
self._resolver = resolver
self._out_value = None
self._dispatch_map = {
'VoidType': self.visit_void,
'IntegerType': self.visit_primitive,
'DoubleType': self.visit_primitive,
'StringType': self.visit_string,
'SecretType': self.visit_primitive,
'BooleanType': self.visit_primitive,
'BlobType': self.visit_primitive,
'OptionalType': self.visit_optional,
'ListType': self.visit_list,
'SetType': self.visit_set,
'MapType': self.visit_map,
'StructType': self.visit_struct,
'ErrorType': self.visit_error,
'ReferenceType': self.visit_reference,
'OpaqueType': self.visit_opaque,
'DynamicStructType': self.visit_dynamic_struct,
'AnyErrorType': self.visit_any_error,
'DateTimeType': self.visit_date_time,
'URIType': self.visit_uri,
'EnumType': self.visit_enum,
'IdType': self.visit_primitive,
}
BindingTypeVisitor.__init__(self)
[docs] def visit(self, typ):
class_name = typ.__class__.__name__
method = self._dispatch_map[class_name]
method(typ)
[docs] def get_out_value(self):
"""
Returns the Python native value converted from the vAPI DataValue
:rtype: :class:`object`
:return: Native python value
"""
return self._out_value
[docs] def visit_primitive(self, typ): # pylint: disable=W0613
"""
Visit one of the primitive DataValues
:type typ: :class:`vmware.vapi.binding.type.BindingType`
:param typ: Binding type of the value
"""
self._out_value = self._in_value.value
[docs] def visit_void(self, typ):
"""
Since there is no VoidValue, just return None
:type typ: :class:`vmware.vapi.binding.type.VoidType`
:param typ: Binding type of the value
"""
self._out_value = None
[docs] def visit_string(self, typ):
"""
Visit StringValue
:type typ: :class:`vmware.vapi.binding.type.StringType`
:param typ: Binding type of the value
"""
try:
self._out_value = str(self._in_value.value)
except UnicodeError:
self._out_value = self._in_value.value
[docs] def visit_opaque(self, typ):
"""
Since there is no OpaqueValue, don't do any conversion
:type typ: :class:`vmware.vapi.binding.type.OpaqueType`
:param typ: Binding type of the value
"""
self._out_value = self._in_value
def _visit_list_element(self, value, typ):
"""
Visit a ListValue element
:type value: :class:`vmware.vapi.data.value.DataValue`
:param value: element value
:type typ: :class:`vmware.vapi.binding.type.ListType`
:param typ: Binding type of the element
"""
self._in_value = value
self.visit(typ)
return self._out_value
[docs] def visit_list(self, typ):
"""
Visit a ListValue
:type typ: :class:`vmware.vapi.binding.type.ListType`
:param typ: Binding type of the value
"""
elt_typ = typ.element_type
in_value = self._in_value
self._out_value = [self._visit_list_element(elt_value, elt_typ)
for elt_value in in_value]
self._in_value = in_value
[docs] def visit_set(self, typ):
"""
Visit a List Value. This ListValue must represent a set i.e. there must not be any duplicate elements
:type typ: :class:`vmware.vapi.binding.type.SetType`
:param typ: Binding type of the value
"""
elt_typ = typ.element_type
in_value = self._in_value
out_value = set()
for elt_value in in_value:
elt = self._visit_list_element(elt_value, elt_typ)
if elt in out_value:
msg = message_factory.get_message(
'vapi.bindings.typeconverter.set.duplicate.element',
elt)
logger.debug(msg)
raise CoreException(msg)
out_value.add(elt)
self._out_value = out_value
self._in_value = in_value
[docs] def visit_map(self, typ):
"""
Visit a List Value. This ListValue must represent a map. Each element
of the ListValue is a StructValue with two fields, namely 'key' and 'value'.
The 'key' field represents the key of the map and the 'value' field
represents the value of the map. Also, since this represents a map, there
should not be duplicate keys.
:type typ: :class:`vmware.vapi.binding.type.MapType`
:param typ: Binding type of the value
"""
in_value = self._in_value
key_typ = typ.key_type
value_typ = typ.value_type
out_value = {}
for elt_value in in_value:
key = self._visit_struct_field(elt_value.get_field(MAP_KEY_FIELD),
key_typ)
if key in out_value:
msg = message_factory.get_message(
'vapi.bindings.typeconverter.map.duplicate.key',
key)
logger.debug(msg)
raise CoreException(msg)
value = self._visit_struct_field(elt_value.get_field(MAP_VALUE_FIELD),
value_typ)
out_value[key] = value
self._out_value = out_value
self._in_value = in_value
def _visit_struct_field(self, value, typ):
"""
Visit a field of a StructValue
:type value: :class:`vmware.vapi.data.value.DataValue`
:param value: field value
:type typ: :class:`vmware.vapi.binding.type.StructType`
:param typ: Binding type of the struct field
"""
self._in_value = value
self.visit(typ)
return self._out_value
[docs] def visit_struct(self, typ):
"""
Visit StructValue
:type typ: :class:`vmware.vapi.binding.type.StructType`
:param typ: Binding type of the value
"""
in_value = self._in_value
out_value = {}
typ_field_names = typ.get_field_names()
value_field_names = in_value.get_field_names()
for field_name in typ_field_names:
pep_name = Converter.canonical_to_pep(field_name)
field_type = typ.get_field(field_name)
# When the request is converted from DataValue to Native value
# on the server side:
# - if the client - server version matches: we wont hit this case
# - if new client talks to old server, runtime would have
# raised InvalidArgument error if there are unexpected optional
# fields
# - if old client talks to new server, runtime adds the missing
# optional fields, so we wont hit this case
#
# When the response is converted from DataValue to Native value
# on the client side:
# - if the client - server version matches: we wont hit this case
# - if new client talks to old server, client bindings should
# tolerate the absence of expected optional properties. So,
# we have to set it to None!
# - if old client talks to new server, we will visit all the known
# fields here and the unexpected fields are added to the VapiStruct
# object
if (isinstance(field_type, OptionalType) and
not in_value.has_field(field_name)):
out_value[pep_name] = None
else:
out_value[pep_name] = self._visit_struct_field(
in_value.get_field(field_name), field_type)
self._in_value = in_value
struct_class = typ.binding_class
if struct_class is not None:
struct_class.validate_struct_value(self._in_value)
self._out_value = struct_class(**out_value)
# Fields present in struct value but not in struct binding type.
unexpected_field_names = set(value_field_names) - set(typ_field_names)
if unexpected_field_names:
unexpected_fields = {}
for field_name in unexpected_field_names:
unexpected_fields[field_name] = in_value.get_field(field_name)
self._out_value._set_unexpected_fields(unexpected_fields) # pylint: disable=E1103,W0212
else:
self._out_value = out_value
[docs] def visit_dynamic_struct(self, typ):
"""
Visit StructValue to convert it into the base VapiStruct
:type typ: :class:`vmware.vapi.binding.type.DynamicStructType`
:param typ: Binding type of the value
"""
self._out_value = VapiStruct(struct_value=self._in_value)
[docs] def visit_any_error(self, typ):
"""
Visit ErrorValue to convert it into the base VapiError
:type typ: :class:`vmware.vapi.binding.type.AnyErrorType`
:param typ: Binding type of the value
"""
typ = self._resolver and self._resolver.resolve(self._in_value.name)
if typ is None:
self._out_value = UnresolvedError({}, self._in_value)
else:
typ.accept(self)
[docs] def visit_error(self, typ):
"""
Visit ErrorValue
:type typ: :class:`vmware.vapi.binding.type.ErrorType`
:param typ: Binding type of the value
"""
self.visit_struct(typ)
[docs] def visit_optional(self, typ):
"""
Visit OptionalValue
:type typ: :class:`vmware.vapi.binding.type.OptionalType`
:param typ: Binding type of the value
"""
if self._in_value.is_set():
self._in_value = self._in_value.value
self.visit(typ.element_type)
else:
self._out_value = None
[docs] def visit_date_time(self, typ):
"""
Visit a datetime value
:type typ: :class:`vmware.vapi.binding.type.DateTimeType`
:param typ: Binding type of the value
"""
self._out_value = DateTimeConverter.convert_to_datetime(
self._in_value.value)
[docs] def visit_uri(self, typ):
"""
Visit an URI value
:type typ: :class:`vmware.vapi.binding.type.UriType`
:param typ: Binding type of the value
"""
uri_string = self._in_value.value
URIValidator.validate(uri_string)
self._out_value = uri_string
[docs] def visit_enum(self, typ):
"""
Visit an Enum value
:type typ: :class:`vmware.vapi.binding.type.EnumType`
:param typ: Binding type of the value
"""
enum_string = self._in_value.value
if typ.binding_class:
try:
self._out_value = getattr(typ.binding_class, enum_string)
except AttributeError:
self._out_value = typ.binding_class(enum_string)
else:
self._out_value = Enum(enum_string)
[docs] def visit_reference(self, typ):
"""
Visit a reference type
:type typ: :class:`vmware.vapi.binding.type.ReferenceType`
:param typ: Binding type of the value
"""
self.visit(typ.resolved_type)
[docs]class TypeConverter(object):
"""
Converter class that converts values from vAPI data model to Python native
data model
"""
@staticmethod
[docs] def convert_to_python(vapi_val, binding_type, resolver=None):
"""
Converts vAPI DataValue to Python native value
:type vapi_val: :class:`vmware.vapi.data.value.DataValue`
:param vapi_val: vAPI DataValue to be converted
:type binding_type: :class:`vmware.vapi.bindings.type.BindingType`
:param binding_type: BindingType for the value
:type resolver: :class:`vmware.vapi.bindings.common.NameToTypeResolver` or
``None``
:param resolver: Type resolver
:rtype: :class:`object`
:return: Python native value
"""
visitor = VapiToPythonVisitor(vapi_val, resolver)
binding_type.accept(visitor)
return visitor.get_out_value()
@staticmethod
[docs] def convert_to_vapi(py_val, binding_type):
"""
Converts Python native value to vAPI DataValue
:type py_val: :class:`object`
:param py_val: Python native value to be converted
:type binding_type: :class:`vmware.vapi.bindings.type.BindingType`
:param binding_type: BindingType for the value
:rtype: :class:`vmware.vapi.data.value.DataValue`
:return: vAPI DataValue
"""
visitor = PythonToVapiVisitor(py_val)
binding_type.accept(visitor)
return visitor.get_out_value()