装饰器

Posted April 14, 2021 by clannadzsy ‐ 2 min read

运行环境：以下独立代码块在 CPython 3.12.14 验证；除定时任务一节外只使用标准库。将某一完整代码块保存为 demo.py，用 python demo.py 运行，对照末尾的输出注释。

普通装饰器的第一个例子只打印日志；第二个例子通过 return ret 保留被装饰函数的返回值。实际封装时还可用 functools.wraps 保留函数名称、文档等元数据。

装饰器

普通装饰器

def logger(func):
    def wrapper(*args, **kwargs):
        func_name = func.__name__
        print(f'执行了 {func_name} 函数')
        func(*args, **kwargs)
        print(f'{func_name} 函数执行完成')

    return wrapper


@logger
def add(x, y):
    print(f'{x} + {y} = {x + y}')


add(1, 2)

# 执行了 add 函数
# 1 + 2 = 3
# add 函数执行完成

def logger(func):
    def wrapper(*args, **kwargs):
        func_name = func.__name__
        print(f'执行了 {func_name} 函数')
        ret = func(*args, **kwargs)
        print(f'{func_name} 函数执行完成')
        return ret

    return wrapper


@logger
def add(x, y):
    print(f'{x} + {y} = {x + y}')
    return x + y


add_res = add(1, 2)
print(add_res)

# 执行了 add 函数
# 1 + 2 = 3
# add 函数执行完成
# 3

`定时执行任务`装饰器

本节另需安装 schedule；验证使用的是 schedule==1.2.2，可运行 python -m pip install schedule==1.2.2。请在独立脚本中运行，下面的演示窗口为 3.2 秒，结束后清理本例注册的任务，不会一直占用终端。

@repeat 在函数定义时注册任务，外层 @run_every 返回的包装函数负责启动轮询；这里的任务不接收参数。调度不是实时保证，运行次数会受任务耗时和系统调度影响；正常空闲环境下可观察到约每秒一次的三行输出。长任务、并行任务和精确计时需另行设计，参见 Schedule 官方示例。

import functools
import time

import schedule
from schedule import every, repeat


def run_task(duration=3.2):
    """
    定时执行任务
    :return:
    """

    deadline = time.monotonic() + duration
    while time.monotonic() < deadline:
        schedule.run_pending()
        time.sleep(0.05)


def run_every(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            run_task()
        except Exception:
            print('任务执行出错啦')
            raise
        finally:
            schedule.clear('decorator-demo')

    return wrapper


@run_every
@repeat(every(1).seconds.tag('decorator-demo'))
def test():
    print('hello world')


test()

# hello world
# hello world
# hello world

带参数的装饰器

import functools


def run_every(param):
    """
    带参数的普通装饰器
    :param param:
    :return:
    """

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            print(f'{param=}')
            return func(*args, **kwargs)

        return wrapper

    return decorator


@run_every('你好')
def test():
    print('hello world')


test()

# param='你好'
# hello world

不带参数的类装饰器

class logger(object):

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

    def __call__(self, *args, **kwargs):
        func_name = self.func.__name__
        print(f'执行了 {func_name} 函数')
        ret = self.func(*args, **kwargs)
        print(f'{func_name} 函数执行完成')
        return ret


@logger
def test():
    print('test()')


ret_print = test()
print(ret_print)

# 执行了 test 函数
# test()
# test 函数执行完成
# None

带参数的类装饰器

class logger(object):
    def __init__(self, level='INFO'):
        self.level = level

    def __call__(self, func):
        def wrapper(*args, **kwargs):

            func_name = func.__name__
            print(f'{self.level} 日志等级')
            print(f'执行了 {func_name} 函数')
            ret = func(*args, **kwargs)
            print(f'{func_name} 函数执行完成')
            return ret

        return wrapper


@logger(level='DEBUG')
def test():
    print('test()')


ret_print = test()
print(ret_print)

# DEBUG 日志等级
# 执行了 test 函数
# test()
# test 函数执行完成
# None

将装饰器定义为类的一部分

from functools import wraps


class A:
    # 装饰器作为实例方法
    def decorator1(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            print('Decorator 1')
            return func(*args, **kwargs)

        return wrapper

    # 装饰器作为类方法
    @classmethod
    def decorator2(cls, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            print('Decorator 2')
            return func(*args, **kwargs)

        return wrapper


a = A()


@a.decorator1
def test1():
    pass


@A.decorator2
def test2():
    pass


test1()
test2()

# Decorator 1
# Decorator 2

装饰器

装饰器