python装饰器参考

装饰器的两种实现方式

# 利用闭包实现
def function_wrapper(wrapped):
        @functools.wraps(wrapped)
        def _wrapper(*args, **kwargs):
            return wrapped(*args, **kwargs)
        return _wrapper

    @function_wrapper
    def function():
        pass
        
# 类实现
class function_wrapper(object):
        def __init__(self, wrapped):
            self.wrapped = wrapped
        def __call__(self, *args, **kwargs):
            return self.wrapped(*args, **kwargs)

    @function_wrapper
    def function():
        pass
        
# @ 带参数的情况
def add_paras2_kws(*wargs,**wkws):
    print ("wargs:{}".format(wargs))
    print ("wkws:{}".format(wkws))
    def warp_f(f):
        @wraps(f)
        def dec(*fargs, **fkws):
            print(f)
            wkws.update(**fkws)
            print ("fargs:{}".format(fargs))
            print ("fkws:{}".format(fkws))
            return f(*fargs, **wkws)
        return dec
    return warp_f

示例一，retry

def __retry_internal(f, exceptions=Exception, tries=-1, delay=0, max_delay=None, backoff=1, jitter=0,
                     logger=logging_logger):
    """
    Executes a function and retries it if it failed.

    :param f: the function to execute.
    :param exceptions: an exception or a tuple of exceptions to catch. default: Exception.
    :param tries: the maximum number of attempts. default: -1 (infinite).
    :param delay: initial delay between attempts. default: 0.
    :param max_delay: the maximum value of delay. default: None (no limit).
    :param backoff: multiplier applied to delay between attempts. default: 1 (no backoff).
    :param jitter: extra seconds added to delay between attempts. default: 0.
                   fixed if a number, random if a range tuple (min, max)
    :param logger: logger.warning(fmt, error, delay) will be called on failed attempts.
                   default: retry.logging_logger. if None, logging is disabled.
    :returns: the result of the f function.
    """
    _tries, _delay = tries, delay
    while _tries:
        try:
            return f()
        except exceptions as e:
            _tries -= 1
            if not _tries:
                raise

            if logger is not None:
                logger.warning('%s, retrying in %s seconds...', e, _delay)

            time.sleep(_delay)
            _delay *= backoff

            if isinstance(jitter, tuple):
                _delay += random.uniform(*jitter)
            else:
                _delay += jitter

            if max_delay is not None:
                _delay = min(_delay, max_delay)


def retry(exceptions=Exception, tries=-1, delay=0, max_delay=None, backoff=1, jitter=0, logger=logging_logger):
    """Returns a retry decorator.

    :param exceptions: an exception or a tuple of exceptions to catch. default: Exception.
    :param tries: the maximum number of attempts. default: -1 (infinite).
    :param delay: initial delay between attempts. default: 0.
    :param max_delay: the maximum value of delay. default: None (no limit).
    :param backoff: multiplier applied to delay between attempts. default: 1 (no backoff).
    :param jitter: extra seconds added to delay between attempts. default: 0.
                   fixed if a number, random if a range tuple (min, max)
    :param logger: logger.warning(fmt, error, delay) will be called on failed attempts.
                   default: retry.logging_logger. if None, logging is disabled.
    :returns: a retry decorator.
    """

    @decorator
    def retry_decorator(f, *fargs, **fkwargs):
        args = fargs if fargs else list()
        kwargs = fkwargs if fkwargs else dict()
        return __retry_internal(partial(f, *args, **kwargs), exceptions, tries, delay, max_delay, backoff, jitter,
                                logger)
    return retry_decorator

comments:

• __retry_internal 调用前先固定住f的参数，简化问题。
• __retry_internal 要处理的只有f的调用执行。

示例二

from functools import wraps

def return_kw(**kw):
    return kw

def add_paras2_kws(*wargs,**wkws):
    print ("wargs:{}".format(wargs))
    print ("wkws:{}".format(wkws))
    def warp_f(f):
        @wraps(f)
        def dec(*fargs, **fkws):
            print(f)
            wkws.update(**fkws)
            print ("fargs:{}".format(fargs))
            print ("fkws:{}".format(fkws))
            return f(*fargs, **wkws)
        return dec
    return warp_f

@add_paras2_kws(7,8,9,a=80,b=81,c=82,d=83)
def test_b(*args,**kw):
    return return_kw(**kw)

if __name__ == '__main__':
    print(test_b(1,2,3,a=4,b=5))

comments：

• 要在常规的装饰器外再套一层接收参数，因为是先传递@的参数，接着传递被装饰的函数，再传递被装饰函数的参数。