装饰器 这个名称可能更适合再编译器领域使用，因为它会遍历并注解语法树

—— PEP 318

装饰器

1. 概述

装饰器（Decorators） 用于源码 ’标记‘ 函数，以某种方式增强函数的行为，必须先掌握 以下三个条件

• 作用域
• 高阶函数
• 闭包 的方方面面

高阶函数

接受函数为参数，或者把函数作为结果返回的函数

1.1 基础知识

装饰器是可调用对象，其参数是另一个函数（被装饰的函数）。装饰器可能会处理被装饰的函数，然后把它返回，或者将其替换成另一个函数或可调用对象

假如有名为 decorate 的装饰器

@decirate
def target():
	print("running target()")

上述写法等价于

def target():
	print("running target()")

target = decorate(target)

• 上述代码执行后得到的 target 不一定是原来的 target 函数，而是 decorate(target) 返回的函数

为了确认装饰器函数会被替换

def deco(func)
	def inner():
        print("running inner()")
    return inner

@deco
def target()
	print("running target()")

target()
print(target)
# 故意不写打印，盲猜试试？

严格来说装饰器是 语法糖，可以像常规调函数传参形式调用，但尤其是 元编程 时，使用 @装饰器 的形式更方便

1.2 Python 何时执行装饰器

装饰器有两个特性

1. 能把被装饰的函数替换为其他函数
2. 装饰器在模块加载时立即执行，这个如何理解？如下

registry = []	# 保存的是被 @register 装饰的函数引用


def register(func):
    print(f"running register({func})")
    registry.append(func)
    return func


@register
def f1():
    print("running f1()")


@register
def f2():
    print("running f2()")


def f3():
    print("running f3()")


def main():
    print("running main()")
    print("registry -> ", registry)
    f1()
    f2()
    f3()


if __name__ == '__main__':
    main()

## 输出结果
# running register(<function f1 at 0x00000206AFD66E58>)
# running register(<function f2 at 0x00000206AFD66EE8>)
# running main()
# registry -> registry ->  [<function f1 at 0x000001B5924DB310>, <function f2 at 0x000001B592502EE0>]
# running f1()
# running f2()
# running f3()

• 会发现 register() 函数先执行并打印了 running register
• Python 加载模块时，会先将定义好的函数占块地址加载到内存，然后解释器顺序执行
• 看似直接执行了装饰器，但不是，而是在调用 @register 时，相当于执行了 func= register(func) ，解释器顺便执行了 register() 而已
• 这样直接接收函数，并 原样返回 的装饰器，并非没用。很多 Python Web 框架使用这种装饰器，把函数添加到某种中央注册处
  • 如：把 URL 模式映射到生成 HTTP 响应的函数上的注册处

1.3 函数的作用域

说闭包前先看一个 作用域 的栗子

def f():
    c = 6

f()
print(c)

这里能拿到 c 的值么? 当然不能，函数在 f() 调用后便在内存中销毁了，因此全局中拿不到局部的变量 c

b = 6

def f3(a):
    # global b
    print(a)
    print(b)
    b = 9

此时会报错，为啥 print(b) 时显示 b 尚未赋值报错？

• 由于 字节码 中，Python 定义了 b 是局部作用域，而并非全局作用域

• 解决办法可以再 print(b) 前添加 global b 指定字节码解析时 b 为全局作用域

  注

  这比 JavaScript 优秀，因为 JavaScript 内部未定义变量 b 就会自动全局 var b ，所以常会莫名其妙的拿到个全局作用域的变量，就很离谱

关于 字节码，可使用反汇编 dis

from dis import dis

dis(f3)

运行字节码的 Cpython VM 是栈机器，LOAD 和 POP 操作引用的是栈

1.4 闭包

通常会将闭包和匿名函数弄混，因为在 函数里写函数不常见，通常使用匿名函数才会这么做，且只有涉及到嵌套函数时才有闭包问题，很多人是同时知道这两个概念的

