Homec4science
Diffusion invenio-infoscience (iugw2017) Browse

R3600/modules/miscutil/lib/pluginutils.py
1fccd1176372iugw2017

Last Change
Raw File

pluginutils.py
View Options

# -*- coding: utf-8 -*-
##
## This file is part of Invenio.
## Copyright (C) 2009, 2010, 2011 CERN.
##
## Invenio is free software; you can redistribute it and/or
## modify it under the terms of the GNU General Public License as
## published by the Free Software Foundation; either version 2 of the
## License, or (at your option) any later version.
##
## Invenio is distributed in the hope that it will be useful, but
## WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
## General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with Invenio; if not, write to the Free Software Foundation, Inc.,
## 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
"""
This module implement a generic plugin container facility.
"""
import sys
import os
import glob
import inspect
import imp
from invenio.config import CFG_PYLIBDIR
from invenio.textutils import wrap_text_in_a_box
class InvenioPluginContainerError(Exception):
"""
Exception raised when some error happens during plugin loading.
"""
pass
class PluginContainer(object):
"""
This class implements a I{plugin container}.
This class implements part of the dict interface with the condition
that only correctly enabled plugins can be retrieved by their plugin_name.
>>> ## Loading all the plugin within a directory.
>>> websubmit_functions = PluginContainer(
... os.path.join(CFG_PYLIBDIR,
... 'invenio', 'websubmit_functions', '*.py')
... )
>>> ## Loading an explicit plugin.
>>> case_eds = websubmit_functions['CaseEDS']
@param plugin_pathnames: zero or more plugins_pathnames from where to load
the plugins.
@type plugin_pathnames: string/list
@param plugin_builder: a callable with the signature
C{plugin_builder(plugin_name, plugin_code)} that will be called
to extract the actual plugin from the module stored in plugin_code.
@type plugin_builder: callable
@param api_version: the API version of the plugin. If specified, plugins
which specify different versions will fail to be loaded. Default value
is C{None} which turns off the version checking.
@type api_version: integer
@param plugin_signature: a stub to be used in order to check if a loaded
plugin respect a particular signature or not.
@type plugin_signature: class/function
@ivar _plugin_map: a map between plugin_name and a dict with keys
"error", "plugin", "plugin_path", "enabled", "api_version"
@type _plugin_map: dict
@ivar _plugin_pathnames: the list of normalized plugin pathnames
corresponding to the plugins to be loaded.
@type _plugin_pathnames: list
@ivar _plugin_builder: the plugin builder as passed to the constructor.
@type plugin_builder: function
@ivar api_version: the version as provided to the constructor.
@type api_version: integer
@group Mapping interface: __contains__,__getitem__,get,has_key,items,
iteritems,iterkeys,itervalues,keys,values,__len__
@group Main API: __init__,add_plugin_pathnames,get_enabled_plugins,
get_broken_plugins,get_plugin,reload_plugins
"""
def __init__(self,
plugin_pathnames=None,
plugin_builder=None,
api_version=None,
plugin_signature=None,
external=False):
self._plugin_map = {}
self._plugin_pathnames = []
self._external = external
self.api_version = api_version
if plugin_builder is None:
self._plugin_builder = self.default_plugin_builder
else:
self._plugin_builder = plugin_builder
self._plugin_signature = plugin_signature
if plugin_pathnames:
self.add_plugin_pathnames(plugin_pathnames)
def default_plugin_builder(plugin_name, plugin_code):
"""
Default plugin builder used to extract the plugin from the module
that contains it.
@note: By default it will look for a class or function with the same
name of the plugin.
@param plugin_name: the name of the plugin.
@type plugin_name: string
@param plugin_code: the code of the module as just read from
filesystem.
@type plugin_code: module
@return: the plugin
"""
return getattr(plugin_code, plugin_name)
default_plugin_builder = staticmethod(default_plugin_builder)
def add_plugin_pathnames(self, plugin_pathnames):
"""
Add a one or more plugin pathnames, i.e. full plugin path exploiting
wildcards, e.g. "bibformat_elements/bfe_*.py".
@note: these plugins_pathnames will be added to the current list of
plugin_pathnames, and all the plugins will be reloaded.
@param plugin_pathnames: one or more plugins_pathnames
@type plugin_pathnames: string/list
"""
if type(plugin_pathnames) is str:
self._plugin_pathnames.append(plugin_pathnames)
else:
self._plugin_pathnames.extend(plugin_pathnames)
self.reload_plugins()
def enable_plugin(self, plugin_name):
"""
Enable plugin_name.
@param plugin_name: the plugin name.
@type plugin_name: string
@raise KeyError: if the plugin does not exists.
"""
self._plugin_map[plugin_name]['enabled'] = True
def disable_plugin(self, plugin_name):
"""
Disable plugin_name.
@param plugin_name: the plugin name.
@type plugin_name: string
@raise KeyError: if the plugin does not exists.
"""
self._plugin_map[plugin_name]['enabled'] = False
def plugin_enabled_p(self, plugin_name):
"""
Returns True if the plugin is correctly enabled.
@param plugin_name: the plugin name.
@type plugin_name: string
@return: True if the plugin is correctly enabled..
@rtype: bool
@raise KeyError: if the plugin does not exists.
"""
return self._plugin_map[plugin_name]['enabled']
def get_plugin_filesystem_path(self, plugin_name):
"""
Returns the filesystem path from where the plugin was loaded.
@param plugin_name: the plugin name.
@type plugin_name: string
@return: the filesystem path.
@rtype: string
@raise KeyError: if the plugin does not exists.
"""
return self._plugin_map[plugin_name]['plugin_path']
def get_plugin(self, plugin_name):
"""
Returns the plugin corresponding to plugin_name.
@param plugin_name: the plugin name,
@type plugin_name: string
@return: the plugin
@raise KeyError: if the plugin does not exists or is not enabled.
"""
if self._plugin_map[plugin_name]['enabled']:
return self._plugin_map[plugin_name]['plugin']
else:
raise KeyError('"%s" is not enabled' % plugin_name)
def get_broken_plugins(self):
"""
Returns a map between plugin names and errors, in the form of
C{sys.exc_info} structure.
@return: plugin_name -> sys.exc_info().
@rtype: dict
"""
ret = {}
for plugin_name, plugin in self._plugin_map.iteritems():
if plugin['error']:
ret[plugin_name] = plugin['error']
return ret
def reload_plugins(self, reload=False):
"""
For the plugins found through iterating in the plugin_pathnames, loads
and working plugin.
@note: if a plugin has the same plugin_name of an already loaded
plugin, the former will override the latter (provided that the
former had a compatible signature to the latter).
@note: any plugin that fails to load will be added to the plugin
map as disabled and the sys.exc_info() captured during the
Exception will be stored. (if the failed plugin was supposed to
override an existing one, the latter will be overridden by
the failed former).
"""
# The reload keyword argument exists for backwards compatibility.
# Previously, reload_plugins, would not reload a module due to a bug.
for plugin_path in self._plugin_pathnames_iterator():
self._load_plugin(plugin_path, reload=reload)
def normalize_plugin_path(self, plugin_path):
"""
Returns a normalized plugin_path.
@param plugin_path: the plugin path.
@type plugin_path: string
@return: the normalized plugin path.
@rtype: string
@raise ValueError: if the path is not under CFG_PYLIBDIR/invenio
"""
invenio_path = os.path.abspath(os.path.join(CFG_PYLIBDIR, 'invenio'))
plugin_path = os.path.abspath(plugin_path)
if not self._external and not os.path.abspath(plugin_path).startswith(invenio_path):
raise ValueError('A plugin should be stored under "%s" ("%s" was'
' specified)' % (invenio_path, plugin_path))
return plugin_path
def _plugin_pathnames_iterator(self):
"""
Returns an iterator over all the normalized plugin path.
@note: older plugin_pathnames are considered first, and newer
plugin_pathnames later, so that plugin overriding is possible.
@return: the iterator over plugin paths.
@rtype: iterator
"""
for plugin_pathname in self._plugin_pathnames:
for plugin_path in glob.glob(plugin_pathname):
yield self.normalize_plugin_path(plugin_path)
def get_plugin_name(plugin_path):
"""
Returns the name of the plugin after the plugin_path.
@param plugin_path: the filesystem path to the plugin code.
@type plugin_path: string
@return: the plugin name.
@rtype: string
"""
plugin_name = os.path.basename(plugin_path)
if plugin_name.endswith('.py'):
plugin_name = plugin_name[:-len('.py')]
return plugin_name
get_plugin_name = staticmethod(get_plugin_name)
def _load_plugin(self, plugin_path, reload=False):
"""
Load a plugin in the plugin map.
@note: if the plugin_name calculated from plugin_path corresponds to
an already existing plugin, the old plugin will be overridden and
if the old plugin was correctly loaded but disabled also the
new plugin will be disabled.
@param plugin_path: the plugin path.
@type plugin_path: string
"""
api_version = None
try:
plugin_name = self.get_plugin_name(plugin_path)
# Let's see if the module is already loaded
plugin = None
if plugin_name in sys.modules:
mod = sys.modules[plugin_name]
if os.path.splitext(mod.__file__)[0] == os.path.splitext(plugin_path)[0]:
plugin = mod
if not plugin or reload:
# Let's load the plugin module.
plugin_fp, plugin_path, plugin_desc = imp.find_module(
plugin_name, [os.path.dirname(plugin_path)]
)
try:
plugin = imp.load_module(
plugin_name, plugin_fp, plugin_path, plugin_desc
)
finally:
if plugin_fp:
plugin_fp.close()
## Let's check for API version.
api_version = getattr(plugin, '__plugin_version__', None)
if self.api_version and api_version != self.api_version:
raise InvenioPluginContainerError("Plugin version mismatch."
" Expected %s, found %s" % (self.api_version, api_version))
## Let's load the actual plugin
plugin = self._plugin_builder(plugin_name, plugin)
## Are we overriding an already loaded plugin?
enabled = True
if plugin_name in self._plugin_map:
old_plugin = self._plugin_map[plugin_name]
if old_plugin['error'] is None:
enabled = old_plugin['enabled']
check_signature(plugin_name, old_plugin['plugin'], plugin)
## Let's check the plugin signature.
if self._plugin_signature:
check_signature(plugin_name, self._plugin_signature, plugin)
self._plugin_map[plugin_name] = {
'plugin': plugin,
'error': None,
'plugin_path': plugin_path,
'enabled': enabled,
'api_version': api_version,
}
except Exception:
self._plugin_map[plugin_name] = {
'plugin': None,
'error': sys.exc_info(),
'plugin_path': plugin_path,
'enabled': False,
'api_version': api_version,
}
def __getitem__(self, plugin_name):
"""
As in C{dict.__getitem__} but apply plugin name normalization and check
if the plugin is correctly enabled.
@param plugin_name: the name of the plugin
@type plugin_name: string
@return: the plugin.
@raise KeyError: if the corresponding plugin is not enabled or there
were some errors.
"""
plugin_name = self.get_plugin_name(plugin_name)
if plugin_name in self._plugin_map and \
self._plugin_map[plugin_name]['enabled'] is True:
return self._plugin_map[plugin_name]['plugin']
else:
raise KeyError('"%s" does not exists or is not correctly enabled' %
plugin_name)
def __contains__(self, plugin_name):
"""
As in C{dict.__contains__} but apply plugin name normalization and
check if the plugin is correctly enabled.
@param plugin_name: the name of the plugin
@type plugin_name: string
@return: True if plugin_name is correctly there.
@rtype: bool
"""
plugin_name = self.get_plugin_name(plugin_name)
return plugin_name in self._plugin_map and \
self._plugin_map[plugin_name]['enabled'] is True
def __len__(self):
"""
As in C{dict.__len__} but consider only correctly enabled plugins.
@return: the total number of plugins correctly enabled.
@rtype: integer
"""
count = 0
for plugin in self._plugin_map.values():
if plugin['enabled']:
count += 1
return count
def get(self, plugin_name, default=None):
"""
As in C{dict.get} but consider only correctly enabled plugins.
@param plugin_name: the name of the plugin
@type plugin_name: string
@param default: the default value to return if plugin_name does not
correspond to a correctly enabled plugin.
@return: the total number of plugins correctly enabled.
@rtype: integer
"""
try:
return self.__getitem__(plugin_name)
except KeyError:
return default
def has_key(self, plugin_name):
"""
As in C{dict.has_key} but apply plugin name normalization and check
if the plugin is correctly enabled.
@param plugin_name: the name of the plugin
@type plugin_name: string
@return: True if plugin_name is correctly there.
@rtype: bool
"""
return self.__contains__(plugin_name)
def items(self):
"""
As in C{dict.items} but checks if the plugin are correctly enabled.
@return: list of (plugin_name, plugin).
@rtype: [(plugin_name, plugin), ...]
"""
ret = []
for plugin_name, plugin in self._plugin_map.iteritems():
if plugin['enabled']:
ret.append((plugin_name, plugin['plugin']))
return ret
def iteritems(self):
"""
As in C{dict.iteritems} but checks if the plugin are correctly enabled.
@return: an iterator over the (plugin_name, plugin) items.
"""
for plugin_name, plugin in self._plugin_map.iteritems():
if plugin['enabled']:
yield (plugin_name, plugin['plugin'])
def iterkeys(self):
"""
As in C{dict.iterkeys} but checks if the plugin are correctly enabled.
@return: an iterator over the plugin_names.
"""
for plugin_name, plugin in self._plugin_map.iteritems():
if plugin['enabled']:
yield plugin_name
__iter__ = iterkeys
def itervalues(self):
"""
As in C{dict.itervalues} but checks if the plugin are correctly
enabled.
@return: an iterator over the plugins.
"""
for plugin in self._plugin_map.itervalues():
if plugin['enabled']:
yield plugin['plugin']
def keys(self):
"""
As in C{dict.keys} but checks if the plugin are correctly enabled.
@return: the list of enabled plugin_names.
@rtype: list of strings
"""
ret = []
for plugin_name, plugin in self._plugin_map.iteritems():
if plugin['enabled']:
ret.append(plugin_name)
return ret
def values(self):
"""
As in C{dict.values} but checks if the plugin are correctly enabled.
@return: the list of enabled plugin codes.
"""
return [plugin['plugin'] \
for plugin in self._plugin_map.values() if plugin['enabled']]
def get_enabled_plugins(self):
"""
Return a map of the correctly enabled plugins.
@return: a map plugin_name -> plugin
@rtype: dict
"""
ret = {}
for plugin_name, plugin in self._plugin_map.iteritems():
if plugin['enabled']:
ret[plugin_name] = plugin['plugin']
return ret
def check_signature(object_name, reference_object, other_object):
"""
Given a reference class or function check if an other class or function
could be substituted without causing any instantiation/usage issues.
@param object_name: the name of the object being checked.
@type object_name: string
@param reference_object: the reference class or function.
@type reference_object: class/function
@param other_object: the other class or function to be checked.
@type other_object: class/function
@raise InvenioPluginContainerError: in case the other object is not
compatible with the reference object.
"""
try:
if inspect.isclass(reference_object):
## if the reference_object is a class
if inspect.isclass(other_object):
## if the other_object is a class
if issubclass(other_object, reference_object):
## if the other_object is derived from the reference we
## should check for all the method in the former that
## exists in the the latter, wethever they recursively have
## the same signature.
reference_object_map = dict(
inspect.getmembers(reference_object,
inspect.isroutine))
for other_method_name, other_method_code in \
inspect.getmembers(other_object, inspect.isroutine):
if other_method_name in reference_object_map:
check_signature(object_name,
reference_object_map[other_method_name],
other_method_code)
else:
## if the other_object is not derived from the
## reference_object then all the method declared in the
## latter should exist in the former and they should
## recursively have the same signature.
other_object_map = dict(
inspect.getmembers(other_object, inspect.isroutine))
for reference_method_name, reference_method_code in \
inspect.getmembers(
reference_object, inspect.isroutine):
if reference_method_name in other_object_map:
check_signature(
object_name, reference_method_code,
other_method_code)
else:
raise InvenioPluginContainerError('"%s", which'
' exists in the reference class, does not'
' exist in the other class, and the reference'
' class is not an anchestor of the other' %
reference_method_name)
else:
## We are comparing apples and oranges!
raise InvenioPluginContainerError("%s (the reference object)"
" is a class while %s (the other object) is not a class" %
(reference_object, other_object))
elif inspect.isroutine(reference_object):
## if the reference_object is a function
if inspect.isroutine(other_object):
## if the other_object is a function we will compare the
## reference_object and other_object function signautre i.e.
## their parameters.
reference_args, reference_varargs, reference_varkw, \
reference_defaults = inspect.getargspec(reference_object)
other_args, other_varargs, other_varkw, \
other_defaults = inspect.getargspec(other_object)
## We normalize the reference_defaults to be a list
if reference_defaults is not None:
reference_defaults = list(reference_defaults)
else:
reference_defaults = []
## We normalize the other_defaults to be a list
if other_defaults is not None:
other_defaults = list(other_defaults)
else:
other_defaults = []
## Check for presence of missing parameters in other function
if not (other_varargs or other_varkw):
for reference_arg in reference_args:
if reference_arg not in other_args:
raise InvenioPluginContainerError('Argument "%s"'
' in reference function %s does not exist in'
' the other function %s' % (reference_arg,
reference_object, other_object))
## Check for presence of additional parameters in other
## function
if not (reference_varargs or reference_varkw):
for other_arg in other_args:
if other_arg not in reference_args:
raise InvenioPluginContainerError('Argument "%s"'
' in other function %s does not exist in the'
' reference function %s' % (other_arg,
other_object, reference_object))
## Check sorting of arguments
for reference_arg, other_arg in map(
None, reference_args, other_args):
if not((reference_arg == other_arg) or
(reference_arg is None and
(reference_varargs or reference_varkw)) or
(other_arg is None and
(other_args or other_varargs))):
raise InvenioPluginContainerError('Argument "%s" in'
' the other function is in the position of'
' argument "%s" in the reference function, i.e.'
' the order of arguments is not respected' %
(other_arg, reference_arg))
if len(reference_defaults) != len(other_defaults) and \
not (reference_args or reference_varargs
or other_args or other_varargs):
raise InvenioPluginContainerError("Default parameters in"
" the other function are not corresponding to the"
" default of parameters of the reference function")
else:
## We are comparing apples and oranges!
raise InvenioPluginContainerError('%s (the reference object)'
' is a function while %s (the other object) is not a'
' function' % (reference_object, other_object))
except InvenioPluginContainerError, err:
try:
sourcefile = inspect.getsourcefile(other_object)
sourceline = inspect.getsourcelines(other_object)[1]
except IOError:
## other_object is not loaded from a real file
sourcefile = 'N/A'
sourceline = 'N/A'
raise InvenioPluginContainerError('Error in checking signature for'
' "%s" as defined at "%s" (line %s): %s' %
(object_name, sourcefile, sourceline, err))
def create_enhanced_plugin_builder(
compulsory_objects=None, optional_objects=None, other_data=None):
"""
Creates a plugin_builder function suitable to extract some specific
objects (either compulsory or optional) and other simpler data
>>> def dummy_needed_funct1(foo, bar):
... pass
>>> class dummy_needed_class1:
... def __init__(self, baz):
... pass
>>> def dummy_optional_funct2(boo):
... pass
>>> create_enhanced_plugin_builder(
... compulsory_objects={
... 'needed_funct1' : dummy_needed_funct1,
... 'needed_class1' : dummy_needed_class1
... },
... optional_objects={
... 'optional_funct2' : dummy_optional_funct2,
... },
... other_data={
... 'CFG_SOME_DATA' : (str, ''),
... 'CFG_SOME_INT' : (int, 0),
... })
<function plugin_builder at 0xb7812064>
@param compulsory_objects: map of name of an object to look for inside
the C{plugin_code} and a I{signature} for a class or callable. Every
name specified in this map B{must exists} in the plugin_code, otherwise
the plugin will fail to load.
@type compulsory_objects: dict
@param optional_objects: map of name of an object to look for inside
the C{plugin_code} and a I{signature} for a class or callable. Every
name specified in this map must B{can exists} in the plugin_code.
@type optional_objects: dict
@param other_data: map of other simple data that can be loaded from
the plugin_code. The map has the same format of the C{content}
parameter of L{invenio.webinterface_handler.wash_urlargd}.
@type other_data: dict
@return: a I{plugin_builder} function that can be used with the
C{PluginContainer} constructor. Such function will build the plugin
in the form of a map, where every key is one of the keys inside
the three maps provided as parameters and the corresponding value
is the expected class or callable or simple data.
"""
from invenio.webinterface_handler import wash_urlargd
def plugin_builder(plugin_name, plugin_code):
"""
Enhanced plugin_builder created by L{create_enhanced_plugin_builder}.
@param plugin_name: the name of the plugin.
@type plugin_name: string
@param plugin_code: the code of the module as just read from
filesystem.
@type plugin_code: module
@return: the plugin in the form of a map.
"""
plugin = {}
if compulsory_objects:
for object_name, object_signature in \
compulsory_objects.iteritems():
the_object = getattr(plugin_code, object_name, None)
if the_object is None:
raise InvenioPluginContainerError('Plugin "%s" does not '
'contain compulsory object "%s"' % (plugin_name,
object_name))
try:
check_signature(object_name, the_object, object_signature)
except InvenioPluginContainerError, err:
raise InvenioPluginContainerError('Plugin "%s" contains '
'object "%s" with a wrong signature: %s' %
(plugin_name, object_name, err))
plugin[object_name] = the_object
if optional_objects:
for object_name, object_signature in optional_objects.iteritems():
the_object = getattr(plugin_code, object_name, None)
if the_object is not None:
try:
check_signature(
object_name,
the_object,
object_signature)
except InvenioPluginContainerError, err:
raise InvenioPluginContainerError('Plugin "%s" '
'contains object "%s" with a wrong signature: %s' %
(plugin_name, object_name, err))
plugin[object_name] = the_object
if other_data:
the_other_data = {}
for data_name, (dummy, data_default) in other_data.iteritems():
the_other_data[data_name] = getattr(
plugin_code, data_name, data_default)
try:
the_other_data = wash_urlargd(the_other_data, other_data)
except Exception, err:
raise InvenioPluginContainerError('Plugin "%s" contains other '
'data with problems: %s' % (plugin_name, err))
plugin.update(the_other_data)
return plugin
return plugin_builder
def get_callable_signature_as_string(the_callable):
"""
Returns a string representing a callable as if it would have been
declared on the prompt.
>>> def foo(arg1, arg2, arg3='val1', arg4='val2', *args, **argd):
... pass
>>> get_callable_signature_as_string(foo)
def foo(arg1, arg2, arg3='val1', arg4='val2', *args, **argd)
@param the_callable: the callable to be analyzed.
@type the_callable: function/callable.
@return: the signature.
@rtype: string
"""
args, varargs, varkw, defaults = inspect.getargspec(the_callable)
tmp_args = list(args)
args_dict = {}
if defaults:
defaults = list(defaults)
else:
defaults = []
while defaults:
args_dict[tmp_args.pop()] = defaults.pop()
while tmp_args:
args_dict[tmp_args.pop()] = None
args_list = []
for arg in args:
if args_dict[arg] is not None:
args_list.append("%s=%s" % (arg, repr(args_dict[arg])))
else:
args_list.append(arg)
if varargs:
args_list.append("*%s" % varargs)
if varkw:
args_list.append("**%s" % varkw)
args_string = ', '.join(args_list)
return "def %s(%s)" % (the_callable.__name__, args_string)
def get_callable_documentation(the_callable):
"""
Returns a string with the callable signature and its docstring.
@param the_callable: the callable to be analyzed.
@type the_callable: function/callable.
@return: the signature.
@rtype: string
"""
return wrap_text_in_a_box(
title=get_callable_signature_as_string(the_callable),
body=(getattr(the_callable, '__doc__') or 'No documentation').replace(
'\n', '\n\n'),
style='ascii_double')
def check_arguments_compatibility(the_callable, argd):
"""
Check if calling the_callable with the given arguments would be correct
or not.
>>> def foo(arg1, arg2, arg3='val1', arg4='val2', *args, **argd):
... pass
>>> try: check_arguments_compatibility(foo, {'arg1': 'bla', 'arg2': 'blo'})
... except ValueError, err: print 'failed'
... else: print 'ok'
ok
>>> try: check_arguments_compatibility(foo, {'arg1': 'bla'})
... except ValueError, err: print 'failed'
... else: print 'ok'
failed
Basically this function is simulating the call:
>>> the_callable(**argd)
but it only checks for the correctness of the arguments, without
actually calling the_callable.
@param the_callable: the callable to be analyzed.
@type the_callable: function/callable
@param argd: the arguments to be passed.
@type argd: dict
@raise ValueError: in case of uncompatibility
"""
if not argd:
argd = {}
args, dummy, varkw, defaults = inspect.getargspec(the_callable)
tmp_args = list(args)
optional_args = []
args_dict = {}
if defaults:
defaults = list(defaults)
else:
defaults = []
while defaults:
arg = tmp_args.pop()
optional_args.append(arg)
args_dict[arg] = defaults.pop()
while tmp_args:
args_dict[tmp_args.pop()] = None
for arg, dummy_value in argd.iteritems():
if arg in args_dict:
del args_dict[arg]
elif not varkw:
raise ValueError('Argument %s not expected when calling callable '
'"%s" with arguments %s' % (
arg, get_callable_signature_as_string(the_callable), argd))
for arg in args_dict.keys():
if arg in optional_args:
del args_dict[arg]
if args_dict:
raise ValueError('Arguments %s not specified when calling callable '
'"%s" with arguments %s' % (
', '.join(args_dict.keys()),
get_callable_signature_as_string(the_callable),
argd))
c4science · Help