python的装饰器(Decorators)定义：具有执行环境的函数，其满足三个条件:

  1. 外部函数中定义一个内部函数
  2. 内部函数中使用外部函数的局部变量
  3. 外部函数将内部函数作为返回值返回

此时的内部函数就叫闭包(Closure), 装饰器允许我们包装内部的函数以扩展函数的行为，而不必对内部函数修改。

  • 装饰器函数（使用装饰器和取消装饰器）

    例子：输出函数执行时间的装饰器。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    def record_time(func):
        """自定义装饰函数的装饰器"""
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time()
            result = func(*args, **kwargs)
            print(f'{func.__name__}: {time() - start}秒')
            return result
            
        return wrapper

    如果装饰器不希望跟print函数耦合，可以编写带参数的装饰器。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    from functools import wraps
    from time import time


    def record(output):
        """自定义带参数的装饰器"""
    	
    	def decorate(func):
    		
    		@wraps(func)
    		def wrapper(*args, **kwargs):
    			start = time()
    			result = func(*args, **kwargs)
    			output(func.__name__, time() - start)
    			return result
                
    		return wrapper
    	
    	return decorate
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    from functools import wraps
    from time import time


    class Record():
        """自定义装饰器类(通过__call__魔术方法使得对象可以当成函数调用)"""

        def __init__(self, output):
            self.output = output

        def __call__(self, func):

            @wraps(func)
            def wrapper(*args, **kwargs):
                start = time()
                result = func(*args, **kwargs)
                self.output(func.__name__, time() - start)
                return result

            return wrapper

    说明：由于对带装饰功能的函数添加了@wraps装饰器，可以通过func.__wrapped__方式获得被装饰之前的函数或类来取消装饰器的作用。

    例子：用装饰器来实现单例模式。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    from functools import wraps


    def singleton(cls):
        """装饰类的装饰器"""
        instances = {}

        @wraps(cls)
        def wrapper(*args, **kwargs):
            if cls not in instances:
                instances[cls] = cls(*args, **kwargs)
            return instances[cls]

        return wrapper


    @singleton
    class President():
        """总统(单例类)"""
        pass

    说明：上面的代码中用到了闭包（closure），不知道你是否已经意识到了。还没有一个小问题就是，上面的代码并没有实现线程安全的单例，如果要实现线程安全的单例应该怎么做呢？

提示：代码用到了with上下文语法来进行锁操作，因为锁对象本身就是上下文管理器对象（支持__enter__和__exit__魔术方法）。在wrapper函数中，我们先做了一次不带锁的检查，然后再做带锁的检查，这样做比直接加锁检查性能要更好，如果对象已经创建就没有必须再去加锁而是直接返回该对象就可以了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from functools import wraps
from threading import Lock


def singleton(cls):
    """线程安全的单例装饰器"""
    instances = {}
    locker = Lock()

    @wraps(cls)
    def wrapper(*args, **kwargs):
        if cls not in instances:
            with locker:
                if cls not in instances:
                    instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return wrapper

fgd

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@singleton
class President():

    def __init__(self, name, country):
        self.name = name
        self.country = country

    def __str__(self):
        return f'{self.country}: {self.name}'


def main():
    print(President.__name__)
    p1 = President('特朗普', '美国')
    p2 = President('奥巴马', '美国')
    print(p1 == p2)
    print(p1)
    print(p2)

Closure

1
2
3
4
5
6
7
8
9
10
11
def hello():
    def world(): 
        print("Hello World!")
    return world

hello
# <function __main__.hello()>
hello()
# <function __main__.hello.<locals>.world()>
hello()()
# Hello World!
1
2
3
4
5
6
7
8
9
10
11
12
13
def hello(msg1):
    def world(msg2): 
        print(msg1 +" " + msg2 + "!")
    return world

hello
# <function __main__.hello(msg1)>
hello("Hello")
f = hello("Hello")
# <function __main__.hello.<locals>.world(msg2)>
hello("Hello")("world")
f("world")
# Hello world!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def hello():
    fs = []
    for i in range(1, 4):
        def world():
             return "world \t" * i + "\n"
        fs.append(world)
    return fs

f1, f2, f3 = hello() 
print(f1(),f2(),f3(), sep="")
# world 	world 	world 	
# world 	world 	world 	
# world 	world 	world 


