Shulou(Shulou.com)06/01 Report--

Most people do not understand the knowledge points of this "Python decorator case analysis" article, so the editor summarizes the following contents, detailed contents, clear steps, and has a certain reference value. I hope you can get something after reading this article. Let's take a look at this "Python decorator case analysis" article.

Task exited during timeout

We use all kinds of network request libraries with timeout parameters, such as request library.

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

What if the method we developed ourselves wants to add this feature?

There are many ways, but the simplest and most straightforward is to use the concurrent library futures. For convenience, I encapsulated it into a decorator with the following code:

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

With the above functions defined, 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 aquib

Task (2,3)

Results:

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 ()

four hundred and thirty five

436 def exception (self, timeout=None):

TimeoutError:

Above, we define the timeout of the function as 1 second through the decorator, and successfully throw the timeout exception when the sleep simulation function is executed for more than 1 second.

When the program can be completed within the timeout:

@ timeout (1)

Def task (a, b):

Time.sleep (0.9)

Return aquib

Task (2,3)

Results:

five

As you can see, we got the result smoothly.

In this way, we can add timeout to any function through a decorator, which can finish the task directly before it can be finished within the specified time.

Earlier, I defined the required variables to the outside. In fact, we can further encapsulate it through the class decorator. The code is 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

It has been tested that the same effect can be obtained by using class decorator.

Note: the purpose of using @ functools.wraps is because the meta-information of the decorated func function is replaced with the meta-information of the wrapper function, while @ functools.wraps (func) replaces 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, the return value of the function object is called closure logging.

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

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 'function {func.__name__} takes time {(end-start) * 1000} ms')

Return res

Return wrapper

The decorator log records the run time of a function and returns its execution result

Test it:

@ log

Def now ():

Print ('2021-7-1')

Now ()

Results:

2021-7-1

Function now takes 0.09933599994838005 ms

Caching

If a function is called frequently and the parameters are often duplicated, if the result is cached, the processing time will be saved the next time the same parameter is called

Define the function:

Import math

Import random

Import time

Def task (x):

Time.sleep (0. 01)

Return round (math.log (xylene 3 / 15), 4)

Execute:

Time

For i in range (500):

Task (random.randrange (5,10))

Results:

Wall time: 5.01 s

At this point, if we use cache, the effect will be very different. There are many decorators to implement cache, so I won't repeat the wheel. Here, we use LRU cache under functools package:

From functools import lru_cache

@ lru_cache ()

Def task (x):

Time.sleep (0. 01)

Return round (math.log (xylene 3 / 15), 4)

Execute:

Time

For i in range (500):

Task (random.randrange (5,10))

Results:

Wall time: 50 ms

Constrain the number of times a function can be executed

If we want a function in a program to be executed only once or N times throughout the program's life cycle, 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))

Results:

one

two

three

None

None above is about the "Python decorator case analysis" of this article, I believe we all have a certain understanding, I hope the editor to share the content to help you, if you want to know more related knowledge, please pay attention to the industry information channel.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.