闭包，即 延伸了作用域 的函数，函数式实现一个计算平均值的 高阶函数

def make_averager():
    series = []

    def averager(new_value):
        series.append(new_value)
        total = sum(series)
        return total / len(series)

    return averager


if __name__ == '__main__':
    avg = make_averager()
    print(avg(10))
    print(avg(11))
    print(avg(12))

# 输出结果
10.0
10.5
11.0

会发现每次调用 avg() 函数，都会将上一次调用的值存储起来，如何做到的？

• 首先 series 是 make_averager() 的局部变量，按理说 make_averager() 返回后局部变量及作用域应该被销毁才对，但其实不然

  • 此时 series 是个 自由变量（free variable）：指在本地作用域中绑定的变量
  • 此时不会触发垃圾回收
  

• 在函数内部，对外部作用域（但不是全局作用域）的变量，进行 引用 的，就是闭包

提示

审查 make_averager() 创建的函数， 使用 __code__ （编译后的函数定义体）

print(avg.__code__.co_varnames)
print(avg.__code__.co_freevars)

# ('new_value', 'total')
# ('series',)

series 的绑定，在返回的 avg 函数的 __closure__ 属性中，各元素对应于 avg.__code__.co_freevars 中的一个名称，这些元素是 cell 对象，有 cell_contents 属性保存真正的值

print(avg.__closure__)
# (<cell at 0x000001B3A5EBBFD0: list object at 0x000001B3A5B84D80>,)

print(avg.__closure__[0].cell_contents)
# [10, 11, 12]

综上，闭包是种函数，会保留定义函数时，存在的自由变量的绑定，虽然作用域不可用，但 绑定依然可用

注意

大量使用闭包是很有风险的操作，尤其定义的还是 可变类型的局部变量，这常导致 内存泄露！

1.5 结合 nonlocal 声明

上文已经造成了内存泄露，此时为了优化，更好思路如下

def make_averager():
    count = 0
    total = 0

    def averager(new_value):
        count += 1
        total += new_value
        return total / count

    return averager

## 输出结果
# count += 1
# UnboundLocalError: local variable 'count' referenced before assignment

• 此时发现，解释器报错，局部变量 count 未定义前引用
• 这是由于 count += 1 本质是 count = count + 1 解释器先执行右边，再未声明 count 前就进行了 +1 操作
• 这时会有疑问，为啥上文 series 不会报错 ?
  • 是因为 count 是不可变对象，不能进行增删改操作，只可读
  • series 也没有进行赋值前引用操作，而是调用了 .append() 进行了增加操作，此时不会影响变量本身的地址

故可以配合 nonlocal 来改写，使其不用列表形式，保存所有的历史值

def make_averager():
    count = 0
    total = 0

    def averager(new_value):
        nonlocal count, total
        count += 1
        total += new_value
        return total / count

    return averager


if __name__ == '__main__':
    avg = make_averager()
    print(avg(10))
    print(avg(11))
    print(avg(12))

## 输出结果
10.0
10.5
11.0

此时 内存同样泄露，但泄露的起码比 list （列表）少太多，稍微变得 相对可控

2. 实现装饰器

2.1 理念

装饰器: 区别于《设计模式》中的装饰器模式，但其描述的：“动态的给一个对象添加一些额外的职责” 的理念是一致的，但在实现层面，与装饰器模式，毫无关系。它符合 开放封闭 及 AOP

• 开放封闭原则： 不修改内部函数的情况下，增加新功能，类似补丁一样

• AOP（面向切面式编程）： 即横向插入一段逻辑，可以减少大量重复代码

  • 常用场景：插入日志、性能测试、事务处理等

2.2 简单装饰器

实现一个简单装饰器：被装饰的函数可以每次调用后，将花费的 时间、传参、结果 都打印出来

import time