def hello():
    def world(i):
        def temp():
             return "world \t" * i + "\n"
        return temp
    fs = []
    for i in range(1, 4):
        fs.append(world(i)) # 立刻被执行，因此i的当前值被传入world()
    return fs

def hello():
    fs = []
    for i in range(1, 4):
        def world(j = i):
            return  "world \t" * j + "\n"

        fs.append(world)
    return fs

def hello():
    fs = []
    for i in range(1,4):
        def world(x):
            return lambda: "world \t" * x + "\n"
        fs.append(world(i))
    return fs

f1, f2, f3 = hello() 
print(f1(),f2(),f3(), sep="")


# world  	
# world  world  	
# world  world  world

Decorators

从最简单的helloword程序说起, 我们先定义一个hello world函数，和另一个叫flower的函数，再把这个hello world函数传给那个flower函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def hello(): 
    print("Hello World!") 
      
hello
# <function __main__.hello()>

def flower(func):
    print("✿")

flower(hello)
# ✿
flower(hello())
# Hello World!
# ✿
1
2
3
4
5
6
7
8
9
10
11
12
13
14
def hello(): 
    print("Hello World!") 
      
def flower(func):
    print("✿")
    func()

flower(hello) 
# ✿
# Hello World!
flower(hello())
# Hello World!
# ✿
# 'NoneType' object is not callable

我们传入一个函数，

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def flower(func):
    print("✿")
    if func is None:
        print("'NoneType' object is not callable")
    else:
        func()
        
#== This was equivalent to hello = flower(hello) 
@flower
def hello(): 
    print("Hello World!") 

flower(hello)
# ✿
# Hello World!
# ✿
# 'NoneType' object is not callable
1
2
3
4
5
6
7
8
9
10
11
12
13
14
def flower(func):
    print("✿")
    def wrapper(*args, **kw):
#         print('call %s():' % func.__name__)
        return func(*args, **kw)
    return wrapper
    
@flower
def hello(): 
    print("Hello World!") 

hello()
# ✿
# Hello World!

我们再来看一个复杂的，

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def flower(message):
    def fl(func):
        half = round(int(len(message))/2)
        wrapper1 = "❀" * half  + func.__name__ + "❀" * half
        wrapper2 = "✿" * half  + func.__name__ + "✿" * half
        def wrap(*args, **kw):
            print(wrapper1)
            func(message.rjust(half * 2 + len( func.__name__)))
            print(wrapper2)
        return wrap
    return fl

@flower("Hello World!")
def hello(msg): 
    print(msg)

flower(hello)
# <function __main__.flower.<locals>.fl(func)>
flower(hello())
# ❀❀❀❀❀❀hello❀❀❀❀❀❀
#      Hello World!
# ✿✿✿✿✿✿hello✿✿✿✿✿✿
# <function __main__.flower.<locals>.fl(func)>

hello()
# ❀❀❀❀❀❀hello❀❀❀❀❀❀
#      Hello World!
# ✿✿✿✿✿✿hello✿✿✿✿✿✿
print(hello.__name__)
# wrap
flower("Hello World!")(hello)()
# ❀❀❀❀❀❀wrap❀❀❀❀❀❀
# ❀❀❀❀❀❀hello❀❀❀❀❀❀
#      Hello World!
# ✿✿✿✿✿✿hello✿✿✿✿✿✿
# ✿✿✿✿✿✿wrap✿✿✿✿✿✿

如果多个函数被两个装饰器装饰时就报错,因为两个函数名一样,第二个函数再去装饰的话就出错, 一般的话，引入@functools.wraps(func)可避免此问题。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import functools

def flower(message):
    def fl(func):
        half = round(int(len(message))/2)
        wrapper1 = "❀" * half  + func.__name__ + "❀" * half
        wrapper2 = "✿" * half  + func.__name__ + "✿" * half
        @functools.wraps(func)   # wrapper.__name__ = func.__name__
        def wrap(*args, **kw):
            print(wrapper1)
            func(message.rjust(half * 2 + len( func.__name__)))
            print(wrapper2)
        return wrap
    return fl

@flower("Hello World!")
def hello(msg): 
    print(msg)

print(hello.__name__)
# hello
flower("Hello World!")(hello)()
# ❀❀❀❀❀❀hello❀❀❀❀❀❀
# ❀❀❀❀❀❀hello❀❀❀❀❀❀
#      Hello World!
# ✿✿✿✿✿✿hello✿✿✿✿✿✿
# ✿✿✿✿✿✿hello✿✿✿✿✿✿

