1、闭包的概念
# 定义一个函数def test(number):# 在函数内部在定义一个函数,并且这个函数用到外围函数的变量# 那么将这个函数及用到的一些变量称之为闭包def test_in(number_in):print("在test_in函数内部,number_in的值为:%d" % number_in)return number + number_in# 其实这里返回的是闭包,也就是内部的函数引用return test_in# 给test函数赋值,这个20就是参数numberret = test(20)# 注意这里的100就是参数number_inprint(ret)# print(ret(10))res = ret(10)print(res)# 上面的代码我们可以简化的写成ret = test(20)(10)
执行结果
<function test.<locals>.test_in at 0x7fdfacad4950>在test_in函数内部,number_in的值为:1030
打印ret的时候我们得到了是一个函数
- 在函数内部在定义一个函数,并且这个函数用到外围函数的变量,那么将这个函数及用到的一些变量称之为闭包
2、hello,装饰器
def decorator(func):def wrapper(*args, **kw):return func()return wrapper@decoratordef function():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、不带参的类装饰器
上面都是基于函数实现的装饰器,通常还会有基于类实现的装饰器。
基于类装饰器的实现,必须实现call和init两个内置方法。
- init: 接收被装饰的函数
call: 实现装饰逻辑 ```python class logger(object): def init(self, func):
self.func = funcdef 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()
