"""
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__)
[docs]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
[docs] 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
[docs] 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
[docs]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)
[docs] def get_identifier(self):
"""
Returns interface identifier
:rtype: :class:`InterfaceIdentifier`
:return: Interface identifier
"""
return self._iface_id
[docs] 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)
[docs] 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)
[docs] 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)
[docs] 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)
[docs]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)
[docs]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
[docs] 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)