from sfa.storage.parameter import Parameter, Mixed, python_type, xmlrpc_type
+
class Method:
"""
Base class for all SfaAPI functions. At a minimum, all SfaAPI
# API may set this to a (addr, port) tuple if known
self.source = None
-
+
def __call__(self, *args, **kwds):
"""
Main entry point for all SFA API functions. Type checks
start = time.time()
methodname = self.name
if not self.api.interface or self.api.interface not in self.interfaces:
- raise SfaInvalidAPIMethod(methodname, self.api.interface)
+ raise SfaInvalidAPIMethod(methodname, self.api.interface)
(min_args, max_args, defaults) = self.args()
-
+
# Check that the right number of arguments were passed in
if len(args) < len(min_args) or len(args) > len(max_args):
- raise SfaInvalidArgumentCount(len(args), len(min_args), len(max_args))
+ raise SfaInvalidArgumentCount(
+ len(args), len(min_args), len(max_args))
for name, value, expected in zip(max_args, args, self.accepts):
self.type_check(name, value, expected, args)
- logger.debug("method.__call__ [%s] : BEG %s"%(self.api.interface,methodname))
+ logger.debug("method.__call__ [%s] : BEG %s" % (
+ self.api.interface, methodname))
result = self.call(*args, **kwds)
runtime = time.time() - start
- logger.debug("method.__call__ [%s] : END %s in %02f s (%s)"%\
- (self.api.interface,methodname,runtime,getattr(self,'message',"[no-msg]")))
+ logger.debug("method.__call__ [%s] : END %s in %02f s (%s)" %
+ (self.api.interface, methodname, runtime, getattr(self, 'message', "[no-msg]")))
return result
caller = ""
# Prepend caller and method name to expected faults
- fault.faultString = caller + ": " + self.name + ": " + fault.faultString
+ fault.faultString = caller + ": " + self.name + ": " + fault.faultString
runtime = time.time() - start
- logger.log_exc("Method %s raised an exception"%self.name)
+ logger.log_exc("Method %s raised an exception" % self.name)
raise fault
-
- def help(self, indent = " "):
+ def help(self, indent=" "):
"""
Text documentation for the method.
"""
(min_args, max_args, defaults) = self.args()
- text = "%s(%s) -> %s\n\n" % (self.name, ", ".join(max_args), xmlrpc_type(self.returns))
+ text = "%s(%s) -> %s\n\n" % (self.name,
+ ", ".join(max_args), xmlrpc_type(self.returns))
text += "Description:\n\n"
- lines = [indent + line.strip() for line in self.__doc__.strip().split("\n")]
+ lines = [indent + line.strip()
+ for line in self.__doc__.strip().split("\n")]
text += "\n".join(lines) + "\n\n"
def param_text(name, param, indent, step):
# Print parameter documentation right below type
if isinstance(param, Parameter):
- wrapper = textwrap.TextWrapper(width = 70,
- initial_indent = " " * param_offset,
- subsequent_indent = " " * param_offset)
+ wrapper = textwrap.TextWrapper(width=70,
+ initial_indent=" " * param_offset,
+ subsequent_indent=" " * param_offset)
text += "\n".join(wrapper.wrap(param.doc)) + "\n"
param = param.type
That represents the minimum and maximum sets of arguments that
this function accepts and the defaults for the optional arguments.
"""
-
+
# Inspect call. Remove self from the argument list.
- max_args = self.call.func_code.co_varnames[1:self.call.func_code.co_argcount]
+ max_args = self.call.func_code.co_varnames[
+ 1:self.call.func_code.co_argcount]
defaults = self.call.func_defaults
if defaults is None:
defaults = ()
min_args = max_args[0:len(max_args) - len(defaults)]
defaults = tuple([None for arg in min_args]) + defaults
-
+
return (min_args, max_args, defaults)
def type_check(self, name, value, expected, args):
which may be a Python type, a typed value, a Parameter, a
Mixed type, or a list or dictionary of possibly mixed types,
values, Parameters, or Mixed types.
-
+
Extraneous members of lists must be of the same type as the
last specified type. For example, if the expected argument
type is [int, bool], then [1, False] and [14, True, False,
# If an authentication structure is expected, save it and
# authenticate after basic type checking is done.
- #if isinstance(expected, Auth):
+ # if isinstance(expected, Auth):
# auth = expected
- #else:
+ # else:
# auth = None
# Get actual expected type from within the Parameter structure
pass
elif not isinstance(value, expected_type):
- raise SfaInvalidArgument("expected %s, got %s" % \
- (xmlrpc_type(expected_type), xmlrpc_type(type(value))),
+ raise SfaInvalidArgument("expected %s, got %s" %
+ (xmlrpc_type(expected_type),
+ xmlrpc_type(type(value))),
name)
# If a minimum or maximum (length, value) has been specified
if issubclass(expected_type, StringType):
if min is not None and \
len(value.encode(self.api.encoding)) < min:
- raise SfaInvalidArgument("%s must be at least %d bytes long" % (name, min))
+ raise SfaInvalidArgument(
+ "%s must be at least %d bytes long" % (name, min))
if max is not None and \
len(value.encode(self.api.encoding)) > max:
- raise SfaInvalidArgument("%s must be at most %d bytes long" % (name, max))
+ raise SfaInvalidArgument(
+ "%s must be at most %d bytes long" % (name, max))
elif expected_type in (list, tuple, set):
if min is not None and len(value) < min:
- raise SfaInvalidArgument("%s must contain at least %d items" % (name, min))
+ raise SfaInvalidArgument(
+ "%s must contain at least %d items" % (name, min))
if max is not None and len(value) > max:
- raise SfaInvalidArgument("%s must contain at most %d items" % (name, max))
+ raise SfaInvalidArgument(
+ "%s must contain at most %d items" % (name, max))
else:
if min is not None and value < min:
raise SfaInvalidArgument("%s must be > %s" % (name, str(min)))
elif isinstance(expected, dict):
for key in value.keys():
if key in expected:
- self.type_check(name + "['%s']" % key, value[key], expected[key], args)
+ self.type_check(name + "['%s']" %
+ key, value[key], expected[key], args)
for key, subparam in expected.iteritems():
if isinstance(subparam, Parameter) and \
subparam.optional is not None and \
not subparam.optional and key not in value.keys():
raise SfaInvalidArgument("'%s' not specified" % key, name)
- #if auth is not None:
+ # if auth is not None:
# auth.check(self, *args)
git://git.onelab.eu / sfa.git / blobdiff