#
import time
-from types import IntType, LongType, StringTypes
+from types import IntType, LongType
import textwrap
-
from sfa.util.sfalogging import logger
+from sfa.util.py23 import StringType
from sfa.util.faults import SfaFault, SfaInvalidAPIMethod, SfaInvalidArgumentCount, SfaInvalidArgument
-from sfa.util.parameter import Parameter, Mixed, python_type, xmlrpc_type
+
+from sfa.storage.parameter import Parameter, Mixed, python_type, xmlrpc_type
+
class Method:
"""
# API may set this to a (addr, port) tuple if known
self.source = None
-
+
def __call__(self, *args, **kwds):
"""
- Main entry point for all SfaAPI functions. Type checks
+ Main entry point for all SFA API functions. Type checks
arguments, authenticates, and executes call().
"""
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)
- if self.api.config.SFA_API_DEBUG:
- 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
- if self.api.config.SFA_API_DEBUG or hasattr(self, 'message'):
- 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
- except SfaFault, fault:
+ except SfaFault as fault:
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
# Indent struct fields and mixed types
if isinstance(param, dict):
- for name, subparam in param.iteritems():
+ for name, subparam in param.items():
text += param_text(name, subparam, indent + step, step)
elif isinstance(param, Mixed):
for subparam in param:
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]
- defaults = self.call.func_defaults
+ max_args = self.call.__code__.co_varnames[
+ 1:self.call.__code__.co_argcount]
+ defaults = self.call.__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,
try:
self.type_check(name, value, item, args)
return
- except SfaInvalidArgument, fault:
+ except SfaInvalidArgument as fault:
pass
raise fault
# 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
# Strings are a special case. Accept either unicode or str
# types if a string is expected.
- if expected_type in StringTypes and isinstance(value, StringTypes):
+ if issubclass(expected_type, StringType) and isinstance(value, StringType):
pass
# Integers and long integers are also special types. Accept
pass
elif not isinstance(value, expected_type):
- raise SfaInvalidArgument("expected %s, got %s" % \
+ 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 expected_type in StringTypes:
+ 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))
+ raise SfaInvalidArgument("%s must be > %s" % (name, str(min)))
if max is not None and value > max:
- raise SfaInvalidArgument, "%s must be < %s" % (name, str(max))
+ raise SfaInvalidArgument("%s must be < %s" % (name, str(max)))
# If a list with particular types of items is expected
if isinstance(expected, (list, tuple, set)):
# If a struct with particular (or required) types of items is
# expected.
elif isinstance(expected, dict):
- for key in value.keys():
+ for key in list(value.keys()):
if key in expected:
- self.type_check(name + "['%s']" % key, value[key], expected[key], args)
- for key, subparam in expected.iteritems():
+ self.type_check(name + "['%s']" %
+ key, value[key], expected[key], args)
+ for key, subparam in expected.items():
if isinstance(subparam, Parameter) and \
subparam.optional is not None and \
- not subparam.optional and key not in value.keys():
+ not subparam.optional and key not in list(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