Recommend a few great Python decorators

Python decorators are powerful and flexible to use to reduce a lot of hassle.

The task timed out to exit

The various network request libraries we use on a daily basis come with timeout parameters, such as the request library

This parameter allows the request to time out and no longer continue, throwing a timeout error directly, avoiding waiting too long

What if we develop our own methods that also want to add this functionality?

There are many methods, but the simplest and most straightforward way is to use concurrent library futures, which I encapsulated as a decorator for ease of use, as follows:

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

By defining the above functions, we have a decorator that ends with a timeout, which can be tested below:

import time

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

task(2, 3)

outcome:

---------------------------------------------------------------------------
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:

Above, we defined a function timeout of 1 second through the decorator, and successfully threw a timeout exception when executed by the sleep simulation function for more than 1 second

When the program is able to complete within the timeout time:

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

task(2, 3)

outcome:

5

As you can see, the results are going well

This allows us to add timeouts to any function through a decorator that can end the task directly if it is not finished within the specified time

Earlier I defined the desired variables to the outside, but we can actually encapsulate them further through the class decorator, as follows:

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

The same effect can be obtained by testing the use of class decorators.

Note: The purpose of using @functools.wraps is to replace the meta-information of the wrapper function with the meta-information of the wrapper function because the decorated func function meta-information is replaced, and the @functools.wraps (func) replace the meta-information of the wrapper function with the meta-information of the func function. In the end, although the wrapper function is returned, the meta-information is still the original func function in functional programming, and the return value of the function is that the function object is called a closure

Logging

If we need to record the execution time of some functions, the decorator is a convenient choice to print some logs before and after the function is executed

The code is as follows:

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

Decorator log records the run time of a function and returns its execution results

Test it out:

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

outcome:

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

cache

If a function is called frequently and the arguments are often duplicated, if the results are cached, processing time is saved the next time the same argument is called

Defining functions:

import math
import random
import time


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

execute:

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

outcome:

Wall time: 5.01 s

At this point, if we use caching, the effect will be very different, there are a lot of decorators to implement the cache, I will not repeat the wheel, here using the LRU cache under the functools package:

from functools import lru_cache

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

execute:

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

outcome:

Wall time: 50 ms

The number of times a function can be executed

If we want a function in a program to execute only once or N times throughout the life of the program, we can write a decorator like this:

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

Test:

@allow_count(3)
def job(x):
    x += 1
    return x


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

outcome:

1
2
3
None
None

Reprinted from: Rookie Python

That's all you need to do to recommend a few great Python decorators.