def clock(func):
    def clocked(*args):
        start_time = time.perf_counter()
        ret = func(*args)	# 闭包
        end_time = time.perf_counter() - start_time
        print(f"[{end_time:.8f}μs], {func.__name__}({args}) -> {ret}")
        return ret

    return clocked


@clock
def snooze(seconds):
    time.sleep(seconds)


@clock
def factorial(n):
    return 1 if n < 2 else n * factorial(n - 1)


if __name__ == '__main__':
    snooze(.123)
    factorial(6)
    print(factorial.__name__)

## 输出结果
[0.11668170μs], snooze((0.123,)) -> None
[0.00000230μs], factorial((1,)) -> 1
[0.00008320μs], factorial((2,)) -> 2
[0.00013500μs], factorial((3,)) -> 6
[0.00018600μs], factorial((4,)) -> 24
[0.00022760μs], factorial((5,)) -> 120
[0.00028440μs], factorial((6,)) -> 720
clocked

• 查看__name__ 属性，会发现输出 clocked，这是由于被装饰后 factorial 保存了 clocked 函数的引用

• 装饰器的典型行为就是 把被装饰的函数替换成新函数，且二者接收到了相同的参数，再做些额外操作，并返回 加工|未加工 的值

如上存在许多缺点，故需要改写

1. 属性 __name__ 不是想要的值
2. 不支持关键字参数
3. 其实还遮盖了 __doc__ 属性

import time
import functools


def clock(func):
    @functools.wraps(func)
    def clocked(*args, **kwargs):
        start_time = time.perf_counter()
        ret = func(*args, **kwargs)
        end_time = time.perf_counter() - start_time
        arg_lst = []
        if args:
            arg_lst.append(",".join(repr(arg) for arg in args))
        if kwargs:
            pairs = [f"{k}={v}" for k, v in sorted(kwargs.items())]
            arg_lst.append(", ".join(pairs))
        arg_str = ", ".join(arg_lst)
        print(f"[{end_time:.8f}μs], {func.__name__}({arg_str}) -> {ret}")
        return ret

    return clocked

• @functools.wraps(func) 详见 标准库，用来还原被装饰器覆盖的 __name__ 和 __doc__ 等属性

2.3 标准库中的装饰器

Python 内置了三个用于装饰器方法的函数，property、classmethod 和 staticmethod， 都是和面向对象相关，其他还有 lru_cache 做轻量缓存加速、singledispatch 做单分派泛函数等

2.4 叠放装饰器

装饰器可以叠放，如下

@d1
@d2
def f():
	pass

• 这种将 @d1 和 @d2 两个装饰器顺序叠放到函数 f() 上，等同于如下

def f():
	pass

f = d1(d2(f))

执行顺序是 由外层向内层 顺序执行

2.5 参数化装饰器

装饰器可以接收被装饰的函数和他的参数，那么如何让装饰器自身来接收额外的参数呢？答案是创建一个 装饰器工厂函数，这个工厂的目的是，一调用这个工厂就能 返回 一个真正的 装饰器

registry = set()


def register(active=True):
    def decorate(func):
        print(f"running register {active} -> decorate({func})")
        if active:
            registry.add(func)
        else:
            registry.discard(func)
        return func

    return decorate


@register(False)
def f1():
    print("running f1()")


@register()
def f2():
    print("running f2()")


def f3():
    print("running f3()")


if __name__ == '__main__':
    print("running main()")
    print(f"registry -> {registry}")

提示

装饰 fx() 的 句法 等同于： f1 = register(active=False)(f)

参数化且万能参数的装饰工厂

import functools

DEFAULT_FMT = "[{elapsed:.8f}s] {name} ({_args}) -> {result}"


def clock(fmt=DEFAULT_FMT):
    def decorate(func):
        @functools.wraps(func)
        def clocked(*args, **kwargs):
            start_time = time.time()
            name = func.__name__
            _args = ",".join(str(i) for i in args)
            _kwargs = ",".join(f" {k}={v}" for k, v in kwargs.items())
            result = func(*args, **kwargs)
            elapsed = time.time() - start_time
            print(fmt.format(**locals()))
            return result

        return clocked

    return decorate


