Python装饰器

纸上得来终觉浅，绝知此事要躬行。

---

1. 装饰器的基础知识

在Java语言中装饰器是一种设计模式，而Python则原生就支持这样使用方式。

装饰器的引入

• 面向切面的编程范式就是在运行时，动态地将代码切入到类的指定方法、指定位置上的编程思想就是面向切面的编程，更通俗一点就是通过在现有代码中添加额外行为而不修改代码本身。
• 装饰器就是通过这样的面向切面的编程的思路进行设计的，不改变现有代码的前提下，对其功能、内容进行扩展和补充。

def use_logging(func):
    def wrapper():
        logging.warn("%s is running" % func.__name__)
        return func()
    return wrapper

装饰器的好处

• 降低模块的耦合度
• 使系统容易扩展
• 更好的代码复用性

装饰器的多继承

• 一个函数或者类拥有多个装饰器，他们的执行顺序从上到下依次包裹

# 一个函数还可以同时定义多个装饰器
@a
@b
@c
def f ():
    pass

# 执行顺序是从里到外，最先调用最里层的装饰器，最后调用最外层的装饰器
f = a(b(c(f)))

使用 wraps 方法

• 官方文档内容

help(wraps)

wraps(wrapped, assigned=('__module__', '__name__', '__qualname__', '__doc__', '__annotations__'), updated=('__dict__',))
    Decorator factory to apply update_wrapper() to a wrapper function

    Returns a decorator that invokes update_wrapper() with the decorated
    function as the wrapper argument and the arguments to wraps() as the
    remaining arguments. Default arguments are as for update_wrapper().
    This is a convenience function to simplify applying partial() to
    update_wrapper().

参考链接地址

• PythonDecoratorLibrary
• attrs 和 Python3.7 的 dataclasses
• decorator for asynchronous code in Python

---

2. 装饰器的使用技巧

2.1 函数装饰器

用函数实现的装饰器，不带参数的装饰器一层嵌套，带参数的装饰器二次嵌套。

不使用装饰器写法

• 不带参数的装饰器 k 就是再其外层进行一层封装嵌套，这里只不过没有使用@装饰器的语法糖而已。

from datetime import datetime

def time_info(fn):
    def wrapper(*args):
        print(datetime.now())
        res = fn(*args)
        return res
    return wrapper

def warn_log(messages):
    print('[Warn]: {0}'.format(messages))

In [1]: from func import warn_log, time_info

In [2]: warn = time_info(warn_log)

In [3]: warn('start write log info...')
2018-07-06 15:23:35.988196
[Warn]: start write log info...

不带参数的装饰器

• @符号是装饰器的语法糖，语法糖指计算机语言中添加的某种语法，这种语法对语言的功能没有影响，但是更方便程序员使用。语法糖让程序更加简洁，有更高的可读性。

from functions import wraps
from datetime import datetime

def time_info(fn):
    @wraps(fn)
    def wrapper(*args):
        print(datetime.now())
        res = fn(*args)
        return res
    return wrapper

@time_info
def warn_log(messages):
    print('[Warn]: {0}'.format(messages))

In [1]: from func import warn_log

# 类似于warn = time_info(warn_log)的使用方式
In [2]: warn = warn_log('start write log info...')
2018-07-06 15:27:59.234672
[Warn]: start write log info...

带参数的装饰器

• 带参数的装饰器写法就是再其外层进行两层封装嵌套，最外层为接收自生传递进来的函数，而第二层就是接收需要装饰的函数了。

from functions import wraps

def check_users(*args):
    allow_users = list(args)
    def wrapper(fn):
        @wraps(fn)
        def _(*args):
            print('Check user with input...')
            if list(args)[0] in allow_users:
                print('Pass to get password.')
                res = fn(*args)
                return res
            else:
                raise ValueError('Input check user is not allow.')
        return _
    return wrapper

@check_users('escape', 'misssun')
def get_password(*args):
    print('Password is 123456.')

In [1]: from func import get_password

In [2]: get_password('escape')
Check user with input...
Pass to get password.
Password is 123456.

In [3]: get_password('newuser')
Check user with input...
----------------------------------------------------------
ValueError               Traceback (most recent call last)
<ipython-input-25-f144bb0f76f6> in <module>()
----> 1 get_password('newuser')

~/Escape/MorePractise/func.py in _(*args)
     10                 return res
     11             else:
---> 12                 raise ValueError('Input check user is not allow.')
     13         return _
     14     return wrapper

ValueError: Input check user is not allow.

functools.wraps

• 用于保持调用时属性一致的情况

In [1]: def warn_log():
  ...:     '''warn_log doc'''
  ...:     print('[Warn]: messages')
  ...:     return 1
  ...:

In [2]: warn_log.__name__
Out[2]: 'warn_log'

In [3]: warn_log.__module__
Out[3]: '__main__'

In [4]: warn_log.__doc__
Out[4]: 'warn_log doc'

In [5]: warn = time_info(warn_log)

In [6]: warn.__name__
Out[6]: 'wrapper'

In [7]: warn.__module__
Out[7]: '__main__'

In [8]: warn.__doc__

In [1]: def warn_log():
  ...:     '''warn_log doc'''
  ...:     print('[Warn]: messages')
  ...:     return 1
  ...:

In [2]: def time_info(fn):
...:     @wraps(fn)
...:     def wrapper():
...:         res = fn()
...:         print(datetime.now())
...:         return res
...:     return wrapper
...:

In [3]: warn = time_info(warn_log)

In [4]: warn.__name__
Out[4]: 'warn'

In [5]: warn.__doc__
Out[5]: 'warn_log doc'

---

