App下載

推薦幾個很出色的的Python裝飾器

猿友 2021-03-29 17:40:58 瀏覽數 (2941)
反饋

Python 裝飾器功能強大,靈活運用能為我們減少很多麻煩。

任務超時退出

我們日常在使用的各種網絡請求庫時都帶有 timeout 參數,例如:request 庫

這個參數可以使請求超時就不再繼續(xù)了,直接拋出超時錯誤,避免等太久

如果我們自己開發(fā)的方法也希望增加這個功能,該如何做呢?

方法很多,但最簡單直接的是使用并發(fā)庫 futures,為了使用方便,我將其封裝成了一個裝飾器,代碼如下:

import functools
from concurrent import futures

executor = 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+b

task(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 秒,通過睡眠模擬函數執(zhí)行超過 1 秒時,成功的拋出了超時異常

程序能夠在超時時間內完成時:

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

task(2, 3)

結果:

5

可以看到,順利的得到了結果

這樣我們就可以通過一個裝飾器給任何函數增加超時時間,這個函數在規(guī)定時間內還處理不完就可以直接結束任務

前面我將這個裝飾器將所需的變量定義到了外部,其實我們還可以通過類裝飾器進一步封裝,代碼如下:

import functools
from concurrent import futures

class 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 函數在函數式編程中,函數的返回值是函數對象被稱為閉包

日志記錄

如果我們需要記錄部分函數的執(zhí)行時間,函數執(zhí)行前后打印一些日志,裝飾器是一種很方便的選擇

代碼如下:

import time
import 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 記錄某個函數的運行時間,并返回其執(zhí)行結果

測試一下:

@log
def now():
print('2021-7-1')

now()

結果:

2021-7-1
函數 now 耗時 0.09933599994838005 ms

緩存

如果經常調用一個函數,而且參數經常會產生重復,如果把結果緩存起來,下次調用同樣參數時就會節(jié)省處理時間

定義函數:

import math
import random
import time


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

執(zhí)行:

%%time
for 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)

執(zhí)行:

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

結果:

Wall time: 50 ms

約束某個函數的可執(zhí)行次數

如果我們希望程序中的某個函數在整個程序的生命周期中只執(zhí)行一次或 N 次,可以寫一個這樣的裝飾器:

import functools


class 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 x


for i in range(5):
print(job(i))

結果:

1
2
3
None
None

轉載自:菜鳥學Python

以上就是小編為您整理的 推薦幾個很出色的的Python裝飾器 的全部內容。


0 人點贊