def token_required(self, func=None, callback=None):
"""Decorator for endpoints which require a verified user."""
# logger.debug("token_required received func={!r}, callback={!r}", func, callback)
if func is None:
# called with optional arguments; return value will be called without, so pass them through
return functools.partial(self.token_required, callback=callback)
@six.wraps(func)
def decorator(request, *args, **kwargs):
userinfo = self.check_token(request)
kwargs['userinfo'] = userinfo
args = (request,) + args
if callback is not None:
# logger.debug("calling {!r} with args={!r} and kwargs={!r}", callback, args, kwargs)
callback(*args, **kwargs)
# logger.debug("calling {!r} with args={!r} and kwargs={!r}", func, args, kwargs)
return func(*args, **kwargs)
return decorator
python类wraps()的实例源码
def serialized_endpoint(*_callbacks):
"""Decorator which wraps an endpoint by applying callbacks to the results then serializing to JSON.
First, the decorated function is called and must return a `defer.Deferred`. The callbacks supplied
to the decorator are then applied followed by any callbacks supplied as keyword arguments to the
decorated function. The result is then serialized and returned in the response (with
``Content-Type`` set to ``application/json``).
"""
def decorator(func):
@six.wraps(func)
def wrapped(request, *args, **kwargs):
callbacks = kwargs.pop('callbacks', [])
# logger.debug("in wrapped:\nargs: {!r}\nkwargs: {!r}", args, kwargs)
deferred_list = func(*args, **kwargs)
for callback in list(_callbacks) + callbacks:
deferred_list.addCallback(callback)
deferred_list.addCallback(json.dumps, default=stethoscope.utils.json_serialize_datetime)
request.setHeader('Content-Type', 'application/json')
return deferred_list
return wrapped
return decorator
def __call__(self, func_or_cls):
condition = self.condition
reason = self.reason
if inspect.isfunction(func_or_cls):
@six.wraps(func_or_cls)
def wrapped(*args, **kwargs):
if condition:
raise testtools.TestCase.skipException(reason)
return func_or_cls(*args, **kwargs)
return wrapped
elif inspect.isclass(func_or_cls):
orig_func = getattr(func_or_cls, 'setUp')
@six.wraps(orig_func)
def new_func(self, *args, **kwargs):
if condition:
raise testtools.TestCase.skipException(reason)
orig_func(self, *args, **kwargs)
func_or_cls.setUp = new_func
return func_or_cls
else:
raise TypeError('skipUnless can be used only with functions or '
'classes')
def queryapi(version, kind, nsarg=True):
"""Make Query API.
.. py:decorator:: queryapi
Creates a named query api.
"""
def decorator(func):
@six.wraps(func)
def handler(self, namespace=DEFAULT_NAMESPACE):
if not nsarg:
namespace = None
url = self._generate_url(api_version=version,
kind=kind,
namespace=namespace)
return Query(self, url)
return handler
return decorator
def template(action):
"""Handle template actions.
.. py:decorator:: template
Checks if the kind is 'template' and processes else default processing.
"""
def decorator(func):
@six.wraps(func)
def handler(self, obj, namespace=None):
apiver, kind, _ = validator.validate(obj)
if kind == 'Template':
return self._process_template(apiver, kind, action, obj, namespace)
else:
return func(self, obj, namespace)
return handler
return decorator
function_pattern_matching.py 文件源码
项目：function-pattern-matching
作者: rasguanabana
项目源码
文件源码
阅读 85
收藏 0
点赞 0
评论 0
def _isotherguard(method):
"""Checks if `other` is a GuardFunc object."""
@six.wraps(method)
def checks(self, other):
if not isinstance(other, GuardFunc):
raise TypeError("The right-hand operand has to be instance of GuardFunc")
return method(self, other)
return checks
def retry(*dargs, **dkw):
"""
Decorator function that instantiates the 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]):
def wrap_simple(f):
@six.wraps(f)
def wrapped_f(*args, **kw):
return Retrying().call(f, *args, **kw)
return wrapped_f
return wrap_simple(dargs[0])
else:
def wrap(f):
@six.wraps(f)
def wrapped_f(*args, **kw):
return Retrying(*dargs, **dkw).call(f, *args, **kw)
return wrapped_f
return wrap
def __call__(self, func_or_cls):
if not self.what:
self.what = func_or_cls.__name__ + '()'
msg, details = self._build_message()
if inspect.isfunction(func_or_cls):
@six.wraps(func_or_cls)
def wrapped(*args, **kwargs):
report_deprecated_feature(LOG, msg, details)
return func_or_cls(*args, **kwargs)
return wrapped
elif inspect.isclass(func_or_cls):
orig_init = func_or_cls.__init__
# TODO(tsufiev): change `functools` module to `six` as
# soon as six 1.7.4 (with fix for passing `assigned`
# argument to underlying `functools.wraps`) is released
# and added to the oslo_incubator requrements
@functools.wraps(orig_init, assigned=('__name__', '__doc__'))
def new_init(self, *args, **kwargs):
report_deprecated_feature(LOG, msg, details)
orig_init(self, *args, **kwargs)
func_or_cls.__init__ = new_init
return func_or_cls
else:
raise TypeError('deprecated can be used only with functions or '
'classes')
def read_locked(*args, **kwargs):
"""Acquires & releases a read lock around call into decorated method.
NOTE(harlowja): if no attribute name is provided then by default the
attribute named '_lock' is looked for (this attribute is expected to be
a :py:class:`.ReaderWriterLock`) in the instance object this decorator
is attached to.
"""
def decorator(f):
attr_name = kwargs.get('lock', '_lock')
@six.wraps(f)
def wrapper(self, *args, **kwargs):
rw_lock = getattr(self, attr_name)
with rw_lock.read_lock():
return f(self, *args, **kwargs)
return wrapper
# This is needed to handle when the decorator has args or the decorator
# doesn't have args, python is rather weird here...
if kwargs or not args:
return decorator
else:
if len(args) == 1:
return decorator(args[0])
else:
return decorator
def write_locked(*args, **kwargs):
"""Acquires & releases a write lock around call into decorated method.
NOTE(harlowja): if no attribute name is provided then by default the
attribute named '_lock' is looked for (this attribute is expected to be
a :py:class:`.ReaderWriterLock` object) in the instance object this
decorator is attached to.
"""
def decorator(f):
attr_name = kwargs.get('lock', '_lock')
@six.wraps(f)
def wrapper(self, *args, **kwargs):
rw_lock = getattr(self, attr_name)
with rw_lock.write_lock():
return f(self, *args, **kwargs)
return wrapper
# This is needed to handle when the decorator has args or the decorator
# doesn't have args, python is rather weird here...
if kwargs or not args:
return decorator
else:
if len(args) == 1:
return decorator(args[0])
else:
return decorator
def __getattr__(self, name):
if name in ('_mock_methods', '_mock_unsafe'):
raise AttributeError(name)
elif self._mock_methods is not None:
if name not in self._mock_methods or name in _all_magics:
raise AttributeError("Mock object has no attribute %r" % name)
elif _is_magic(name):
raise AttributeError(name)
if not self._mock_unsafe:
if name.startswith(('assert', 'assret')):
raise AttributeError(name)
result = self._mock_children.get(name)
if result is _deleted:
raise AttributeError(name)
elif result is None:
wraps = None
if self._mock_wraps is not None:
# XXXX should we get the attribute without triggering code
# execution?
wraps = getattr(self._mock_wraps, name)
result = self._get_child_mock(
parent=self, name=name, wraps=wraps, _new_name=name,
_new_parent=self
)
self._mock_children[name] = result
elif isinstance(result, _SpecState):
result = create_autospec(
result.spec, result.spec_set, result.instance,
result.parent, result.name
)
self._mock_children[name] = result
return result
def __init__(self, spec=None, side_effect=None, return_value=DEFAULT,
wraps=None, name=None, spec_set=None, parent=None,
_spec_state=None, _new_name='', _new_parent=None, **kwargs):
self.__dict__['_mock_return_value'] = return_value
_safe_super(CallableMixin, self).__init__(
spec, wraps, name, spec_set, parent,
_spec_state, _new_name, _new_parent, **kwargs
)
self.side_effect = side_effect
def decorate_callable(self, func):
if hasattr(func, 'patchings'):
func.patchings.append(self)
return func
@wraps(func)
def patched(*args, **keywargs):
extra_args = []
entered_patchers = []
exc_info = tuple()
try:
for patching in patched.patchings:
arg = patching.__enter__()
entered_patchers.append(patching)
if patching.attribute_name is not None:
keywargs.update(arg)
elif patching.new is DEFAULT:
extra_args.append(arg)
args += tuple(extra_args)
return func(*args, **keywargs)
except:
if (patching not in entered_patchers and
_is_started(patching)):
# the patcher may have been started, but an exception
# raised whilst entering one of its additional_patchers
entered_patchers.append(patching)
# Pass the exception to __exit__
exc_info = sys.exc_info()
# re-raise the exception
raise
finally:
for patching in reversed(entered_patchers):
patching.__exit__(*exc_info)
patched.patchings = [self]
return patched
def __call__(self, f):
if isinstance(f, ClassTypes):
return self.decorate_class(f)
@wraps(f)
def _inner(*args, **kw):
self._patch_dict()
try:
return f(*args, **kw)
finally:
self._unpatch_dict()
return _inner
def _check_return_type(self, func):
@wraps(func)
def check_return_type(*args, **kwargs):
result = func(*args, **kwargs)
self._test_case.assertIsInstance(result, self._return_types)
return result
return check_return_type
def _check_input_not_mutated(self, func):
@wraps(func)
def check_not_mutated(*args, **kwargs):
# Copy inputs to compare them to originals later.
arg_copies = [(i, arg.copy()) for i, arg in enumerate(args)
if isinstance(arg, (NDFrame, np.ndarray))]
kwarg_copies = {
k: v.copy() for k, v in iteritems(kwargs)
if isinstance(v, (NDFrame, np.ndarray))
}
result = func(*args, **kwargs)
# Check that inputs weren't mutated by func.
for i, arg_copy in arg_copies:
assert_allclose(
args[i],
arg_copy,
atol=0.5 * 10 ** (-DECIMAL_PLACES),
err_msg="Input 'arg %s' mutated by %s"
% (i, func.__name__),
)
for kwarg_name, kwarg_copy in iteritems(kwarg_copies):
assert_allclose(
kwargs[kwarg_name],
kwarg_copy,
atol=0.5 * 10 ** (-DECIMAL_PLACES),
err_msg="Input '%s' mutated by %s"
% (kwarg_name, func.__name__),
)
return result
return check_not_mutated
def read_locked(*args, **kwargs):
"""Acquires & releases a read lock around call into decorated method.
NOTE(harlowja): if no attribute name is provided then by default the
attribute named '_lock' is looked for (this attribute is expected to be
a :py:class:`.ReaderWriterLock`) in the instance object this decorator
is attached to.
"""
def decorator(f):
attr_name = kwargs.get('lock', '_lock')
@six.wraps(f)
def wrapper(self, *args, **kwargs):
rw_lock = getattr(self, attr_name)
with rw_lock.read_lock():
return f(self, *args, **kwargs)
return wrapper
# This is needed to handle when the decorator has args or the decorator
# doesn't have args, python is rather weird here...
if kwargs or not args:
return decorator
else:
if len(args) == 1:
return decorator(args[0])
else:
return decorator
def write_locked(*args, **kwargs):
"""Acquires & releases a write lock around call into decorated method.
NOTE(harlowja): if no attribute name is provided then by default the
attribute named '_lock' is looked for (this attribute is expected to be
a :py:class:`.ReaderWriterLock` object) in the instance object this
decorator is attached to.
"""
def decorator(f):
attr_name = kwargs.get('lock', '_lock')
@six.wraps(f)
def wrapper(self, *args, **kwargs):
rw_lock = getattr(self, attr_name)
with rw_lock.write_lock():
return f(self, *args, **kwargs)
return wrapper
# This is needed to handle when the decorator has args or the decorator
# doesn't have args, python is rather weird here...
if kwargs or not args:
return decorator
else:
if len(args) == 1:
return decorator(args[0])
else:
return decorator
def keras_test(func):
'''Clean up after tensorflow tests.
'''
@six.wraps(func)
def wrapper(*args, **kwargs):
output = func(*args, **kwargs)
if K._BACKEND == 'tensorflow':
K.clear_session()
return output
return wrapper
def dummy_decorator(f):
@six.wraps(f)
def wrapper(*args, **kwargs):
return f(*args, **kwargs)
return wrapper
