image.png

1、闭包的概念

  1. # 定义一个函数
  2. def test(number):
  3. # 在函数内部在定义一个函数,并且这个函数用到外围函数的变量
  4. # 那么将这个函数及用到的一些变量称之为闭包
  5. def test_in(number_in):
  6. print("在test_in函数内部,number_in的值为:%d" % number_in)
  7. return number + number_in
  8. # 其实这里返回的是闭包,也就是内部的函数引用
  9. return test_in
  10. # 给test函数赋值,这个20就是参数number
  11. ret = test(20)
  12. # 注意这里的100就是参数number_in
  13. print(ret)
  14. # print(ret(10))
  15. res = ret(10)
  16. print(res)
  17. # 上面的代码我们可以简化的写成
  18. ret = test(20)(10)

执行结果

  1. <function test.<locals>.test_in at 0x7fdfacad4950>
  2. test_in函数内部,number_in的值为:10
  3. 30

打印ret的时候我们得到了是一个函数.test_in at 0x7fdfacad4950>,函数是可以通过函数名加括号进行调用的。ret(10)

  • 在函数内部在定义一个函数,并且这个函数用到外围函数的变量,那么将这个函数及用到的一些变量称之为闭包

2、hello,装饰器

  1. def decorator(func):
  2. def wrapper(*args, **kw):
  3. return func()
  4. return wrapper
  5. @decorator
  6. def function():
  7. print("hello, decorator")

我们看到decorator函数和我们上面写的闭包函数一样都是函数内部再定义一个函数。上面的代码可以抽象的形容为:decorator函数是一个帽子,function函数是一个人,然后把帽子带到人头上。那么当我们去看一个人的时候肯定是从头看到尾。装饰器的执行顺序也是自顶向下。

3、无参装饰器

3.1 日志打印装饰器

# 这是装饰器函数,参数 func 是被装饰的函数
def logger(func):
    def wrapper(*args, **kw):
        print('主人,我准备开始执行:{} 函数了:'.format(func.__name__))
        # 真正执行的是这行。
        func(*args, **kw)
        print('主人,我执行完啦。')
    return wrapper

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

add(10,20)

执行结果

主人,我准备开始执行:add 函数了:
200 + 50 = 250
主人,我执行完啦。

4、带参的函数装饰器

上面的装饰器中并不能携带参数,只能在简单的场景中运用。装饰器的本身就是一个函数,一个函数如果不能传参那么该函数的功能就很单一,只能执行固定的逻辑,意味着复用性很低。如果被装饰的逻辑代码的执行需要根据不同场景进行调整,使用不能传参的装饰器,就需要重新编写装饰器,这样装饰器就没有太大的意义。比如当我想实现一个定时发送消息的功能,A任务我想要每隔20s发送一条消息,B任务我想要每隔60s发送一条消息,我们就可实现一个定时任务的装饰器,该装饰器可以接受一个时间间隔的参数,间隔多久执行一次任务。

import time


def periodic_task(spacing):
    def decorate(func):
        def wrapper(*args, **kwargs):
            time.sleep(spacing)
            return func(*args, **kwargs)
        return wrapper

    return decorate


@periodic_task(spacing=20)
def send_msg1():
    print("%s | 发送一条消息"% time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()))

@periodic_task(spacing=60)
def send_msg2():
    pass


if __name__ == '__main__':
    for i in range(10):
        send_msg1()

实现记录日志功能,包含级别

import logging
from functools import wraps


def logged(level, name=None, message=None):
    def decorate(func):
        logname = name if name else func.__module__
        log = logging.getLogger(logname)
        logmsg = message if message else func.__name__

        @wraps(func)
        def wrapper(*arg, **kwargs):
            log.log(level, logmsg)
            return func(*arg, **kwargs)

        return wrapper

    return decorate


@logged(logging.DEBUG)
def add(x, y):
    return x + y


@logged(logging.CRITICAL, 'example')
def spam():
    print('Spam!')


add(10, 10)
spam()

5、不带参的类装饰器

上面都是基于函数实现的装饰器,通常还会有基于类实现的装饰器。
基于类装饰器的实现,必须实现callinit两个内置方法。

  • init: 接收被装饰的函数
  • call: 实现装饰逻辑 ```python class logger(object): def init(self, func):

      self.func = func
    

    def call(self, args, *kwargs):

      print("[INFO]: the function {func}() is running...".format(func=self.func.__name__))
      return self.func(*args, **kwargs)
    

@logger def say(something): print(“say {}!”.format(something))

say(“hello”)

<a name="Zhj9f"></a>
# 6、带参数的类装饰器
上面不带参数的例子,你发现没有,只能打印INFO级别的日志,正常情况下,我们还需要打印DEBUG WARNING等级别的日志。 这就需要给类装饰器传入参数,给这个函数指定级别了。
> **带参数和不带参数的类装饰器有很大的不同。**
> **__init__ :不再接收被装饰函数,而是接收传入参数。**__call__ :接收被装饰函数,实现装饰逻辑。**

<a name="kb7ap"></a>
## 6.1 打印日志
```python
class logger(object):
    def __init__(self, level='INFO'):
        self.level = level

    def __call__(self, func): # 接受函数
        def wrapper(*args, **kwargs):
            print("[{level}]: the function {func}() is running...".format(level=self.level, func=func.__name__))
            func(*args, **kwargs)
        return wrapper  #返回函数

