我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用decorator.decorator()。
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
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 _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 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 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 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)
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.
def _method_magic_marker(magic_kind): """Decorator factory for methods in Magics subclasses. """ validate_type(magic_kind) # This is a closure to capture the magic_kind. We could also use a class, # but it's overkill for just that one bit of state. def magic_deco(arg): call = lambda f, *a, **k: f(*a, **k) if callable(arg): # "Naked" decorator call (just @foo, no args) func = arg name = func.__name__ retval = decorator(call, func) record_magic(magics, magic_kind, name, name) elif isinstance(arg, str): # Decorator called with arguments (@foo('bar')) name = arg def mark(func, *a, **kw): record_magic(magics, magic_kind, name, func.__name__) return decorator(call, func) retval = mark else: raise TypeError("Decorator can only be called with " "string or function") return retval # Ensure the resulting decorator has a usable docstring magic_deco.__doc__ = _docstring_template.format('method', magic_kind) return magic_deco
def register(self, *magic_objects): """Register one or more instances of Magics. Take one or more classes or instances of classes that subclass the main `core.Magic` class, and register them with IPython to use the magic functions they provide. The registration process will then ensure that any methods that have decorated to provide line and/or cell magics will be recognized with the `%x`/`%%x` syntax as a line/cell magic respectively. If classes are given, they will be instantiated with the default constructor. If your classes need a custom constructor, you should instanitate them first and pass the instance. The provided arguments can be an arbitrary mix of classes and instances. Parameters ---------- magic_objects : one or more classes or instances """ # Start by validating them to ensure they have all had their magic # methods registered at the instance level for m in magic_objects: if not m.registered: raise ValueError("Class of magics %r was constructed without " "the @register_magics class decorator") if isinstance(m, type): # If we're given an uninstantiated class m = m(shell=self.shell) # Now that we have an instance, we can register it and update the # table of callables self.registry[m.__class__.__name__] = m for mtype in magic_kinds: self.magics[mtype].update(m.magics[mtype])
def apply_wrapper(wrapper, func): """Apply a wrapper to a function for decoration. This mixes Michele Simionato's decorator tool with nose's make_decorator, to apply a wrapper in a decorator so that all nose attributes, as well as function signature and other properties, survive the decoration cleanly. This will ensure that wrapped functions can still be well introspected via IPython, for example. """ warnings.warn("The function `apply_wrapper` is deprecated since IPython 4.0", DeprecationWarning, stacklevel=2) import nose.tools return decorator(wrapper,nose.tools.make_decorator(func)(wrapper))
def module_not_available(module): """Can module be imported? Returns true if module does NOT import. This is used to make a decorator to skip tests that require module to be available, but delay the 'import numpy' to test execution time. """ try: mod = import_module(module) mod_not_avail = False except ImportError: mod_not_avail = True return mod_not_avail
def icache(f): f.cache = {} return decorator.decorator(_icache,f)
def indicator(f): f.storage = {} aspecs = getargspec(f).args f.tpos = aspecs.index('_ts') assert f.tpos == len(aspecs)-1,'position of _ts is invalid' #_ts????????????? return decorator.decorator(_indicator,f) ############## # ???? # ???ma ############## #@icache
def aop_check_session(f): """ CherryPy-specific session checking decorator. :param f: the function """ return decorator(_check_session_aspect, f)
def aop_login_json(f): """ CherryPy-specific session login decorator. """ return decorator(_login_json_aspect, f)
def aop_check_session(f): """ A session validation decorator, searches that parameters for _session_id and checks that session """ return decorator(check_session_aspect, f)
def not_implemented(f): """ Raises a NotImplementedError error. Example: "Internal error in function_name: Not implemented." """ return decorator(_not_implemented, f)
def decorator(w): """ The C{decorator} module was not found. You can install it from: U{http://pypi.python.org/pypi/decorator/} """ def d(fn): @functools.wraps(fn) def x(*argv, **argd): return w(fn, *argv, **argd) return x return d #------------------------------------------------------------------------------
def __call__(self, f): """ Method call wrapper. As this decorator has arguments, this method will only be called once as a part of the decoration process, receiving only one argument: the decorated function ('f'). As a result of this kind of decorator, this method must return the callable that will wrap the decorated function. """ if isinstance(f, property): f.fget._aliases = self.aliases else: f._aliases = self.aliases return f
def apply_to_mask(f, clip, *a, **k): """ This decorator will apply the same function f to the mask of the clip created with f """ newclip = f(clip, *a, **k) if hasattr(newclip, 'mask') and (newclip.mask is not None): newclip.mask = f(newclip.mask, *a, **k) return newclip
def apply_to_audio(f, clip, *a, **k): """ This decorator will apply the function f to the audio of the clip created with f """ newclip = f(clip, *a, **k) if hasattr(newclip, 'audio') and (newclip.audio is not None): newclip.audio = f(newclip.audio, *a, **k) return newclip
def audio_video_fx(f, clip, *a, **k): """ Use an audio function on a video/audio clip This decorator tells that the function f (audioclip -> audioclip) can be also used on a video clip, at which case it returns a videoclip with unmodified video and modified audio. """ if hasattr(clip, "audio"): newclip = clip.copy() if clip.audio is not None: newclip.audio = f(clip.audio, *a, **k) return newclip else: return f(clip, *a, **k)
def cache_disk(function): """ Lazy evaluation of function, by storing the results to the disk, and not rerunning the function twice for the same arguments. It works by explicitly saving the output to a file and it is designed to work with non-hashable and potentially large input and output data types such as numpy arrays Note ---- Any change in the function code will result re-cache Example ------- >>> import numpy as np >>> @cache_memory >>> def doit(x): ... y = np.random.rand(1208, 1208) ... print("Function called:", np.sum(y)) ... return y >>> for _ in range(12): ... print(np.sum(doit(8))) """ def decorator_apply(dec, func): """Decorate a function by preserving the signature even if dec is not a signature-preserving decorator. This recipe is derived from http://micheles.googlecode.com/hg/decorator/documentation.html#id14 """ return FunctionMaker.create( func, 'return decorated(%(signature)s)', dict(decorated=dec(func)), __wrapped__=func) return decorator_apply(__memory.cache, function) # =========================================================================== # Cache # ===========================================================================
def authorize(authorizer): """Constructs authorizer decorator.""" @decorator.decorator def decorated(func, *args, **kwargs): """Decorated function.""" action = getattr(func, 'auth_action', func.__name__.strip('_')) resource = getattr(func, 'auth_resource', func.__module__.strip('_')) _LOGGER.debug('Authorize: %s %s %r %r', resource, action, args, kwargs) authorizer.authorize(resource, action, args, kwargs) return func(*args, **kwargs) return decorated
def _method_magic_marker(magic_kind): """Decorator factory for methods in Magics subclasses. """ validate_type(magic_kind) # This is a closure to capture the magic_kind. We could also use a class, # but it's overkill for just that one bit of state. def magic_deco(arg): call = lambda f, *a, **k: f(*a, **k) if callable(arg): # "Naked" decorator call (just @foo, no args) func = arg name = func.__name__ retval = decorator(call, func) record_magic(magics, magic_kind, name, name) elif isinstance(arg, string_types): # Decorator called with arguments (@foo('bar')) name = arg def mark(func, *a, **kw): record_magic(magics, magic_kind, name, func.__name__) return decorator(call, func) retval = mark else: raise TypeError("Decorator can only be called with " "string or function") return retval # Ensure the resulting decorator has a usable docstring magic_deco.__doc__ = _docstring_template.format('method', magic_kind) return magic_deco
