温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

Python装饰器实例分析

发布时间:2022-02-19 16:54:40 阅读:124 作者:iii 栏目:开发技术
Python开发者专用服务器限时活动,0元免费领,库存有限,领完即止! 点击查看>>

这篇“Python装饰器实例分析”文章的知识点大部分人都不太理解,所以小编给大家总结了以下内容,内容详细,步骤清晰,具有一定的借鉴价值,希望大家阅读完这篇文章能有所收获,下面我们一起来看看这篇“Python装饰器实例分析”文章吧。

任务超时退出

我们日常在使用的各种网络请求库时都带有 timeout 参数,例如:request 库

这个参数可以使请求超时就不再继续了,直接抛出超时错误,避免等太久

如果我们自己开发的方法也希望增加这个功能,该如何做呢?

方法很多,但最简单直接的是使用并发库 futures,为了使用方便,我将其封装成了一个装饰器,代码如下:

import functoolsfrom concurrent import futuresexecutor = futures.ThreadPoolExecutor(1)def timeout(seconds):    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kw):            future = executor.submit(func, *args, **kw)            return future.result(timeout=seconds)        return wrapper    return decorator

定义了以上函数,我们就有了一个超时结束的装饰器,下面可以测试一下:

import time@timeout(1)def task(a, b):    time.sleep(1.2)    return a+btask(2, 3)

结果:

---------------------------------------------------------------------------TimeoutError                              Traceback (most recent call last)...D:\Anaconda3\lib\concurrent\futures\_base.py in result(self, timeout)    432                 return self.__get_result()    433             else:--> 434                 raise TimeoutError()    435     436     def exception(self, timeout=None):TimeoutError:

上面我们通过装饰器定义了函数的超时时间为 1 秒,通过睡眠模拟函数执行超过 1 秒时,成功的抛出了超时异常

程序能够在超时时间内完成时:

@timeout(1)def task(a, b):    time.sleep(0.9)    return a+btask(2, 3)

结果:

5

可以看到,顺利的得到了结果

这样我们就可以通过一个装饰器给任何函数增加超时时间,这个函数在规定时间内还处理不完就可以直接结束任务

前面我将这个装饰器将所需的变量定义到了外部,其实我们还可以通过类装饰器进一步封装,代码如下:

import functoolsfrom concurrent import futuresclass timeout:    __executor = futures.ThreadPoolExecutor(1)    def __init__(self, seconds):        self.seconds = seconds    def __call__(self, func):        @functools.wraps(func)        def wrapper(*args, **kw):            future = timeout.__executor.submit(func, *args, **kw)            return future.result(timeout=self.seconds)        return wrapper

经测试使用类装饰器能得到同样的效果。

注意:使用 @functools.wraps 的目的是因为被装饰的 func 函数元信息会被替换为 wrapper 函数的元信息,而 @functools.wraps(func) 将 wrapper 函数的元信息替换为 func 函数的元信息。最终虽然返回的是 wrapper 函数,元信息却依然是原有的 func 函数在函数式编程中,函数的返回值是函数对象被称为闭包

日志记录

如果我们需要记录部分函数的执行时间,函数执行前后打印一些日志,装饰器是一种很方便的选择

代码如下:

import timeimport functools def log(func):    @functools.wraps(func)    def wrapper(*args, **kwargs):        start = time.perf_counter()        res = func(*args, **kwargs)        end = time.perf_counter()        print(f'函数 {func.__name__} 耗时 {(end - start) * 1000} ms')        return res    return wrapper

装饰器 log 记录某个函数的运行时间,并返回其执行结果

测试一下:

@logdef now():    print('2021-7-1')    now()

结果:

2021-7-1函数 now 耗时 0.09933599994838005 ms

缓存

如果经常调用一个函数,而且参数经常会产生重复,如果把结果缓存起来,下次调用同样参数时就会节省处理时间

定义函数:

import mathimport randomimport timedef task(x):    time.sleep(0.01)    return round(math.log(x**3 / 15), 4)

执行:

%%timefor i in range(500):    task(random.randrange(5, 10))

结果:

Wall time: 5.01 s

此时如果我们使用缓存的效果就会大不一样,实现缓存的装饰器有很多,我就不重复造轮子了,这里使用 functools 包下的 LRU 缓存:

from functools import lru_cache@lru_cache()def task(x):    time.sleep(0.01)    return round(math.log(x**3 / 15), 4)

执行:

%%timefor i in range(500):    task(random.randrange(5, 10))

结果:

Wall time: 50 ms

约束某个函数的可执行次数

如果我们希望程序中的某个函数在整个程序的生命周期中只执行一次或 N 次,可以写一个这样的装饰器:

import functoolsclass allow_count:    def __init__(self, count):        self.count = count        self.i = 0    def __call__(self, func):        @functools.wraps(func)        def wrapper(*args, **kw):            if self.i >= self.count:                return            self.i += 1            return func(*args, **kw)        return wrapper

测试:

@allow_count(3)def job(x):    x += 1    return xfor i in range(5):    print(job(i))

结果:

123NoneNone

以上就是关于“Python装饰器实例分析”这篇文章的内容,相信大家都有了一定的了解,希望小编分享的内容对大家有帮助,若想了解更多相关的知识内容,请关注亿速云行业资讯频道。

亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>

向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI

开发者交流群×