Functools

  • functool.update_wrapper：将被包装函数的__name__等属性，拷贝到新的函数中去。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    from functools import update_wrapper
    def wrap2(func):
        def inner(*args):
            return func(*args)
        return update_wrapper(inner, func)

    @wrap2
    def demo():
        print('hello world')

    demo()
    # hello world
    print(demo.__name__)
    # demo

  • functool.wraps: 在update_wrapper上进行一个包装, 拷贝被装饰函数的__name__。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    from functools import wraps
    def wrap2(func):
        @wraps(func)	# 去掉就会返回inner
        def inner(*args):
            print(func.__name__)
            return func(*args)
        return inner

    @wrap2
    def demo():
        print('hello world')

    demo()
    # demo
    # hello world
    print(demo.__name__)
    # demo

  • functools.lru_cache：将一个函数的返回值快速地缓存或取消缓存。

    该装饰器会将不同的调用结果缓存在内存中，因此需要注意内存占用问题。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    from functools import lru_cache
    @lru_cache(maxsize=30)  # maxsize参数告诉lru_cache缓存最近多少个返回值
    def fib(n):
        if n < 2:
            return n
        return fib(n-1) + fib(n-2)
    print([fib(n) for n in range(10)])
    # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
    fib.cache_clear()   # 清空缓存

  • functools.singledispatch: 用于实现泛型函数。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    from functools import singledispatch
    from decimal import Decimal

    @singledispatch
    def fun(arg, verbose=False):
        if verbose:
            print("Let me just say,", end=" ")
        print(arg)

    @fun.register(int)
    def _(arg, verbose=False):
        if verbose:
            print("Strength in numbers, eh?", end=" ")
        print(arg)


    @fun.register(list)
    def _(arg, verbose=False):
        if verbose:
            print("Enumerate this:")
        for i, elem in enumerate(arg):
            print(i, elem)

    @fun.register(float)
    @fun.register(Decimal)
    def fun_num(arg, verbose=False):
        if verbose:
            print("Half of your number:", end=" ")
        print(arg / 2)

    fun('hello world!', True)
    # Let me just say, hello world!
    fun(123, True)
    # Strength in numbers, eh? 123
    fun(['a','b','c'], True)
    # Enumerate this:
    # 0 a
    # 1 b
    # 2 c
    fun(1.23, True)
    # Half of your number: 0.615
    print(fun.registry)	# 所有的泛型函数
    # {<class 'object'>: <function fun at 0x0000000005069F78>, <class 'int'>: <function _ at 0x0000000005098558>, \
    # <class 'list'>: <function _ at 0x0000000005098678>, <class 'decimal.Decimal'>: <function fun_num at 0x0000000005098828>,\
    # <class 'float'>: <function fun_num at 0x0000000005098828>}
    print(fun.registry[float])	# 获取float的泛型函数
    # <function fun_num at 0x0000000005098E58>

fedf

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
"""
装饰器 - 装饰器中放置的通常都是横切关注（cross-concern）功能
所谓横切关注功能就是很多地方都会用到但跟正常业务又逻辑没有必然联系的功能
装饰器实际上是实现了设计模式中的代理模式 - AOP（面向切面编程）
"""
from functools import wraps
from random import randint
from time import time, sleep

import pymysql


def record(output):

    def decorate(func):

        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time()
            ret_value = func(*args, **kwargs)
            output(func.__name__, time() - start)
            return ret_value

        return wrapper

    return decorate


def output_to_console(fname, duration):
    print('%s: %.3f秒' % (fname, duration))


def output_to_file(fname, duration):
    with open('log.txt', 'a') as file_stream:
        file_stream.write('%s: %.3f秒\n' % (fname, duration))


def output_to_db(fname, duration):
    con = pymysql.connect(host='localhost', port=3306,
                          database='test', charset='utf8',
                          user='root', password='123456',
                          autocommit=True)
    try:
        with con.cursor() as cursor:
            cursor.execute('insert into tb_record values (default, %s, %s)',
                           (fname, '%.3f' % duration))
    finally:
        con.close()


@record(output_to_console)
def random_delay(min, max):
    sleep(randint(min, max))


def main():
    for _ in range(3):
        # print(random_delay.__name__)
        random_delay(3, 5)
    # for _ in range(3):
    #     # 取消掉装饰器
    #     random_delay.__wrapped__(3, 5)


if __name__ == '__main__':
    main()