CUSTOM_FMT = "[{elapsed:.3f}s] {name}({_args},{_kwargs}): -> {result}"


@clock(CUSTOM_FMT)
def snooze(seconds, a, b):
    time.sleep(seconds)
    return f"sleep: {seconds}s {a + b}"


if __name__ == '__main__':
    ret = snooze(1, a=1, b=2)
    print(ret, snooze.__name__)
    for i in range(3):
        snooze(.123, a=1, b=i)

提示

装饰工厂句法：snooze(1, a=1, b=2) = clock(fmt=DEFAULT_FMT)(snooze)(*args, **kwargs)

装饰器最好通过实现 __call__ 方法的类实现，使用函数的语言特性更容易理解

参考

3. 类装饰器

根据 类的特性 封装的装饰器

import time


class Clock:
    def __init__(self, *args, **kwargs):
        self.fmt = kwargs.get("fmt", None)

    def __call__(self, func):

        def inner(*args, **kwargs):
            start_time = time.time()
            name = func.__name__
            _args = ",".join(str(i) for i in args)
            _kwargs = ",".join(f" {k}={v}" for k, v in kwargs.items())
            result = func(*args, **kwargs)
            elapsed = time.time() - start_time
            if self.fmt:
                print(self.fmt.format(**locals()))
            return result

        return inner


CUSTOM_FMT = "[{elapsed:.3f}s] {name}({_args},{_kwargs}): -> {result}"


@Clock(fmt=CUSTOM_FMT)
def foo(a, b):
    return a + b


foo(1, b=2)

当类实例化为 Clock 对象时，即 Clock() 会触发类的 __init__() 和 __call__() 此时意味着装饰器已调用，句法为 foo(1, b=2) = Clock(fmt)(foo)(1, b=2)

• 其中 __init__() ，会实例化装饰器的参数
• __call__() ，会调用被装饰的 foo()

4. 异步装饰器

一个 xmlrpc 的例子，如何通过装饰器，实现将同步的客户端请求改为异步，避免阻塞

server端：

import datetime
import sys
from xmlrpc.server import SimpleXMLRPCServer


class ExampleService:
    def getData(self):
        return '42'

    class currentTime:
        @staticmethod
        def getCurrentTime():
            print(datetime.datetime.now())
            return datetime.datetime.now()


if __name__ == '__main__':

    with SimpleXMLRPCServer(("localhost", 9000)) as server:
        server.register_instance(ExampleService(), allow_dotted_names=True)
        server.register_multicall_functions()
        print('Serving XML-RPC on localhost port 8000')
        print('It is advisable to run this example server within a secure, closed network.')
        try:
            server.serve_forever()
        except KeyboardInterrupt:
            print("\nKeyboard interrupt received, exiting.")
            sys.exit(0)

client端： 通过装饰器改为异步

import asyncio
from functools import wraps
from xmlrpc import client


def async_executor(func):
    @wraps(func)
    async def inner(self):
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, func, self)

    return inner


class RpcProxy:
    def __init__(self, host: str = "localhost", port: int = 9000):
        self.proxy = client.ServerProxy(f"http://{host}:{port}")  # noqa


class ClientRpc(RpcProxy):
    def __init__(self, host: str = "localhost", port: int = 9000):
        super(ClientRpc, self).__init__(host, port)

    @async_executor
    def get_server_time(self):  # 若此处加上async 上文method
        """ 同步改异步 """
        result = self.proxy.currentTime.getCurrentTime()
        print(f"server time: {result}")
        return result


async def main():
    for i in range(10):
        # ClientRpc().get_server_time()
        asyncio.create_task(ClientRpc().get_server_time())
    print("running main()")


if __name__ == '__main__':
    asyncio.run(main())

如上，通过装饰器，将同步函数包装，改写为异步形式