### Navigation

- index
- modules |
- vSphere Automation SDK for Python 6.5.0 documentation »
- Module code »

# Source code for vmware.vapi.bindings.skeleton

    """
    Skeleton helper classes
    """
    __author__ = 'VMware, Inc.'
    __copyright__ = 'Copyright (c) 2015 VMware, Inc.  All rights reserved.'

    import sys
    import threading
    import logging
    import six

    from vmware.vapi.bindings.error import VapiError
    from vmware.vapi.bindings.converter import TypeConverter
    from vmware.vapi.bindings.common import NameToTypeResolver
    from vmware.vapi.bindings.type import (
        ListType, OptionalType, ReferenceType, StructType, DynamicStructType)
    from vmware.vapi.core import ApiInterface, MethodIdentifier
    from vmware.vapi.core import InterfaceDefinition, MethodResult, MethodDefinition
    from vmware.vapi.core import InterfaceIdentifier
    from vmware.vapi.data.value import OptionalValue
    from vmware.vapi.exception import CoreException
    from vmware.vapi.lib.converter import Converter
    from vmware.vapi.lib.std import make_error_value_from_msgs, make_std_error_def
    from vmware.vapi.l10n.runtime import message_factory

    # TLS object to store the execution context of a request
    TLS = threading.local()
    TLS.ctx = None

    logger = logging.getLogger(__name__)

    class ApiInterfaceSkeleton(ApiInterface):
        """
        Skeleton class for :class:`ApiInterface`. This class implements the
        :class:`ApiInterface` interface.
        """

        # XXX These error definitions should be eliminated when we figure out
        #     where/how to get the error definitions used by the local provider
        _internal_server_error_def = make_std_error_def(
            'com.vmware.vapi.std.errors.internal_server_error')
        _invalid_argument_def = make_std_error_def(
            'com.vmware.vapi.std.errors.invalid_argument')
        _unexpected_input_def = make_std_error_def(
            'com.vmware.vapi.std.errors.unexpected_input')

        def __init__(self, iface_name, impl, operations, error_types):
            """
            Initialize the ApiInterface skeleton class

            :type  iface_name: :class:`str`
            :param iface_name: Interface name
            :type  impl: :class:`VapiInterface`
            :param impl: Class that implements this interface
            :type  operations: :class:`dict`
            :param operations: Description of the operations in this service
            :type  error_types: :class:`list` of :class:`vmware.vapi.bindings.type.ErrorType`
            :param error_types: error types to be registered in this configuration
            """
            self._operations = operations
            self._iface_id = InterfaceIdentifier(iface_name)
            operation_ids = [MethodIdentifier(self._iface_id, operation_name)
                             for operation_name in six.iterkeys(self._operations)]
            self._iface_def = InterfaceDefinition(self._iface_id,
                                                  operation_ids)
            self._impl = impl
            error_types = error_types or []
            self._resolver = NameToTypeResolver(
                dict([(e.definition.name, e) for e in error_types]))
            ApiInterface.__init__(self)

        @staticmethod
        def _pepify_args(meth_args):
            """
            Converts all the keys of given keyword arguments into PEP8 standard
            names

            :type  meth_args: :class:`dict`
            :param meth_args: The keyword arguments to be converted
            :rtype:  :class:`dict`
            :return: The converted keyword arguments
            """
            new_args = {}
            for k, v in six.iteritems(meth_args):
                new_args[Converter.canonical_to_pep(k)] = v
            return new_args

        @classmethod
        def is_set_optional_field(cls, value):
            """
            Returns true if the value is OptionalValue and it is set

            :type  value: :class:`vmware.vapi.data.value.DataValue`
            :param value: value to be checked
            :rtype: :class:`bool`
            :return: True if the value is OptionalValue and is set, False otherwise
            """
            return not isinstance(value, OptionalValue) or value.is_set()

        @classmethod
        def check_for_unknown_fields(cls, value_type, value):
            """
            Check if the StructValues inside the input DataValue has any
            unknown fields

            :type  value_type: :class:`vmware.vapi.bindings.type.BindingType`
            :param value_type: Binding Type
            :type  value: :class:`vmware.vapi.data.value.DataValue`
            :param value: DataValue
            :rtype: :class:`vmware.vapi.data.value.ErrorValue` or ``None``
            :return: ErrorValue describing the unknown fields or None if no
                     unknown fields were found
            """
            if isinstance(value_type, ReferenceType):
                value_type = value_type.resolved_type

            if isinstance(value_type, DynamicStructType):
                pass
            elif isinstance(value_type, StructType):
                expected_field_names = value_type.get_field_names()
                unexpected_fields = set([
                    field for field in value.get_field_names()
                    if field not in expected_field_names and
                    cls.is_set_optional_field(value.get_field(field))
                ])
                if unexpected_fields:
                    msg = message_factory.get_message(
                        'vapi.data.structure.field.unexpected',
                        repr(unexpected_fields), value.name)
                    logger.debug(msg)
                    return make_error_value_from_msgs(cls._unexpected_input_def,
                                                      msg)
                for field in expected_field_names:
                    field_value = value.get_field(field)
                    field_type = value_type.get_field(field)
                    error = ApiInterfaceSkeleton.check_for_unknown_fields(
                        field_type, field_value)
                    if error:
                        return error
            elif isinstance(value_type, ListType):
                element_type = value_type.element_type
                if (isinstance(element_type, ListType) or
                        isinstance(element_type, OptionalType) or
                        isinstance(element_type, StructType) or
                        isinstance(element_type, ReferenceType)):
                    for element in value:
                        error = ApiInterfaceSkeleton.check_for_unknown_fields(
                            element_type, element)
                        if error:
                            return error
            elif isinstance(value_type, OptionalType):
                if value.is_set():
                    return ApiInterfaceSkeleton.check_for_unknown_fields(
                        value_type.element_type, value.value)
            return None

        def get_identifier(self):
            return self._iface_id

        def get_definition(self):
            return self._iface_def

        def get_method_definition(self, method_id):
            opInfo = self._operations.get(method_id.get_name())
            errors_defs = [e.definition for e in opInfo['errors']]
            if self._internal_server_error_def not in errors_defs:
                errors_defs.append(self._internal_server_error_def)
            if self._invalid_argument_def not in errors_defs:
                errors_defs.append(self._invalid_argument_def)
            if self._unexpected_input_def not in errors_defs:
                errors_defs.append(self._unexpected_input_def)
            return MethodDefinition(method_id,
                                    opInfo['input_type'].definition,
                                    opInfo['output_type'].definition,
                                    errors_defs)

        def invoke(self, ctx, method_id, input_value):
            """
            Invokes the specified method using the execution context and
            the input provided

            :type  ctx: :class:`vmware.vapi.core.ExecutionContext`
            :param ctx: Execution context for this method
            :type  input_value: :class:`vmware.vapi.data.value.StructValue`
            :param input_value: Method input parameters

            :rtype: :class:`vmware.vapi.core.MethodResult`
            :return: Result of the method invocation
            """
            opInfo = self._operations.get(method_id.get_name())

            # Set execution context in the TLS object of the impl
            TLS.ctx = ctx

            try:
                # input validators
                validators = opInfo['input_validator_list']
                input_type = opInfo['input_type']
                for validator in validators:
                    msg_list = validator.validate(input_value, input_type)
                    if msg_list:
                        error_value = make_error_value_from_msgs(
                            self._invalid_argument_def, *msg_list)
                        return MethodResult(error=error_value)
                error_value = ApiInterfaceSkeleton.check_for_unknown_fields(
                    input_type, input_value)
                if error_value:
                    return MethodResult(error=error_value)

                try:
                    meth_args = TypeConverter.convert_to_python(
                        input_value, binding_type=input_type,
                        resolver=self._resolver)
                except CoreException as e:
                    error_value = make_error_value_from_msgs(
                        self._invalid_argument_def, *e.messages)
                    return MethodResult(error=error_value)

                meth_args = ApiInterfaceSkeleton._pepify_args(meth_args)
                method_name = Converter.canonical_to_pep(method_id.get_name())
                method = getattr(self._impl, method_name)
                meth_output = method(**meth_args)

                output_type = opInfo['output_type']
                output = TypeConverter.convert_to_vapi(meth_output,
                                                       output_type)
                # output validators
                validators = opInfo['output_validator_list']
                for validator in validators:
                    msg_list = validator.validate(output, output_type)
                    if msg_list:
                        error_value = make_error_value_from_msgs(
                            self._internal_server_error_def, *msg_list)
                        return MethodResult(error=error_value)
                result = MethodResult(output=output)
            except CoreException as e:
                logger.exception("Error in invoking %s - %s",
                                 str(method_id), e)
                error_value = make_error_value_from_msgs(
                    self._internal_server_error_def,
                    *e.messages)
                result = MethodResult(error=error_value)
            except VapiError as e:
                exc_info = sys.exc_info()
                error_type = e.__class__.get_binding_type()
                try:
                    error = TypeConverter.convert_to_vapi(e, error_type)
                except CoreException as ce:
                    logger.error('Failed to convert %s to error type %s because %s',
                                 e, error_type.name,
                                 ' because '.join((e.def_msg for e in ce.messages)),
                                 exc_info=exc_info)
                    raise
                result = MethodResult(error=error)
            finally:
                # Reset the execution context after the operation
                TLS.ctx = None

            return result

    class VapiInterface(object):
        """
        vAPI Interface class is used by the python server side bindings. This
        encapsulates the :class:`vmware.vapi.bindings.skeleton.ApiInterfaceSkeleton` class.
        """
        def __init__(self, api_interface, error_types=None):
            """
            Initialize the VapiInterface class

            :type  api_interface: :class:`vmware.vapi.bindings.skeleton.ApiInterfaceSkeleton`
            :param api_interface: Api Interface skeleton class for this interface
            :type  error_types: :class:`list` of :class:`vmware.vapi.bindings.type.ErrorType`
            :param error_types: error types to be registered in this configuration
            """
            self._api_interface = api_interface(self, error_types=error_types)

        @property
        def api_interface(self):
            """
            Returns the ApiInterfaceSkeleton instance. Local Provider uses this
            method to discover the ApiInterface so that it can route method calls
            for the methods implemented by this interface.

            :rtype:  :class:`vmware.vapi.bindings.skeleton.ApiInterfaceSkeleton`
            :return: Api Interface skeleton class for this interface
            """
            return self._api_interface

        @property
        def execution_context(self):
            """
            Returns the execution context of a method invocation

            :rtype: :class:`vmware.vapi.core.ExecutionContext`
            :return: Execution context of a method invocation
            """
            return TLS.ctx

    class VapiFilter(logging.Filter):
        """
        This is a filter that injects request context into the log
        """
        def filter(self, record):
            if TLS.ctx is not None:
                record.op_id = TLS.ctx.application_context.get('opId')
            else:
                record.op_id = None
            return True

### Navigation

- index
- modules |
- vSphere Automation SDK for Python 6.5.0 documentation »
- Module code »