def as_view(path):
def decorator(func):
# ..
path_name, klass_name = (path.split(':'))
# ...
@inlineCallbacks
def wrapper(router, request, *args, **kwargs):
# ...
module = importlib.import_module(path_name)
Klass = getattr(module,klass_name)
klass = Klass(router, request,*args, **kwargs)
# ..
result = yield defer.maybeDeferred(klass)
defer.returnValue(result)
# ..
# _conspect_name(wrapper, klass_name)
_conspect_name(wrapper, func.__name__)
_conspect_param(wrapper, func)
_conspect_param_defaults(wrapper, func)
return wrapper
return decorator
python类decorator()的实例源码
def with_error_settings(**new_settings):
"""
TODO.
Arguments:
**new_settings: TODO
Returns:
"""
@decorator.decorator
def dec(f, *args, **kwargs):
old_settings = np.geterr()
np.seterr(**new_settings)
ret = f(*args, **kwargs)
np.seterr(**old_settings)
return ret
return dec
def catch_config_error(method, app, *args, **kwargs):
"""Method decorator for catching invalid config (Trait/ArgumentErrors) during init.
On a TraitError (generally caused by bad config), this will print the trait's
message, and exit the app.
For use on init methods, to prevent invoking excepthook on invalid input.
"""
try:
return method(app, *args, **kwargs)
except (TraitError, ArgumentError) as e:
app.print_help()
app.log.fatal("Bad config encountered during initialization:")
app.log.fatal(str(e))
app.log.debug("Config at the time: %s", app.config)
app.exit(1)
def magics_class(cls):
"""Class decorator for all subclasses of the main Magics class.
Any class that subclasses Magics *must* also apply this decorator, to
ensure that all the methods that have been decorated as line/cell magics
get correctly registered in the class instance. This is necessary because
when method decorators run, the class does not exist yet, so they
temporarily store their information into a module global. Application of
this class decorator copies that global data to the class instance and
clears the global.
Obviously, this mechanism is not thread-safe, which means that the
*creation* of subclasses of Magic should only be done in a single-thread
context. Instantiation of the classes has no restrictions. Given that
these classes are typically created at IPython startup time and before user
application code becomes active, in practice this should not pose any
problems.
"""
cls.registered = True
cls.magics = dict(line = magics['line'],
cell = magics['cell'])
magics['line'] = {}
magics['cell'] = {}
return cls
def skip(msg=None):
"""Decorator factory - mark a test function for skipping from test suite.
Parameters
----------
msg : string
Optional message to be added.
Returns
-------
decorator : function
Decorator, which, when applied to a function, causes SkipTest
to be raised, with the optional message added.
"""
return skipif(True,msg)
def check_session_aspect(func, *args, **kwargs):
"""
The authentication aspect code, provides an aspect for the aop_check_session decorator
"""
# Get a list of the names of the non-keyword arguments
_argument_specifications = getfullargspec(func)
try:
# Try and find _session_id
arg_idx = _argument_specifications.args.index("_session_id")
except:
raise Exception("Authentication aspect for \"" + func.__name__ + "\": No _session_id parameter")
# Check if the session is valid.
_user = check_session(args[arg_idx])
# Call the function and set the _user parameter, if existing.
return alter_function_parameter_and_call(func, args, kwargs, '_user', _user)
def preprocess_args(fun,varnames):
""" Applies fun to variables in varnames before launching the function """
def wrapper(f, *a, **kw):
if hasattr(f, "func_code"):
func_code = f.func_code # Python 2
else:
func_code = f.__code__ # Python 3
names = func_code.co_varnames
new_a = [fun(arg) if (name in varnames) else arg
for (arg, name) in zip(a, names)]
new_kw = {k: fun(v) if k in varnames else v
for (k,v) in kw.items()}
return f(*new_a, **new_kw)
return decorator.decorator(wrapper)
def catch_config_error(method, app, *args, **kwargs):
"""Method decorator for catching invalid config (Trait/ArgumentErrors) during init.
On a TraitError (generally caused by bad config), this will print the trait's
message, and exit the app.
For use on init methods, to prevent invoking excepthook on invalid input.
"""
try:
return method(app, *args, **kwargs)
except (TraitError, ArgumentError) as e:
app.print_help()
app.log.fatal("Bad config encountered during initialization:")
app.log.fatal(str(e))
app.log.debug("Config at the time: %s", app.config)
app.exit(1)
def magics_class(cls):
"""Class decorator for all subclasses of the main Magics class.
Any class that subclasses Magics *must* also apply this decorator, to
ensure that all the methods that have been decorated as line/cell magics
get correctly registered in the class instance. This is necessary because
when method decorators run, the class does not exist yet, so they
temporarily store their information into a module global. Application of
this class decorator copies that global data to the class instance and
clears the global.
Obviously, this mechanism is not thread-safe, which means that the
*creation* of subclasses of Magic should only be done in a single-thread
context. Instantiation of the classes has no restrictions. Given that
these classes are typically created at IPython startup time and before user
application code becomes active, in practice this should not pose any
problems.
"""
cls.registered = True
cls.magics = dict(line = magics['line'],
cell = magics['cell'])
magics['line'] = {}
magics['cell'] = {}
return cls
def skip(msg=None):
"""Decorator factory - mark a test function for skipping from test suite.
Parameters
----------
msg : string
Optional message to be added.
Returns
-------
decorator : function
Decorator, which, when applied to a function, causes SkipTest
to be raised, with the optional message added.
"""
return skipif(True,msg)
def _name(cls, l):
if getattr(cls, '_alias', None):
_alias = cls._alias
else:
return l
if isinstance(l, (set, list, tuple)):
return [ _alias.get(i, i) for i in l ]
elif isinstance(l, (dict, ExpSpace)):
d = dict(l)
for k, v in d.items():
if isinstance(v, basestring) and v in _alias:
d[k] = _alias[v]
return d
else :
return _alias.get(l, l)
### Todo:
# try to read the decorator that were called for _[post|pre]process
# * if @plot then display
# * if @tabulate then ...
def tabulate(*groups):
''' TODO
'''
if len(groups[1:]) == 0 and callable(groups[0]):
# decorator whithout arguments
return groups[0]
else:
row = groups[0]
column = groups[1]
def decorator(fun):
@wraps(fun)
def wrapper(*args, **kwargs):
kernel = fun(*args, **kwargs)
return kernel
return wrapper
return decorator
def preprocess_args(fun, varnames):
""" Applies fun to variables in varnames before launching the function """
def wrapper(f, *a, **kw):
""" MPOCode.
if hasattr(f, "func_code"):
func_code = f.func_code # Python 2
else:
func_code = f.__code__ # Python 3
"""
func_code = f.__code__
names = func_code.co_varnames
new_a = [
fun(arg) if (name in varnames) else arg
for (arg, name) in zip(a, names)
]
new_kw = {k: fun(v) if k in varnames else v for (k, v) in kw.items()}
return f(*new_a, **new_kw)
return decorator.decorator(wrapper)
def preprocess_args(fun, varnames):
""" Applies fun to variables in varnames before launching the function """
def wrapper(f, *a, **kw):
""" MPOCode.
if hasattr(f, "func_code"):
func_code = f.func_code # Python 2
else:
func_code = f.__code__ # Python 3
"""
func_code = f.__code__
names = func_code.co_varnames
new_a = [
fun(arg) if (name in varnames) else arg
for (arg, name) in zip(a, names)
]
new_kw = {k: fun(v) if k in varnames else v for (k, v) in kw.items()}
return f(*new_a, **new_kw)
return decorator.decorator(wrapper)
def memoize(f):
'''
Memoization decorator
Parameters
----------
Returns
-------
'''
cache = {}
info = defaultdict(dict)
@robust_decorator
def wrapped(f,*args,**kwargs):
sig = argument_signature(f,*args,**kwargs)
if not sig in cache:
time,result = f(*args,**kwargs)
info[sig]['density'] = time/sys.getsizeof(result)
cache[sig] = result
return cache[sig]
wrapped.__cache__ = cache
wrapped.__info__ = info
return wrapped(timed(f))
def catch_config_error(method, app, *args, **kwargs):
"""Method decorator for catching invalid config (Trait/ArgumentErrors) during init.
On a TraitError (generally caused by bad config), this will print the trait's
message, and exit the app.
For use on init methods, to prevent invoking excepthook on invalid input.
"""
try:
return method(app, *args, **kwargs)
except (TraitError, ArgumentError) as e:
app.print_help()
app.log.fatal("Bad config encountered during initialization:")
app.log.fatal(str(e))
app.log.debug("Config at the time: %s", app.config)
app.exit(1)
def magics_class(cls):
"""Class decorator for all subclasses of the main Magics class.
Any class that subclasses Magics *must* also apply this decorator, to
ensure that all the methods that have been decorated as line/cell magics
get correctly registered in the class instance. This is necessary because
when method decorators run, the class does not exist yet, so they
temporarily store their information into a module global. Application of
this class decorator copies that global data to the class instance and
clears the global.
Obviously, this mechanism is not thread-safe, which means that the
*creation* of subclasses of Magic should only be done in a single-thread
context. Instantiation of the classes has no restrictions. Given that
these classes are typically created at IPython startup time and before user
application code becomes active, in practice this should not pose any
problems.
"""
cls.registered = True
cls.magics = dict(line = magics['line'],
cell = magics['cell'])
magics['line'] = {}
magics['cell'] = {}
return cls
def skip(msg=None):
"""Decorator factory - mark a test function for skipping from test suite.
Parameters
----------
msg : string
Optional message to be added.
Returns
-------
decorator : function
Decorator, which, when applied to a function, causes SkipTest
to be raised, with the optional message added.
"""
return skipif(True,msg)
def require_roles(roles, func, *args, **kwargs):
"""
endpoints setup with this decorator require the defined roles.
"""
request = args[1]
req_roles = set(roles)
if not request.keystone_user.get('authenticated', False):
return Response({'errors': ["Credentials incorrect or none given."]},
401)
roles = set(request.keystone_user.get('roles', []))
if roles & req_roles:
return func(*args, **kwargs)
return Response({'errors': ["Must have one of the following roles: %s" %
list(req_roles)]}, 403)
def catch_config_error(method, app, *args, **kwargs):
"""Method decorator for catching invalid config (Trait/ArgumentErrors) during init.
On a TraitError (generally caused by bad config), this will print the trait's
message, and exit the app.
For use on init methods, to prevent invoking excepthook on invalid input.
"""
try:
return method(app, *args, **kwargs)
except (TraitError, ArgumentError) as e:
app.print_help()
app.log.fatal("Bad config encountered during initialization:")
app.log.fatal(str(e))
app.log.debug("Config at the time: %s", app.config)
app.exit(1)
def magics_class(cls):
"""Class decorator for all subclasses of the main Magics class.
Any class that subclasses Magics *must* also apply this decorator, to
ensure that all the methods that have been decorated as line/cell magics
get correctly registered in the class instance. This is necessary because
when method decorators run, the class does not exist yet, so they
temporarily store their information into a module global. Application of
this class decorator copies that global data to the class instance and
clears the global.
Obviously, this mechanism is not thread-safe, which means that the
*creation* of subclasses of Magic should only be done in a single-thread
context. Instantiation of the classes has no restrictions. Given that
these classes are typically created at IPython startup time and before user
application code becomes active, in practice this should not pose any
problems.
"""
cls.registered = True
cls.magics = dict(line = magics['line'],
cell = magics['cell'])
magics['line'] = {}
magics['cell'] = {}
return cls
def skip(msg=None):
"""Decorator factory - mark a test function for skipping from test suite.
Parameters
----------
msg : string
Optional message to be added.
Returns
-------
decorator : function
Decorator, which, when applied to a function, causes SkipTest
to be raised, with the optional message added.
"""
return skipif(True,msg)
def api_call(*dargs, **dkwargs):
def internal_call(func):
@wraps(func)
def wrapped_call(func, *args, **kwargs):
try:
result = func(*args, **kwargs)
return result
except QconfException, ex:
raise
except Exception, ex:
raise QconfException(exception=ex)
return decorator(wrapped_call, func)
if len(dargs) == 1 and callable(dargs[0]):
return internal_call(dargs[0])
else:
return internal_call
def method(name):
def decorator(func):
# ...
_conspect_name(func, name)
_conspect_param(func,func)
# ...
return func
return decorator
# ...
# ...
# ...
def preprocess_args(fun: Callable, varnames: List[str]):
"""
Applies fun to variables in varnames before launching the function
"""
def wrapper(f, *a, **kw):
func_code = f.__code__
names = func_code.co_varnames
new_a = [fun(arg) if (name in varnames) else arg
for (arg, name) in zip(a, names)]
new_kw = {k: fun(v) if k in varnames else v
for (k, v) in kw.items()}
return f(*new_a, **new_kw)
return decorator.decorator(wrapper)
def jsonpify(func, *args, **kwargs):
"""A decorator that reformats the output as JSON; or, if the
*callback* parameter is specified (in the HTTP request), as JSONP.
Very much modelled after pylons.decorators.jsonify .
"""
data = func(*args, **kwargs)
return to_jsonp(data)
test_merger_arxiv2arxiv.py 文件源码
项目：inspire-json-merger
作者: inspirehep
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def cover(key):
# collect the schema's key verified by a test
def tag(func):
def wrapper(func, *args, **kwargs):
COVERED_SCHEMA_KEYS.append(key)
return func(*args, **kwargs)
return decorator.decorator(wrapper, func)
return tag
def _get_async_wrapper(self):
def async_wrapper(*args, **kwargs):
# TODO handle this better in the future, but stop the massive error
# caused by MacOSX async runs for now.
try:
import matplotlib as plt
if plt.rcParams['backend'].lower() == 'macosx':
raise EnvironmentError(backend_error_template %
plt.matplotlib_fname())
except ImportError:
pass
# This function's signature is rewritten below using
# `decorator.decorator`. When the signature is rewritten, args[0]
# is the function whose signature was used to rewrite this
# function's signature.
args = args[1:]
pool = concurrent.futures.ProcessPoolExecutor(max_workers=1)
future = pool.submit(_subprocess_apply, self, args, kwargs)
pool.shutdown(wait=False)
return future
async_wrapper = self._rewrite_wrapper_signature(async_wrapper)
self._set_wrapper_properties(async_wrapper)
self._set_wrapper_name(async_wrapper, 'async')
return async_wrapper
def _rewrite_wrapper_signature(self, wrapper):
# Convert the callable's signature into the wrapper's signature and set
# it on the wrapper.
return decorator.decorator(
wrapper, self._callable_sig_converter_(self._callable))
def validate_type(magic_kind):
"""Ensure that the given magic_kind is valid.
Check that the given magic_kind is one of the accepted spec types (stored
in the global `magic_spec`), raise ValueError otherwise.
"""
if magic_kind not in magic_spec:
raise ValueError('magic_kind must be one of %s, %s given' %
magic_kinds, magic_kind)
# The docstrings for the decorator below will be fairly similar for the two
# types (method and function), so we generate them here once and reuse the
# templates below.
