Python functools 模块，wraps() 实例源码

def auth(func):
    @wraps(func)
    def _decorator(request, *args, **kwargs):
        auth = get_auth()
        if auth.get('disable', False) is True:
            return func(request, *args, **kwargs)
        if 'authorized_ips' in auth:
            ip = get_client_ip(request)
            if is_authorized(ip, auth['authorized_ips']):
                return func(request, *args, **kwargs)
        prepare_credentials(auth)
        if request.META.get('HTTP_AUTHORIZATION'):
            authmeth, auth = request.META['HTTP_AUTHORIZATION'].split(' ')
            if authmeth.lower() == 'basic':
                auth = base64.b64decode(auth).decode('utf-8')
                username, password = auth.split(':')
                if (username == HEARTBEAT['auth']['username'] and
                        password == HEARTBEAT['auth']['password']):
                    return func(request, *args, **kwargs)

        response = HttpResponse(
            "Authentication failed", status=401)
        response['WWW-Authenticate'] = 'Basic realm="Welcome to 1337"'
        return response
    return _decorator

def meta_compile(compile):
    "Wraps the compile() method to do dark magic, see `meta_shellcode`."

    @functools.wraps(compile)
    def compile_wrapper(self, state = None):
        if state is None:
            state = CompilerState()
        self.offset = state.offset
        if hasattr(self, '_compile_hook'):
            bytes = self._compile_hook(compile, state)
        else:
            bytes = compile(self, state)
        state.next_piece( len(bytes) )
        return bytes

    return compile_wrapper

def timeout(seconds=10, error_message=os.strerror(errno.ETIME)):
    def decorator(func):
        def _handle_timeout(signum, frame):
            raise TimeoutError(error_message)

        def wrapper(*args, **kwargs):
            signal.signal(signal.SIGALRM, _handle_timeout)
            signal.alarm(seconds)
            try:
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)
            return result

        return wraps(func)(wrapper)

    return decorator

