def retry(*dargs, **dkw):
"""Wrap a function with a new `Retrying` object.
:param dargs: positional arguments passed to Retrying object
:param dkw: keyword arguments passed to the Retrying object
"""
# support both @retry and @retry() as valid syntax
if len(dargs) == 1 and callable(dargs[0]):
return retry()(dargs[0])
else:
def wrap(f):
if asyncio and asyncio.iscoroutinefunction(f):
r = AsyncRetrying(*dargs, **dkw)
elif tornado and tornado.gen.is_coroutine_function(f):
r = TornadoRetrying(*dargs, **dkw)
else:
r = Retrying(*dargs, **dkw)
return r.wraps(f)
return wrap
python类callable()的实例源码
def __translate_root__(self, item):
'''
Creates a root controller instance from a string root, e.g.,
> __translate_root__("myproject.controllers.RootController")
myproject.controllers.RootController()
:param item: The string to the item
'''
if '.' in item:
parts = item.split('.')
name = '.'.join(parts[:-1])
fromlist = parts[-1:]
module = __import__(name, fromlist=fromlist)
kallable = getattr(module, parts[-1])
msg = "%s does not represent a callable class or function."
if not six.callable(kallable):
raise TypeError(msg % item)
return kallable()
raise ImportError('No item named %s' % item)
def invoke_or_get(f, *args, **kwargs):
"""if f is callable this function invoke f with given args and kwargs
and return the value. If f is not callable return f directly.
This function is used to provide options to give methods instead of
attributes.
"""
if six.callable(f):
return f(*args, **kwargs)
else:
return f
def invoke_or_get(f, *args, **kwargs):
"""if f is callable this function invoke f with given args and kwargs
and return the value. If f is not callable return f directly.
This function is used to provide options to give methods instead of
attributes.
"""
if six.callable(f):
return f(*args, **kwargs)
else:
return f
def invoke_or_get(f, *args, **kwargs):
"""if f is callable this function invoke f with given args and kwargs
and return the value. If f is not callable return f directly.
This function is used to provide options to give methods instead of
attributes.
"""
if six.callable(f):
return f(*args, **kwargs)
else:
return f
def _maybe_transform_result(self, idx, result, **kwargs):
try:
grh = self.request.girder_result_hooks[idx]
if hasattr(grh, 'transform') and \
six.callable(grh.transform):
return grh.transform(result, **kwargs)
except IndexError:
return result
def _maybe_transform_argument(self, arg, **kwargs):
if hasattr(arg, 'transform') and six.callable(arg.transform):
return arg.transform(**kwargs)
return arg
def _maybe_cleanup(self, arg, **kwargs):
if hasattr(arg, 'cleanup') and six.callable(arg.cleanup):
arg.cleanup(**kwargs)
def _maybe_model_repr(obj):
if hasattr(obj, '_repr_model_') and six.callable(obj._repr_model_):
return obj._repr_model_()
return obj
def _maybe_transform(obj, *args, **kwargs):
if hasattr(obj, 'transform') and six.callable(obj.transform):
return obj.transform(*args, **kwargs)
return obj
def _hashfn(hashfn):
"""Return an initialised hash object, by function, name or integer id
>>> _hashfn(SHA1) # doctest: +ELLIPSIS
<sha1 HASH object @ 0x...>
>>> _hashfn('sha2-256') # doctest: +ELLIPSIS
<sha256 HASH object @ 0x...>
>>> _hashfn('18') # doctest: +ELLIPSIS
<sha256 HASH object @ 0x...>
>>> _hashfn('md5')
Traceback (most recent call last):
...
ValueError: Unknown hash function "md5"
"""
if six.callable(hashfn):
return hashfn()
elif isinstance(hashfn, six.integer_types):
return FUNCS[hashfn]()
elif isinstance(hashfn, six.string_types):
if hashfn in NAMES:
return FUNCS[NAMES[hashfn]]()
elif hashfn.isdigit():
return _hashfn(int(hashfn))
raise ValueError('Unknown hash function "{0}"'.format(hashfn))
def get_checker(x):
if inspect.isclass(x): return default_checker
if six.callable(x): return callable_checker
if is_iterable(x):
def checker(obj, _type_):
checkers = map(lambda e:get_checker(e)(obj, e), _type_)
return any_checkers(checkers), format_checkers(checkers)
return checker
def get_param_values(cls, *args, **kwargs):
values = super(BaseTask, cls).get_param_values(*args, **kwargs)
if six.callable(cls.modify_param_values):
return cls.modify_param_values(OrderedDict(values)).items()
else:
return values
def _waiter_takes_last_result(waiter):
if not six.callable(waiter):
return False
if isinstance(waiter, _wait.wait_base):
waiter = waiter.__call__
waiter_spec = inspect.getargspec(waiter)
return 'last_result' in waiter_spec.args
def test_callable():
class X:
def __call__(self):
pass
def method(self):
pass
assert six.callable(X)
assert six.callable(X())
assert six.callable(test_callable)
assert six.callable(hasattr)
assert six.callable(X.method)
assert six.callable(X().method)
assert not six.callable(4)
assert not six.callable("string")
def __init__(self, root, default_renderer='mako',
template_path='templates', hooks=lambda: [],
custom_renderers={}, extra_template_vars={},
force_canonical=True, guess_content_type_from_ext=True,
context_local_factory=None, request_cls=Request,
response_cls=Response, **kw):
if isinstance(root, six.string_types):
root = self.__translate_root__(root)
self.root = root
self.request_cls = request_cls
self.response_cls = response_cls
self.renderers = RendererFactory(custom_renderers, extra_template_vars)
self.default_renderer = default_renderer
# pre-sort these so we don't have to do it per-request
if six.callable(hooks):
hooks = hooks()
self.hooks = list(sorted(
hooks,
key=operator.attrgetter('priority')
))
self.template_path = template_path
self.force_canonical = force_canonical
self.guess_content_type_from_ext = guess_content_type_from_ext
def test_callable():
class X:
def __call__(self):
pass
def method(self):
pass
assert six.callable(X)
assert six.callable(X())
assert six.callable(test_callable)
assert six.callable(hasattr)
assert six.callable(X.method)
assert six.callable(X().method)
assert not six.callable(4)
assert not six.callable("string")
def get_name(component):
component = resolve_alias(component)
if six.callable(component):
return '.'.join([component.__module__, component.__name__])
return str(component)
def get_simple_name(component):
component = resolve_alias(component)
if six.callable(component):
return component.__name__
return str(component)
def __get__(self, instance, instance_type=None):
if instance is not None:
value = instance._data.get(self.attr_name)
if value is None:
value = copy.deepcopy(self.field.default)
instance._data[self.attr_name] = value
if isinstance(value, dict):
value = ObjectDict(value)
if value and not isinstance(value, (dict, list, tuple)) and \
six.callable(self.field.converter):
value = self.field.converter(value)
return value
return self.field
def add_field(self, field):
"""Add the received field to the model."""
self.remove_field(field.name)
self._fields[field.name] = field
if field.default is not None:
if six.callable(field.default):
self._default_callables[field.key] = field.default
else:
self._defaults[field.key] = field.default
def remove_field(self, field_name):
"""Remove the field with the received field name from model."""
field = self._fields.pop(field_name, None)
if field is not None and field.default is not None:
if six.callable(field.default):
self._default_callables.pop(field.key, None)
else:
self._defaults.pop(field.key, None)
def verify_black_box_function(predict_method, number_of_features,
number_of_data_points=10):
# check estimator variable is a callable.
if not six.callable(predict_method):
raise Exception("Please pass in a callable.")
# now generate test data to verify that estimator is working
covariance = np.eye(number_of_features)
mean = np.zeros(number_of_features)
data = np.random.multivariate_normal(mean, covariance,
number_of_data_points)
try:
output = predict_method(data)
# check to make sure that the estimator returns a numpy
# array
if type(output).__module__ != 'numpy':
raise ValueError("Output of predict function is not "
"a numpy array")
if output.shape[0] != number_of_data_points:
raise Exception("Predict does not return an output "
"for every data point.")
except:
print("Unexpected error: ", sys.exc_info()[0])
return True
def __get__(self, instance, instance_type=None):
if instance is not None:
value = instance._data.get(self.attr_name)
if value is None:
value = copy.deepcopy(self.field.default)
instance._data[self.attr_name] = value
if isinstance(value, dict):
value = ObjectDict(value)
if value and not isinstance(value, (dict, list, tuple)) and \
six.callable(self.field.converter):
value = self.field.converter(value)
return value
return self.field
def test_callable():
class X:
def __call__(self):
pass
def method(self):
pass
assert six.callable(X)
assert six.callable(X())
assert six.callable(test_callable)
assert six.callable(hasattr)
assert six.callable(X.method)
assert six.callable(X().method)
assert not six.callable(4)
assert not six.callable("string")
def default(self, obj):
'''
Converts an object and returns a ``JSON``-friendly structure.
:param obj: object or structure to be converted into a
``JSON``-ifiable structure
Considers the following special cases in order:
* object has a callable __json__() attribute defined
returns the result of the call to __json__()
* date and datetime objects
returns the object cast to str
* Decimal objects
returns the object cast to float
* SQLAlchemy objects
returns a copy of the object.__dict__ with internal SQLAlchemy
parameters removed
* SQLAlchemy ResultProxy objects
Casts the iterable ResultProxy into a list of tuples containing
the entire resultset data, returns the list in a dictionary
along with the resultset "row" count.
.. note:: {'count': 5, 'rows': [('Ed Jones',), ('Pete Jones',),
('Wendy Williams',), ('Mary Contrary',), ('Fred Smith',)]}
* SQLAlchemy RowProxy objects
Casts the RowProxy cursor object into a dictionary, probably
losing its ordered dictionary behavior in the process but
making it JSON-friendly.
* webob_dicts objects
returns webob_dicts.mixed() dictionary, which is guaranteed
to be JSON-friendly.
'''
if hasattr(obj, '__json__') and six.callable(obj.__json__):
return obj.__json__()
elif isinstance(obj, (date, datetime)):
return str(obj)
elif isinstance(obj, Decimal):
# XXX What to do about JSONEncoder crappy handling of Decimals?
# SimpleJSON has better Decimal encoding than the std lib
# but only in recent versions
return float(obj)
elif is_saobject(obj):
props = {}
for key in obj.__dict__:
if not key.startswith('_sa_'):
props[key] = getattr(obj, key)
return props
elif isinstance(obj, ResultProxy):
props = dict(rows=list(obj), count=obj.rowcount)
if props['count'] < 0:
props['count'] = len(props['rows'])
return props
elif isinstance(obj, RowProxy):
return dict(obj)
elif isinstance(obj, webob_dicts):
return obj.mixed()
else:
return JSONEncoder.default(self, obj)
def getargspec(method):
"""
Drill through layers of decorators attempting to locate the actual argspec
for a method.
"""
argspec = _getargspec(method)
args = argspec[0]
if args and args[0] == 'self':
return argspec
if hasattr(method, '__func__'):
method = method.__func__
func_closure = six.get_function_closure(method)
# NOTE(sileht): if the closure is None we cannot look deeper,
# so return actual argspec, this occurs when the method
# is static for example.
if not func_closure:
return argspec
closure = None
# In the case of deeply nested decorators (with arguments), it's possible
# that there are several callables in scope; Take a best guess and go
# with the one that looks most like a pecan controller function
# (has a __code__ object, and 'self' is the first argument)
func_closure = filter(
lambda c: (
six.callable(c.cell_contents) and
hasattr(c.cell_contents, '__code__')
),
func_closure
)
func_closure = sorted(
func_closure,
key=lambda c: 'self' in c.cell_contents.__code__.co_varnames,
reverse=True
)
closure = func_closure[0]
method = closure.cell_contents
return getargspec(method)
def get_filter(name_or_function, **kwargs):
'''Retrieve a window given its name or function handle.
Parameters
----------
name_or_function : str or callable
If a function, returns `name_or_function(**kwargs)`.
If a string, and it matches the name of one of the defined
filter functions, the corresponding function is called with `**kwargs`.
If a string, and it matches the name of a pre-computed filter,
the corresponding filter is retrieved, and kwargs is ignored.
Valid pre-computed filter names are:
- 'kaiser_fast'
- 'kaiser_best'
Returns
-------
half_window : np.ndarray
The right wing of the interpolation filter
precision : int > 0
The number of samples between zero-crossings of the filter
rolloff : float > 0
The roll-off frequency of the filter as a fraction of Nyquist
Raises
------
NotImplementedError
If `name_or_function` cannot be found as a filter.
'''
if name_or_function in FILTER_FUNCTIONS:
return getattr(sys.modules[__name__], name_or_function)(**kwargs)
elif six.callable(name_or_function):
return name_or_function(**kwargs)
else:
try:
return load_filter(name_or_function)
except (IOError, ValueError):
raise NotImplementedError('Cannot load filter definition for '
'{}'.format(name_or_function))