@logger(level='WARNING')
def say(something):
    print("say {}!".format(something))

say("hello")

6.2 实现falsk中Router功能

class Router(object):

    def __init__(self):
        self.url_map = {}

    def register(self, url):
        def wrapper(func):
            self.url_map[url] = func
        return wrapper

    def call(self, url):
        func = self.url_map.get(url)
        if not func:
            raise ValueError('No url function: %s', url)
        return func()

router = Router()

@router.register('/page1')
def page1():
    return 'this is page1'

@router.register('/page2')
def page2():
    return 'this is page2'

print(router.call('/page1'))
print(router.call('/page2'))

7、偏函数与类实现装饰器

绝大多数装饰器都是基于函数和闭包实现的,但这并不是制造装饰器的唯一方式。
事实上,Python 对某个对象是否能通过装饰器( @decorator)形式使用只有一个要求:decorator 必须是一个“可被调用(callable)的对象。
对于这个 callable 对象,我们最熟悉的就是函数了。
除函数之外,类也可以是 callable 对象,只要实现了call 函数(上面几个例子已经接触过了)。
还有容易被人忽略的偏函数其实也是 callable 对象。
接下来就来说说,如何使用 类和偏函数结合实现一个与众不同的装饰器

import time
import functools

class DelayFunc:
    def __init__(self,  duration, func):
        self.duration = duration
        self.func = func

    def __call__(self, *args, **kwargs):
        print(f'Wait for {self.duration} seconds...')
        time.sleep(self.duration)
        return self.func(*args, **kwargs)

    def eager_call(self, *args, **kwargs):
        print('Call without delay')
        return self.func(*args, **kwargs)

def delay(duration):
    """
    装饰器:推迟某个函数的执行。
    同时提供 .eager_call 方法立即执行
    """
    # 此处为了避免定义额外函数,
    # 直接使用 functools.partial 帮助构造 DelayFunc 实例
    return functools.partial(DelayFunc, duration)

@delay(duration=2)
def add(a, b):
    return a+b

执行过程

>>> add    # 可见 add 变成了 Delay 的实例
<__main__.DelayFunc object at 0x107bd0be0>
>>> 
>>> add(3,5)  # 直接调用实例,进入 __call__
Wait for 2 seconds...
8
>>> 
>>> add.func # 实现实例方法
<function add at 0x107bef1e0>

8、类的类装饰器

class decorateClass: #装饰器类
    def __init__(self,wrapedClass,*a,**k): #wrapedClass代表被装饰类
        print("准备执行装饰类初始化")

        class wrapClass(wrapedClass):
            def __init__(self,*a,**k):
                print(f"初始化被封装类实例开始,位置参数包括:{a}, 关键字参数为{k}")
                super().__init__(*a,**k)
                print(f"初始化被封装类实例结束")

        self.wrapedClass=wrapClass
        print("装饰类初始化完成")

    def __call__(self, *a, **k):
        print("被装饰类对象初始化开始")
        wrapedClassObj = self.wrapedClass(*a,**k)
        print("被装饰类对象初始化结束")
        return wrapedClassObj

@decorateClass
class car:
    def __init__(self,type,weight,cost):
        print("class car __init__ start...")
        self.type = type
        self.weight = weight
        self.cost = cost
        self.distance = 0
        print("class car __init__ end.")

    def driver(self,distance):
        self.distance += distance
        print(f"{self.type}已经累计行驶了{self.distance}公里")

c = car('爱丽舍','1.2吨',8)
c.driver(10)
c.driver(110)

9、装饰器实战

9.1 通过装饰器实现参数校验

项目开发中进行参数校验是个再正常不过得事情,毕竟我们无法保证前端传入正确的数据。比如在 fastapi 中最常使用的就是 pydantic 库,这个库十分强大,使用起来也很简单。该如何实现该功能?想对传入的参数进行非空校验,但是又不想在函数里面调用其他函数进行参数处理。

def para_none_check(func):
    """参数非空校验"""

    @wraps(func)
    def none_check(data):
        if not data:
            raise TypeError("传入参数不能为空")
        return func(data)
    return none_check


@para_none_check
def func_test(data):
    print("func_test running")


if __name__ == '__main__':
    try:
        func_test([])
    except TypeError as e:
        print(e)  # 传入参数不能为空

9.2 实现控制函数运行超时的装饰器,如果超时则直接抛出异常

import signal
import time

class TimeoutException(Exception):
    def __init__(self, error='Timeout waiting for response from Cloud'):
        Exception.__init__(self, error)


def timeout_limit(timeout_time):
    def wraps(func):
        def handler(signum, frame):
            raise TimeoutException()

        def deco(*args, **kwargs):
            signal.signal(signal.SIGALRM, handler)
            signal.alarm(timeout_time)
            func(*args, **kwargs)
            signal.alarm(0)
        return deco
    return wraps

@timeout_limit(1)
def add():
    time.sleep(2)
    print("xxx")

add()