def locked(path, timeout=None):
    """Decorator which enables locks for decorated function.

    Arguments:
     - path: path for lockfile.
     - timeout (optional): Timeout for acquiring lock.

     Usage:
         @locked('/var/run/myname', timeout=0)
         def myname(...):
             ...
    """
    def decor(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            lock = FileLock(path, timeout=timeout)
            lock.acquire()
            try:
                return func(*args, **kwargs)
            finally:
                lock.release()
        return wrapper
    return decor

def set_package(fxn):
    """Set __package__ on the returned module.

    This function is deprecated.

    """
    @functools.wraps(fxn)
    def set_package_wrapper(*args, **kwargs):
        warnings.warn('The import system now takes care of this automatically.',
                      DeprecationWarning, stacklevel=2)
        module = fxn(*args, **kwargs)
        if getattr(module, '__package__', None) is None:
            module.__package__ = module.__name__
            if not hasattr(module, '__path__'):
                module.__package__ = module.__package__.rpartition('.')[0]
        return module
    return set_package_wrapper

def __getattr__(self, name):
        # Attribute lookups are delegated to the underlying file
        # and cached for non-numeric results
        # (i.e. methods are cached, closed and friends are not)
        file = self.__dict__['file']
        a = getattr(file, name)
        if hasattr(a, '__call__'):
            func = a
            @_functools.wraps(func)
            def func_wrapper(*args, **kwargs):
                return func(*args, **kwargs)
            # Avoid closing the file as long as the wrapper is alive,
            # see issue #18879.
            func_wrapper._closer = self._closer
            a = func_wrapper
        if not isinstance(a, int):
            setattr(self, name, a)
        return a

    # The underlying __enter__ method returns the wrong object
    # (self.file) so override it to return the wrapper

def login_required(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.info ("get request, path: %s" % request.path)
        token = request.form.get("token", None)
        if (token == None):
            logger.info ("get request without token, path: %s" % request.path)
            return json.dumps({'success':'false', 'message':'user or key is null'})
        result = post_to_user("/authtoken/", {'token':token})
        if result.get('success') == 'true':
            username = result.get('username')
            beans = result.get('beans')
        else:
            return result
        #if (cur_user == None):
        #    return json.dumps({'success':'false', 'message':'token failed or expired', 'Unauthorized': 'True'})
        return func(username, beans, request.form, *args, **kwargs)

    return wrapper

def test_wraps(self):
        self.assert_ok("""\
            from functools import wraps
            def my_decorator(f):
                dec = wraps(f)
                def wrapper(*args, **kwds):
                    print('Calling decorated function')
                    return f(*args, **kwds)
                wrapper = dec(wrapper)
                return wrapper

            @my_decorator
            def example():
                '''Docstring'''
                return 17

            assert example() == 17
            """)

def locked(path, timeout=None):
    """Decorator which enables locks for decorated function.

    Arguments:
     - path: path for lockfile.
     - timeout (optional): Timeout for acquiring lock.

     Usage:
         @locked('/var/run/myname', timeout=0)
         def myname(...):
             ...
    """
    def decor(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            lock = FileLock(path, timeout=timeout)
            lock.acquire()
            try:
                return func(*args, **kwargs)
            finally:
                lock.release()
        return wrapper
    return decor

def ensure_empty_collections(*collections):
    """
    Clears collections listed after function has completed.
    Will throw an assertion if any collection is not empty when called.
    """

    def clear(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            db = api.common.get_conn()
            collection_size = lambda name: len(list(db[name].find()))
            for collection in collections:
                assert collection_size(collection) == 0, "Collection was not empty: " + collection
            result = f(*args, **kwargs)
            return result
        return wrapper
    return clear

def clear_collections(*collections):
    """
    Clears collections listed after function has completed.
    Will throw an assertion if any collection is not empty when called.
    """

    def clear(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            db = api.common.get_conn()
            try:
                result = f(*args, **kwargs)
            finally:
                #Clear the collections.
                for collection in collections:
                    db[collection].remove()
                #Ensure they are then empty.
                for collection in collections:
                    collection_size = lambda collection: len(list(db[collection].find()))
                    assert collection_size(collection) == 0, "Collection: {} was not able to be cleared.".format(collection)
            return result
        return wrapper
    return clear

def block_before_competition(return_result):
    """
    Wraps a routing function that should be blocked before the start time of the competition
    """

    def decorator(f):
        """
        Inner decorator
        """

        @wraps(f)
        def wrapper(*args, **kwds):
            if datetime.utcnow().timestamp() > api.config.get_settings()["start_time"].timestamp():
                return f(*args, **kwds)
            else:
                return return_result
        return wrapper
    return decorator

def _silent_connection_failure(func):
    """Decorator used to avoid raising an exception when the database timeouts

    Parameters:
        func: Function to decorate.
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        """Wraps the function to catch timeout exception.
        """
        if not FLAGS.disable_mongodb_exception:
            return func(*args, **kwargs)

        try:
            result = func(*args, **kwargs)
        except pymongo.errors.ServerSelectionTimeoutError as e:
            logging.error("Unable to reach the caching server: %s", e)
            return None
        return result

    return wrapper

def array_stream(func):
    """ 
    Decorates streaming functions to make sure that the stream
    is a stream of ndarrays. Objects that are not arrays are transformed 
    into arrays. If the stream is in fact a single ndarray, this ndarray 
    is repackaged into a sequence of length 1.

    The first argument of the decorated function is assumed to be an iterable of
    arrays, or an iterable of objects that can be casted to arrays.
    """
    @wraps(func)    # thanks functools
    def decorated(arrays, *args, **kwargs):
        if isinstance(arrays, ndarray):
            arrays = (arrays,)
        return func(map(atleast_1d, arrays), *args, **kwargs)
    return decorated

def call(f):
    """
    Wrap a function in a processor that calls `f` on the argument before
    passing it along.

    Useful for creating simple arguments to the `@preprocess` decorator.

    Parameters
    ----------
    f : function
        Function accepting a single argument and returning a replacement.

    Usage
    -----
    >>> @preprocess(x=call(lambda x: x + 1))
    ... def foo(x):
    ...     return x
    ...
    >>> foo(1)
    2
    """
    @wraps(f)
    def processor(func, argname, arg):
        return f(arg)
    return processor

def require_initialized(exception):
    """
    Decorator for API methods that should only be called after
    TradingAlgorithm.initialize.  `exception` will be raised if the method is
    called before initialize has completed.

    Usage
    -----
    @require_initialized(SomeException("Don't do that!"))
    def method(self):
        # Do stuff that should only be allowed after initialize.
    """
    def decorator(method):
        @wraps(method)
        def wrapped_method(self, *args, **kwargs):
            if not self.initialized:
                raise exception
            return method(self, *args, **kwargs)
        return wrapped_method
    return decorator

def coerce_numbers_to_my_dtype(f):
    """
    A decorator for methods whose signature is f(self, other) that coerces
    ``other`` to ``self.dtype``.

    This is used to make comparison operations between numbers and `Factor`
    instances work independently of whether the user supplies a float or
    integer literal.

    For example, if I write::

        my_filter = my_factor > 3

    my_factor probably has dtype float64, but 3 is an int, so we want to coerce
    to float64 before doing the comparison.
    """
    @wraps(f)
    def method(self, other):
        if isinstance(other, Number):
            other = coerce_to_dtype(self.dtype, other)
        return f(self, other)
    return method

def with_algo(f):
    name = f.__name__
    if not name.startswith('test_'):
        raise ValueError('This must decorate a test case')

    tfm_name = name[len('test_'):]

    @wraps(f)
    def wrapper(self, data_frequency, days=None):
        sim_params, source = self.sim_and_source[data_frequency]

        algo = TradingAlgorithm(
            initialize=initialize_with(self, tfm_name, days),
            handle_data=handle_data_wrapper(f),
            sim_params=sim_params,
            env=self.env,
        )
        algo.run(source)

    return wrapper

def authorized(admin_only=False):
    def wrap(user_handler):
        @wraps(user_handler)
        def authorized_handler(self, *args, **kwargs):
            self.set_cache(is_public=False)
            request = self.request
            if request.method == 'GET':
                if not self.current_user_id:
                    next_url = self.get_argument('next', '/')
                    self.redirect(self.get_login_url() + "?next=" + next_url, status=302 if request.version == 'HTTP/1.0' else 303)
                elif admin_only and not self.is_admin:
                    raise HTTPError(403)
                else:
                    user_handler(self, *args, **kwargs)
            elif not self.current_user_id:
                raise HTTPError(403)
            elif admin_only and not self.is_admin:
                raise HTTPError(403)
            else:
                user_handler(self, *args, **kwargs)
        return authorized_handler
    return wrap

def memoized(func):
    """Decorator that caches a function's return value each time it is called.
    If called later with the same arguments, the cached value is returned, and
    the function is not re-evaluated.

    Based upon from http://wiki.python.org/moin/PythonDecoratorLibrary#Memoize
    Nota bene: this decorator memoizes /all/ calls to the function.
    For a memoization decorator with limited cache size, consider:
    http://code.activestate.com/recipes/496879-memoize-decorator-function-with-cache-size-limit/
    """
    cache = {}

    @wraps(func)
    def func_wrapper(*args, **kwargs):
        key = cPickle.dumps((args, kwargs))
        if key not in cache:
            cache[key] = func(*args, **kwargs)
        return cache[key]
    return func_wrapper

def calculate_mantl_vars(func):
    """calculate Mantl vars"""

    @wraps(func)
    def inner(*args, **kwargs):
        name, attrs, groups = func(*args, **kwargs)

        # attrs
        if attrs.get('role', '') == 'control':
            attrs['consul_is_server'] = True
        else:
            attrs['consul_is_server'] = False

        # groups
        if attrs.get('publicly_routable', False):
            groups.append('publicly_routable')

        return name, attrs, groups

    return inner

def _check_inputs_and_outputs(fcn):
    @functools.wraps(fcn)
    def call(conn, inputs, outputs, identifier, *a, **kw):
        assert isinstance(inputs, (list, tuple, set)), inputs
        assert isinstance(outputs, (list, tuple, set)), outputs
        assert '' not in inputs, inputs
        assert '' not in outputs, outputs
        # this is for actually locking inputs/outputs
        inputs, outputs = list(map(str, inputs)), list(map(str, outputs))
        locks = inputs + [''] + outputs

        if kw.pop('history', None):
            igraph = [EDGE_RE.sub('*', inp) for inp in inputs]
            ograph = [EDGE_RE.sub('*', out) for out in outputs]
            graph_id = EDGE_RE.sub('*', str(identifier))
            graph = igraph + [''] + ograph + ['', graph_id]
            if all(x.startswith('test.') for x in igraph + ograph):
                graph = ['', '']
        else:
            graph = ['', '']

        return fcn(conn, locks, graph, str(identifier), *a, **kw)
    return call

def immutable(cls):
    cls.__frozen = False

    def frozensetattr(self, key, value):
        if self.__frozen and not hasattr(self, key):
            print("Class {} is frozen. Cannot set {} = {}"
                  .format(cls.__name__, key, value))
        else:
            object.__setattr__(self, key, value)

    def init_decorator(func):
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            func(self, *args, **kwargs)
            self.__frozen = True
        return wrapper

    cls.__setattr__ = frozensetattr
    cls.__init__ = init_decorator(cls.__init__)

    return cls

def set_self_blocking(function):

    @functools.wraps(function)
    def wrapper(*args, **kwargs):
        self = args[0]
        try:
            _is_blocking =  self.gettimeout()
            if _is_blocking == 0:
                self.setblocking(True)
            return function(*args, **kwargs)
        except Exception as e:
            raise
        finally:
            # set orgin blocking
            if _is_blocking == 0:
                self.setblocking(False)
    return wrapper

def adapt_rgb(apply_to_rgb):
    """Return decorator that adapts to RGB images to a gray-scale filter.

    This function is only intended to be used for functions that don't accept
    volumes as input, since checking an image's shape is fragile.

    Parameters
    ----------
    apply_to_rgb : function
        Function that returns a filtered image from an image-filter and RGB
        image. This will only be called if the image is RGB-like.
    """
    def decorator(image_filter):
        @functools.wraps(image_filter)
        def image_filter_adapted(image, *args, **kwargs):
            if is_rgb_like(image):
                return apply_to_rgb(image_filter, image, *args, **kwargs)
            else:
                return image_filter(image, *args, **kwargs)
        return image_filter_adapted
    return decorator

def patch(obj, func_name, before_func, after_func):
    if not hasattr(obj, func_name):
        return

    target_func = getattr(obj, func_name)

    # already patched
    if hasattr(target_func, '__stackimpact_orig__'):
        return

    @wraps(target_func)
    def wrapper(*args, **kwds):
        if before_func:
            before_func(*args, **kwds)

        ret = target_func(*args, **kwds)

        if after_func:
            after_func(ret)

        return ret

    wrapper.__orig__ = target_func
    setattr(obj, func_name, wrapper)

def adapted(func):
    @wraps(func)
    def wrapper(**kwargs):
        final_args = {}

        for name, value in kwargs.items():
            adapt = func.__annotations__.get(name)
            if adapt is not None:
                final_args[name] = adapt(value)
            else:
                final_args[name] = value

        result = func(**final_args)

        adapt = func.__annotations__.get('return')
        if adapt is not None:
            return adapt(result)
        return result

    return wrapper

def admin_access():
    def access(func):
        @wraps(func)
        def wrapped(self, bot: Bot, update: Update, *args, **kwargs):
            user = update.effective_user
            admins = getattr(self, 'admins', None)

            if admins is None:
                self.logger.warning('Specify self.admins (list of users ids) parameter in '
                                    'manager or channel classes in order to restrict access to bot.')
                return func(self, bot, update, *args, **kwargs)

            if user.id not in admins:
                self.logger.info(f"Unauthorized access denied for {user}.")
                return

            return func(self, bot, update, *args, **kwargs)
        return wrapped
    return access

def __call__(self, func_or_msg=None, **kwargs):
        """ If the first argument is a function, return a decorator which runs
        the wrapped function inside Timer's context manager.
        Otherwise, treat the first argument as a 'msg' string and return an updated
        Timer instance, referencing the same logger.
        A 'level' keyword can also be passed to override self.level.
        """
        if isinstance(func_or_msg, collections.Callable):
            func = func_or_msg
            # use the function name when no explicit 'msg' is provided
            if not self.msg:
                self.msg = "run '%s'" % func.__name__

            @wraps(func)
            def wrapper(*args, **kwds):
                with self:
                    return func(*args, **kwds)
            return wrapper
        else:
            msg = func_or_msg or kwargs.get("msg")
            level = kwargs.get("level", self.level)
            return self.__class__(self.logger, msg, level)

def retry(times=3):
    def wrapper(func):
        @wraps(func)
        def fun(*args, **kwargs):
            count = 0
            while count < times:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    count = count + 1

            logger.error("connection failed after retried 3 times. {} {}".format(args, kwargs))
            raise Exception("connection failed after retried 3 times. {} {}".format(args, kwargs))

        return fun

    return wrapper

def verify_request_signature(func):
    @wraps(func)
    def decorated(*args, **kwargs):
        signature = request.headers.get('x-hub-signature', None)
        if signature:
            elements = signature.split('=')
            method = elements[0]
            signature_hash = elements[1]
            expected_hash = hmac.new(APP_SECRET, msg=request.get_data(), digestmod=method).hexdigest()
            if signature_hash != expected_hash:
                LOGGER.error('Signature was invalid')
                return make_response('', 403)
        else:
            LOGGER.error('Could not validate the signature')
        return func(*args, **kwargs)

    return decorated

def logger(command):
    """
    Add a logger to the decorated command.
    """

    @wraps(command)
    # pylint: disable=bad-whitespace
    # pylint: disable=unused-argument
    def wrapper(bot, update, **kwargs):
        command(     update, **kwargs)
        if command.__name__ == 'unknown':
            command.__name__ = _get_command_name(update.message.text)
        message = LOG_TEMPLATE.format(user=update.user.first_name,
                                      command=command.__name__)
        logging.info(message)
    return wrapper

def run_in_executor(f):
    """
    A decorator to run the given method in the ThreadPoolExecutor.
    """

    @wraps(f)
    def new_f(self, *args, **kwargs):

        if self.is_shutdown:
            return
        try:
            future = self.executor.submit(f, self, *args, **kwargs)
            future.add_done_callback(_future_completed)
        except Exception:
            log.exception("Failed to submit task to executor")

    return new_f

def handle_bad_input_mc(f):
    @wraps(f)
    def handle(*args, **kwargs):
        pre_type = args[0].pre_type
        if pre_type == PreType.None:
            return handle_bad_input(f)(*args, **kwargs)
        EType = {
            PreType.SimplePre : SimplePre.InvalidMcOperation,
            PreType.SimplePreLAG : SimplePreLAG.InvalidMcOperation
        }[pre_type]
        Codes = {
            PreType.SimplePre : SimplePre.McOperationErrorCode,
            PreType.SimplePreLAG : SimplePreLAG.McOperationErrorCode
        }[pre_type]
        try:
            return handle_bad_input(f)(*args, **kwargs)
        except EType as e:
            error = Codes._VALUES_TO_NAMES[e.code]
            print "Invalid PRE operation (%s)" % error
    return handle

def validate(validation_class, silent=not settings.DEBUG):
    def decorator(fun):
        @functools.wraps(fun)
        def wrapper(*a, **kw):
            if flask.request.method == 'GET':
                p = {k: v for k, v in flask.request.values and flask.request.values.items() or {}}
            else:
                try:
                    p = flask.request.data and json.loads(flask.request.data.decode())
                except Exception:
                    raise exceptions.WrongParametersException
            if silent:
                try:
                    validation_class(p)
                except PyCombValidationError:
                    raise exceptions.WrongParametersException
            else:
                validation_class(p)
            return fun(*a, params=p, **kw)
        return wrapper
    return decorator

def handle_errors(f):
    """A decorator that allows to ignore certain types of errors."""

    @functools.wraps(f)
    def wrapper(*args, **kwargs):
        param_name = 'ignore_errors'
        ignored_errors = kwargs.get(param_name, tuple())

        if param_name in kwargs:
            del kwargs[param_name]

        try:
            return f(*args, **kwargs)
        except ignored_errors:
            # Silently ignore errors
            pass

    return wrapper

def add_arg_scope(func):
  """Decorates a function with args so it can be used within an arg_scope.

  Args:
    func: function to decorate.

  Returns:
    A tuple with the decorated function func_with_args().
  """
  @functools.wraps(func)
  def func_with_args(*args, **kwargs):
    current_scope = _current_arg_scope()
    current_args = kwargs
    key_func = (func.__module__, func.__name__)
    if key_func in current_scope:
      current_args = current_scope[key_func].copy()
      current_args.update(kwargs)
    return func(*args, **current_args)
  _add_op(func)
  return func_with_args

def locked(path, timeout=None):
    """Decorator which enables locks for decorated function.

    Arguments:
     - path: path for lockfile.
     - timeout (optional): Timeout for acquiring lock.

     Usage:
         @locked('/var/run/myname', timeout=0)
         def myname(...):
             ...
    """
    def decor(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            lock = FileLock(path, timeout=timeout)
            lock.acquire()
            try:
                return func(*args, **kwargs)
            finally:
                lock.release()
        return wrapper
    return decor

def overloaded(func):
    @wraps(func)
    def overloaded_func(*args, **kwargs):
        for f in overloaded_func.overloads:
            try:
                return f(*args, **kwargs)
            except TypeError:
                pass
        else:
            # it will be nice if the error message prints a list of 
            # possible signatures here
            raise TypeError("No compatible signatures")

    def overload_with(func):
        overloaded_func.overloads.append(func)
        return overloaded_func

    overloaded_func.overloads = [func]
    overloaded_func.overload_with = overload_with
    return overloaded_func


#############

def consumer(func):
    """A decorator, advances func to its first yield point when called.

    Modifed this original example code from PEP 342 to use the new
    functools.wraps decorator. This convenience function makes it look
    like the original function, which is almost always what we want,
    especially if we designed the original function to be wrapped in
    the first place!

    Maybe `consumer` should go into functools too!
    """

    from functools import wraps

    @wraps(func)
    def wrapper(*args,**kw):
        gen = func(*args, **kw)
        gen.next()
        return gen
    return wrapper

def retry_with(handle_retry, exceptions, conditions, max_attempts):
    def wrapper(func):
        @functools.wraps(func)
        def decorated(*args, **kwargs):
            error = None
            result = None
            for attempt in range(1, max_attempts + 1):
                try:
                    result = func(*args, **kwargs)
                    if any(guard(result) for guard in conditions):
                        handle_retry.retry(func, args, kwargs, None, attempt)
                        continue
                    return result
                except Exception as err:
                    error = err
                    if any(isinstance(err, exc) for exc in exceptions):
                        handle_retry.retry(func, args, kwargs, err, attempt)
                        continue
                    break
            logger.info('failed after {} attempts'.format(attempt))
            if error is not None:
                raise error
            return result
        return decorated
    return wrapper

def restart_on_change(restart_map, stopstart=False, restart_functions=None):
    """Restart services based on configuration files changing

    This function is used a decorator, for example::

        @restart_on_change({
            '/etc/ceph/ceph.conf': [ 'cinder-api', 'cinder-volume' ]
            '/etc/apache/sites-enabled/*': [ 'apache2' ]
            })
        def config_changed():
            pass  # your code here

    In this example, the cinder-api and cinder-volume services
    would be restarted if /etc/ceph/ceph.conf is changed by the
    ceph_client_changed function. The apache2 service would be
    restarted if any file matching the pattern got changed, created
    or removed. Standard wildcards are supported, see documentation
    for the 'glob' module for more information.

    @param restart_map: {path_file_name: [service_name, ...]
    @param stopstart: DEFAULT false; whether to stop, start OR restart
    @param restart_functions: nonstandard functions to use to restart services
                              {svc: func, ...}
    @returns result from decorated function
    """
    def wrap(f):
        @functools.wraps(f)
        def wrapped_f(*args, **kwargs):
            return restart_on_change_helper(
                (lambda: f(*args, **kwargs)), restart_map, stopstart,
                restart_functions)
        return wrapped_f
    return wrap

def cached(func):
    """Cache return values for multiple executions of func + args

    For example::

        @cached
        def unit_get(attribute):
            pass

        unit_get('test')

    will cache the result of unit_get + 'test' for future calls.
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        global cache
        key = str((func, args, kwargs))
        try:
            return cache[key]
        except KeyError:
            pass  # Drop out of the exception handler scope.
        res = func(*args, **kwargs)
        cache[key] = res
        return res
    wrapper._wrapped = func
    return wrapper

def translate_exc(from_exc, to_exc):
    def inner_translate_exc1(f):
        @wraps(f)
        def inner_translate_exc2(*args, **kwargs):
            try:
                return f(*args, **kwargs)
            except from_exc:
                raise to_exc

        return inner_translate_exc2

    return inner_translate_exc1

def os_requires_version(ostack_release, pkg):
    """
    Decorator for hook to specify minimum supported release
    """
    def wrap(f):
        @wraps(f)
        def wrapped_f(*args):
            if os_release(pkg) < ostack_release:
                raise Exception("This hook is not supported on releases"
                                " before %s" % ostack_release)
            f(*args)
        return wrapped_f
    return wrap

def os_workload_status(configs, required_interfaces, charm_func=None):
    """
    Decorator to set workload status based on complete contexts
    """
    def wrap(f):
        @wraps(f)
        def wrapped_f(*args, **kwargs):
            # Run the original function first
            f(*args, **kwargs)
            # Set workload status now that contexts have been
            # acted on
            set_os_workload_status(configs, required_interfaces, charm_func)
        return wrapped_f
    return wrap

def admin_login_required(method):
    def is_admin(user):
        if isinstance(user.is_admin, bool):
            return user.is_admin
        else:
            return user.is_admin()

    @functools.wraps(method)
    def wrapper(*args, **kwargs):
        if not current_user.is_authenticated:
            flash("This section is for logged in users only.", 'warning')
            return redirect(url_for('redberry.home'))

        if not hasattr(current_user, 'is_admin'):
            flash("Redberry expects your user instance to implement an `is_admin` boolean attribute "
                  "or an `is_admin()` method.", 'warning')
            return redirect(url_for('redberry.home'))

        if not is_admin(current_user):
            flash("This section is for admin users only.", 'warning')
            return redirect(url_for('redberry.home'))

        return method(*args, **kwargs)

    return wrapper


############
# CMS ROUTES
############

def with_context_manager(ctx_manager):
    def decorator(fn):
        @wraps(fn)
        def context_manager_wrapper(*a, **kw):
            with ctx_manager:
                return fn(*a, **kw)
        context_manager_wrapper.djpt_patched = True
        return context_manager_wrapper

    return decorator

def lazy_property(func):
    attribute = '_lazy_' + func.__name__

    @property
    @functools.wraps(func)
    def wrapper(self):
        if not hasattr(self, attribute):
            setattr(self, attribute, func(self))
        return getattr(self, attribute)
    return wrapper


# ?bias_shape ? None??????bias

def parse_args(**k):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            pv = {}
            for key, val in k.iteritems():
                arg_type, default_val = val
                pv[key] = parse_single_arg(key, kwargs, arg_type, default_val)
            kwargs.update(pv)
            return func(*args, **kwargs)
        return wrapper
    return decorator