2.2 类装饰器

用类实现的装饰器，主要是因为__call__的重载。

给函数的类装饰器

• 有两种表示方式，还分带参数和不带参数

# 不带参数的类装饰器
In [1]: class Common:
   ...:     def __init__(self, func):
   ...:         self.func = func
   ...:     def __call__(self, *args, **kwargs):
   ...:         print(f'args: {args}')
   ...:         return self.func(*args, **kwargs)
   ...:

In [2]: @Common
   ...: def test(num):
   ...:     print(f'Number: {num}')
   ...:

# 等价于Common(test)(10)格式
In [3]: test(10)
args: (10,)
Number: 10

# 这个是实现同样功能的函数装饰器
In [4]: def common(func):
   ...:     def wrapper(*args, **kwargs):
   ...:         print(f'args: {args}')
   ...:         return func(*args, **kwargs)
   ...:     return wrapper
   ...:

给类用的函数装饰器

• borg是一个设计模式，它保证了同一个类的实例属性的一致性问题。比较相近的是单例模式，它实现了全局只有一个实例的限制。

In [1]: def borg(cls):
   ...:     cls._state = {}
   ...:     orig_init = cls.__init__
   ...:     def new_init(self, *args, **kwargs):
   ...:         self.__dict__ = cls._state
   ...:         orig_init(self, *args, **kwargs)
   ...:     cls.__init__ = new_init
   ...:     return cls
   ...:

In [2]: @borg
   ...: class A:
   ...:     def common(self):
   ...:         print(hex(id(self)))...:

In [3]: a, b = A(), A()

In [4]: b.d
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-104-1dbeb93aa9bb> in <module>()
----> 1 b.d
AttributeError: 'A' object has no attribute 'd'

In [5]: b.d = 1

In [6]: a.d
Out[6]: 1

In [7]: a.common()
0x104f0c198

# Python3.7中引入的新模块attr，其中大量使用到了给类用的类装饰器

In [1]: import attr

In [2]: @attr.s(hash=True)
   ...: class Product(object):
   ...:     id = attr.ib()
   ...:     author_id = attr.ib()
   ...:

---

3. 装饰器的应用场景

主要应用场景

• 记录函数行为 (日志统计、缓存、计时)
• 预处理/后处理 (配置上下文、参数字段检查、统一返回格式)
• 注入/移除参数
• 修改调用时的上下文 (实现异步或者并行)

3.1 路由装饰

from flask import Flask
app = Flask(__name__)

@app.route("/")
def hello():
    return "Hello World!"

if __name__ == "__main__":
    app.run()

---

3.2 权限控制

In [1]: def check(allows):
   ...:     def deco(fn):
   ...:         def wrap(username, *args, **kwargs):
   ...:             if username in allows:
   ...:                 return fn(username, *args, **kwargs)
   ...:             return "not allow"
   ...:         return wrap
   ...:     return deco
   ...:

In [2]: @check(['escape','misssun'])
   ...: def private(username):
   ...:     print('Hi,boy')
   ...:

In [3]: private('escape')
Hi,boy

In [4]: private('mom')
Out[4]: 'not allow'

---

3.3 打点监控

In [9]: def timeit(process_time=False):
   ....:     if process_time:
   ....:         cacl = time.clock
   ....:     else:
   ....:         cacl = time.time
   ....:     def inner_timeit(fn):
   ....:         def wrap(*args, **kwargs):
   ....:             start = cacl()
   ....:             ret = fn(*args, **kwargs)
   ....:             print(cacl() - start)
   ....:             # send to monitor system
   ....:             return ret
   ....:         return wrap
   ....:     return inner_timeit

---

3.4 缓存容器

import time
from functools import wraps


class DictCache(object):
    def __init__(self):
        self.cache = dict()

    def get(self, key):
        return self.cache.get(key)

    def set(self, key, value):
        self.cache[key] = value

    def __str__(self):
        return str(self.cache)

    def __repr__(self):
        return repr(self.cache)


def cache(instance):
    def dec(fn):
        @wraps(fn)
        def wrap(*args, **kwargs):
            pos = ','.join((str(x) for x in args))
            kw = ','.join('{}={}'.format(k, v) for k,v in sorted(kwargs.items()))
            key = '{}::{}::{}'.format(fn.__name__, pos, kw)
            ret = instance.get(key)
            if ret is not None:
                return ret
            ret = fn(*args, **kwargs)
            instance.set(key, ret)
            return ret
        return wrap
    return dec


cache_instance = DictCache()
@cache(cache_instance)
def long_time_fun(n):
    time.sleep(n)
    return n

print('=' * 20)
print(long_time_fun(3))
print(cache_instance)
print('=' * 20)
print(long_time_fun(3))
print('=' * 20)

---

3.5 异步并发

from functools import wraps
from concurrent.futures import ThreadPoolExecutor


class Tomorrow():
    def __init__(self, future, timeout):
        self._future = future
        self._timeout = timeout

    def __getattr__(self, name):
        result = self._wait()
        return result.__getattribute__(name)

    def _wait(self):
        return self._future.result(self._timeout)


def async(n, base_type, timeout=None):
    def decorator(f):
        if isinstance(n, int):
            pool = base_type(n)
        elif isinstance(n, base_type):
            pool = n
        else:
            raise TypeError(
                "Invalid type: %s" % type(base_type)
            )
        @wraps(f)
        def wrapped(*args, **kwargs):
            return Tomorrow(
                pool.submit(f, *args, **kwargs),
                timeout = timeout
            )
        return wrapped
    return decorator


def threads(n, timeout=None):
    return async(n, ThreadPoolExecutor, timeout)