Source code for vmware.vapi.bindings.converter

"""
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()