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().
参考链接地址
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)
