Decorators, Logging, and Threading in Python

Síguenos / Follow us!
Facebooktwitterlinkedinrssyoutubeinstagram

Decorators


Functions are first-class citizens in Python. They behave like any other object. In other words, you can use functions as arguments to other functions, store functions as dictionary values, or return a function from another function.

Decorators are functions that modify the functionality of other functions. They wrap a function and extend or modify its behavior in one way or another. We are going to define a decorator that is going to track the time taken by a function to execute.

import random
import time
def timing(func):
    def wrapper(*arg, **kw): # args and kw collect all arguments and keywords. 
        '''source: http://www.daniweb.com/code/snippet368.html'''
        t1 = time.time()
        res = func(*arg, **kw) # The wrapped function is executed at this time.
        t2 = time.time()
        # Finally, we print the function's name and the time taken by the function to execute (t1 - t2)
        print("%r  %2.2f ms" % \ 
                  (func.__name__, (t2 - t1) * 1000))
    return wrapper

# This extra function is going to generate a random list of integers. 
def generateRandom(minimum = 1, maximum = 100, numberElements = 10):
    return [int(random.randint(minimum, maximum)) for e in range(numberElements) ]

user@pc:~$ python
Python 3.9.5 (default, May 11 2021, 08:20:37)
[GCC 10.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> mylist = [] # If we want to get a random list of integers, one of the fist things that come to mind would be to use a "for loop".
>>> for e in range (10):
... mylist.append(random.randint(1, 10))
>>> print(mylist)
[10, 5, 9, 1, 7, 8, 10, 1, 6, 6] # However, there is a more elegant and less verbose way to do it by using Python list comprehension.
>>> [random.randint(1,10) for e in range (10)]
[10, 8, 9, 4, 1, 8, 10, 6, 9, 7]

@timing #  @my_decorator is just syntactic sugar, an easier way of saying func = my_decorator(func).
def comparing_quick_sort():
    myRandomList = generateRandom(1, 100, 20)
    myOrderList = quick_sort(myRandomList)
    print(myOrderList)
    
@timing
def comparing_default_sort(): # We are going to compare quick_sort with Python List sort() method.
    myRandomList = generateRandom(1, 100, 20)
    myRandomList.sort()
    print(myRandomList)

def quick_sort(myList):
    if len(myList) <= 1:
        return myList
    else:
        return quick_sort([x for x in myList[1:] if x < myList[0]]) \
            + [myList[0]] + quick_sort([x for x in myList[1:] if x >= myList[0]])

if __name__ == '__main__':
    comparing_quick_sort()
    comparing_default_sort()
[19, 30, 40, 44, 46, 51, 52, 54, 55, 58, 61, 62, 65, 66, 67, 71, 80, 82, 84, 93] 'comparing_quick_sort' 0.07 ms
[5, 8, 10, 23, 25, 30, 32, 38, 43, 53, 53, 53, 54, 58, 61, 64, 69, 75, 100, 100] 'comparing_default_sort' 0.05 ms

As a general rule of thumb, recursion is fairly expensive compared to iteration. Let's study this code and discuss what's going on.

def fib(n): # We are implementing the Fibonacci sequence in Python. The classical approach is a recursive one.
	if n<=1:
		return n
	return fib(n-1) + fib(n-2)

@timing
def main():
	for i in range(40):
		print(i, fib(i))

if __name__=='__main__':
	main()

The printed result is 'main' 44101.24 ms. That's insane! This recursive version is very naive or plainly moronic, it calculates everything over and over again.

The functools module is for higher-order functions: functions that act on or return other functions. @Cache is a decorator. It is a simple lightweight unbounded function cache that will cache all successful results, to speed up our function calls. An alternative is: @lru_cache(maxsize=5). It saves up to the maxsize most recent calls.

from functools import cache #
@cache 
def fib(n):
	if n<=1:
		return n
	return fib(n-1) + fib(n-2)

@timing
def main():
	for i in range(40):
		print(i, fib(i))

if __name__=='__main__':
	main()

The results is 'main' 0.66 ms. Wow! That's a difference.

Logging in Python

Logging is the process of tracking and recording events that happen when our code runs, so we can trace, understand, and debug it. The logging module defines functions and classes which implement a flexible event logging system for applications and makes logging very easily implementable.

import logging 
class Debugger(object):
    """ Debug a method and return it back"""
    def __init__(self, func):
        self.func = func # When a class decorator is defined on a wrapped function, only the __init__ method of the decorator class is executed, with the first argument being the wrapped function and we save it as a class attribute to be retrieved at execution time. 
        LOG_FORMAT = '%(asctime)s - %(levelname)s - %(message)s' # The format is: date/time levelname (severity), and message. 
         logging.basicConfig(level=logging.DEBUG, filemode = 'w'
                            filename="testing.log", format=LOG_FORMAT)

The last line of our code does basic configuration for the logging system:

        self.logger = logging.getLogger(__name__) # It creates a logger instance which is the root logger of the hierarchy.

    def __call__(self, *args, **kwargs):       
        self.logger.debug(f'Entering : {self.func.__name__}') # It logs a message with level DEBUG on the root logger.
        self.logger.debug(f'args, kwargs : {args, kwargs}')
        self.logger.debug(f'{self.func.__name__} returned : {self.func(*args, **kwargs)}')

        return self.func(*args, **kwargs)

@Debugger # Another important use of decorators is for debugging. Observe that our function is being replaced by an instance of the class Debugger. 
def my_swap(a, b): # When this method is executed, the __call__ method of Debugger is executed.
    return (b, a)

if __name__ == "__main__":
    print(my_swap(5, 9))

(9, 5)
The content of testing.log is:
2021-11-12 11:40:06,849 - DEBUG - Entering : my_swap
2021-11-12 11:40:06,849 - DEBUG - args, kwargs : ((5, 9), {})
2021-11-12 11:40:06,849 - DEBUG - my_swap returned : (9, 5)

Threading in Python

A thread is a single flow of execution within a program with its own program counter that keeps track of which instruction to execute next, system registers, etc. It is the smallest sequence of programmed instructions that can be managed independently by a scheduler (an essential operating system module that selects the next jobs to be admitted into the system and the next process to run).

Python supports multi-threaded programming, but the interpreter is inherently single-threaded. The Python Global Interpreter Lock or GIL is a mutex (or a lock) that allows only one thread at a time to execute a Python opcode ("the portion of a machine language instruction that specifies the operation to be performed").

We are going to calculate pi using the Leibniz formula for π. It states that 1 - 13 + 15 - 17 - .... = π4. In other words, π = 4 - 43 + 45 - 47 - ....

import concurrent.futures
def pi(d): # It calculates (10000000-1) terms of the series starting with 4/d as the denominator.
    pi = 0 
    for i in range(1, 10000000):
        a = (2 * (i % 2))-1 // 4*a is the numerator. It alternates between 4 and -4.
        pi += a*4/d
        d += 2 // d is the denominator. It needs to increase by 2 every loop.

    return pi
    
@timing
def main():
    my_end_result = 0 # It is our calculation of π
    piRanges = [1, 19999999, 29999998, 39999997, 49999996, 59999995, 69999994, 79999993, 89999992, 99999991, 109999990, 119999989, 129999988] # The first time we call pi(), d = 1; Next, 1 + (10000000-1) = 19999999; 19999999 + 9999999 = 29999998, and so on.

    with concurrent.futures.ThreadPoolExecutor() as executor: # To start up a group of threads we are going to use a ThreadPoolExecutor. A context manager is used to create a ThreadPoolExecutor instance (aka "executor") that will clean up threads upon completion. 
        futures = [executor.submit(pi, i) for i in piRanges] # We submit as many jobs as elements are in the piRanges list to the executor. Each call to executor.submits returns a Future instance that is stored in the futures list. A future represents an eventual result of an asynchronous operation.
        for f in concurrent.futures.as_completed(futures): # concurrent.futures.as_completed(futures) returns an iterator over the Future instances given by futures that yields futures as they complete.
            my_end_result += f.result() # f.result returns the value returned by the job, first submitted and now completed.
            print(f.result())

    print("The end result is: " + str(my_end_result))
if __name__ == '__main__':
    main()

The end result is: 3.1415935946188567
'main' 25014.04 ms

Compártelo / Share it!
Facebooktwitterredditpinterestlinkedinmail

Author: Anawim

I am a social activist. I have two Bachelor's degrees, Maths and Computer & Software Engineering. I also have a Ph.D. in Psychology. I have written nine published books, four scientific articles, and five scientific presentations. I simply want to contribute to making a difference where it counts, so that we make the world a better, more sustainable, prosperous, and fairer place. I am always willing to give free talks and lectures about the social problems that exist in our world today.

Leave a Reply

Your email address will not be published. Required fields are marked *