"""
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__)
[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.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
[docs] 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
[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:`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
[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:`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
[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
@property
[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