### Navigation

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

# Source code for vmware.vapi.bindings.stub

    """
    Stub helper classes
    """

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

    import logging
    import six

    from vmware.vapi.core import MethodIdentifier, InterfaceDefinition, MethodDefinition
    from vmware.vapi.core import InterfaceIdentifier, ApiInterface
    from vmware.vapi.bindings.converter import TypeConverter
    from vmware.vapi.bindings.error import UnresolvedError
    from vmware.vapi.bindings.common import (
        raise_core_exception, NameToTypeResolver)
    from vmware.vapi.lib.converter import Converter

    logger = logging.getLogger(__name__)

    class StubConfiguration(object):
        """
        Configuration data for vAPI stub classes

        :type connector: :class:`vmware.vapi.protocol.client.connector.Connector`
        :ivar connector: Connection to be used to talk to the remote ApiProvider
        """
        def __init__(self, connector, *error_types):
            """
            Initialize the stub configuration

            :type  connector: :class:`vmware.vapi.protocol.client.connector.Connector`
            :param connector: Connection to be used to talk to the remote
                              ApiProvider
            :type  error_types: :class:`list` of :class:`vmware.vapi.bindings.type.ErrorType`
            :param error_types: error types to be registered in this configuration
            """
            if connector is None:
                raise TypeError('Input parameter connector is None')
            self._connector = connector
            self._resolver = NameToTypeResolver(
                dict([(e.definition.name, e) for e in error_types]))

        @property
        def connector(self):
            """
            :rtype: :class:`vmware.vapi.protocol.client.connector.Connector`
            :return: Connection to be used to talk to the remote ApiProvider
            """
            return self._connector

        @property
        def resolver(self):
            """
            Type resolver that can resolve canonical names to its binding types

            :rtype: :class:`vmware.vapi.bindings.common.NameToTypeResolver`
            :return: Type resolver
            """
            return self._resolver

    # We don't need all the methods in ApiMethod in the stub.
    # So disabling method not implemented pylint error
    class ApiInterfaceStub(ApiInterface):  # pylint: disable=W0223
        """
        Stub class for Api Interface
        """
        def __init__(self, iface_name, config, operations):
            """
            Initialize the ApiMethod skeleton object

            :type  iface_name: :class:`str`
            :param iface_name: Interface name
            :type  config: :class:`StubConfiguration`
            :param config: Configuration data for vAPI stubs
            :type  operations: :class:`dict`
            :param operations: Dictionary of operation name to operation information
            """
            self._iface_id = InterfaceIdentifier(iface_name)
            self._config = config
            self._operations = operations
            self._api_provider = config.connector.get_api_provider()
            ApiInterface.__init__(self)

        def get_identifier(self):
            """
            Returns interface identifier

            :rtype: :class:`InterfaceIdentifier`
            :return: Interface identifier
            """
            return self._iface_id

        def get_definition(self):
            """
            Returns interface definition

            :rtype: :class:`InterfaceDefinition`
            :return: Interface definition
            """
            operations = [MethodIdentifier(self._iface_id, operation_name)
                          for operation_name in six.iterkeys(self._operations)]
            return InterfaceDefinition(self._iface_id, operations)

        def get_method_definition(self, method_id):
            opInfo = self._operations.get(method_id.get_name())
            errors_defs = [e.definition for e in six.itervalues(opInfo.get('errors'))]
            return MethodDefinition(method_id,
                                    opInfo.get('input_type').definition,
                                    opInfo.get('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  method_id: :class:`vmware.vapi.core.MethodIdentifier`
            :param method_id: Method identifier
            :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
            """
            return self._api_provider.invoke(self._iface_id.get_name(),
                                             method_id.get_name(),
                                             input_value,
                                             ctx)

        def native_invoke(self, ctx, method_name, kwargs):
            """
            Invokes the method corresponding to the given method name
            with the kwargs.

            In this method, python native values are converted to vAPI runtime values,
            operation is invoked and the result are converted back to python native
            values

            :type  method_name: :class:`str`
            :param method_name: Method name
            :type  kwargs: :class:`dict`
            :param kwargs: arguments to be passed to the method
            :rtype: :class:`object`
            :return: Method result
            """
            opInfo = self._operations.get(method_name)
            if opInfo is None:
                raise Exception('Could not find %s method in %s interface' %
                                (method_name, str(self._iface_id)))

            # Convert input
            input_type = opInfo['input_type']
            data_val = TypeConverter.convert_to_vapi(kwargs, input_type)

            # Validate input
            validators = opInfo['input_validator_list']
            for validator in validators:
                msg_list = validator.validate(data_val, input_type)
                raise_core_exception(msg_list)

            # Invoke
            method_id = MethodIdentifier(self._iface_id, method_name)
            method_result = self.invoke(ctx, method_id, data_val)

            # Validate output
            if method_result.success():
                validators = opInfo['output_validator_list']
                output_type = opInfo['output_type']
                for validator in validators:
                    msg_list = validator.validate(method_result.output, output_type)
                    raise_core_exception(msg_list)

                # Convert output
                return TypeConverter.convert_to_python(method_result.output,
                                                       output_type,
                                                       self._config.resolver)
            else:
                # Convert error
                errors = opInfo['errors']
                error_type = errors.get(method_result.error.name)
                if error_type is None:
                    error_type = self._config.resolver.resolve(method_result.error.name)
                if error_type is None:
                    logger.warning('Unable to convert unexpected vAPI error %s ' +
                                   'to native Python exception',
                                   method_result.error.name)
                    vapi_error = UnresolvedError({}, method_result)
                    raise vapi_error
                raise TypeConverter.convert_to_python(method_result.error,  # pylint: disable=E0702
                                                      error_type,
                                                      self._config.resolver)

    class VapiInterface(object):
        """
        vAPI Interface class is used by the python client side bindings. This
        encapsulates the ApiInterfaceStub instance
        """
        def __init__(self, config, api_interface):
            """
            Initialize VapiInterface object

            :type  config: :class:`StubConfiguration`
            :param config: Configuration data for vAPI stubs
            :type  api_interface: :class:`ApiInterfaceStub`
            :param api_interface: Instance of ApiInterfaceStub class that can
                                  execute the ApiMethods
            """
            if config is None:
                raise TypeError('Input parameter config is None')
            if isinstance(config, StubConfiguration):
                self._config = config
            else:
                raise TypeError('Input parameter config is not a StubConfiguration')
            self._api_interface = api_interface(self._config)

        def _invoke(self, _method_name, kwargs):
            """
            Invokes the ApiMethod corresponding to the given method name
            with the kwargs

            :type  _method_name: :class:`str`
            :param _method_name: Method name
            :type  kwargs: :class:`dict` of :class:`str` and :class:`object` or :class:`None`
            :param kwargs: arguments to be passed to the method
            :return: Method result
            """
            kwargs = kwargs or {}
            # Argument name is _method_name to make sure it doesn't collide
            # with actual parameter names of the method
            ctx = self._config.connector.new_context()
            return self._api_interface.native_invoke(ctx, _method_name, kwargs)

    class StubFactory(object):
        """
        Factory for client-side vAPI stubs
        """
        def __init__(self, config):
            """
            Initialize the stub factory

            :type  config: :class:`StubConfiguration`
            :param config: Configuration data for vAPI stubs
            """
            if config is None:
                raise TypeError('Input parameter config is None')
            if not isinstance(config, StubConfiguration):
                raise TypeError('Input parameter config is not a StubConfiguration')
            self._config = config

        def create_stub(self, service_name):
            """
            Create a stub corresponding to the specified service name

            :type  service_name: :class:`str`
            :param service_name: Name of the service

            :rtype: :class:`VapiInterface`
            :return: The stub correspoding to the specified service name
            """
            path_split = service_name.split('.')
            module_name = '%s_client' % '.'.join(path_split[:-1])
            class_name = Converter.underscore_to_capwords(path_split[-1])
            module = __import__(module_name, globals(), locals(), class_name)
            cls = getattr(module, class_name)
            return cls(self._config)

### Navigation

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