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 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__)

[docs]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) 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
[docs] def is_set_optional_field(cls, value): """ Returns true if the value is OptionalValue and it is set :type value: :class:`` :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()
[docs] 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:`` :param value: DataValue :rtype: :class:`` 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( '', repr(unexpected_fields), 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
[docs] def get_identifier(self): return self._iface_id
[docs] def get_definition(self): return self._iface_def
[docs] 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)
[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 input_value: :class:`` :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,, ' 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
[docs]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
[docs] 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
[docs] 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
[docs]class VapiFilter(logging.Filter): """ This is a filter that injects request context into the log """
[docs] 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