Fabulous Python Decorators

Fabulous Python Decorators

@cache

@functools.cache(user_function)

Simple lightweight unbounded function cache. Sometimes called "memoize" .

Returns the same as lru_cache(maxsize=None), creating a thin wrapper around a dictionary lookup for the function arguments. Because it never needs to evict (remove) old values, this is smaller and faster than lru_cache() with a size limit.

example:

from functools import cache

@cache
def factorial(n):
    return n * factorial(n-1) if n else 1

>>> factorial(10)      # no previously cached result, makes 11 recursive calls
3628800
>>> factorial.cache_info()
CacheInfo(hits=0, misses=11, maxsize=None, currsize=11)

>>> factorial(5)       # just looks up cached value result
120
>>> factorial.cache_info()
CacheInfo(hits=1, misses=11, maxsize=None, currsize=11)

>>> factorial(12)      # makes two new recursive calls, the other 10 are cached
479001600
>>> factorial.cache_info()
CacheInfo(hits=2, misses=13, maxsize=None, currsize=13)

@lru_cache

This decorator comes to us from the functools module. This module is included in the standard library, and is incredibly easy to use. This decorator can be used to speed up consecutive runs of functions and operations using cache. @lru_cache decorator wrap a function with a memoizing callable that saves up to the maxsize most recent calls. It can save time when an expensive or I/O bound function is periodically called with the same arguments.

@functools.lru_cache(user_function)
@functools.lru_cache(maxsize=128, typed=False)

Distinct argument patterns may be considered to be distinct calls with separate cache entries. For example, f(a=1, b=2) and f(b=2, a=1) differ in their keyword argument order and may have two separate cache entries.

If typed is set to true, function arguments of different types will be cached separately. For example, f(3) and f(3.0) will always be treated as distinct calls with distinct results. If typed is false, the implementation will usually but not always regard them as equivalent calls and only cache a single result.

The wrapped function is instrumented with a cache_parameters() function that returns a new dict showing the values for maxsize and typed. This is for information purposes only. Mutating the values has no effect.

To help measure the effectiveness of the cache and tune the maxsize parameter, the wrapped function is instrumented with a cache_info() function that returns a named tuple showing hits, misses, maxsize and currsize.

The decorator also provides a cache_clear() function for clearing or invalidating the cache. The original underlying function is accessible through the __wrapped__ attribute. This is useful for introspection, for bypassing the cache, or for rewrapping the function with a different cache.

example:

@lru_cache(maxsize=None)
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)

>>> [fib(n) for n in range(16)]
[0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610]

>>> fib.cache_info()
CacheInfo(hits=28, misses=16, maxsize=None, currsize=16)

>>> fib.cache_parameters()
{'maxsize': None, 'typed': False}

>>> fib.cache_clear()
>>> fib.cache_info()
CacheInfo(hits=0, misses=0, maxsize=None, currsize=0)

@debug

The @debug decorator is used to print debug information about a function call, including its arguments and return value. This can be useful for debugging complex functions or finding performance bottlenecks.

from functools import wraps

def debug(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"Calling {func.__name__} with args={args} kwargs={kwargs}")
        result = func(*args, **kwargs)
        print(f"{func.__name__} returned {result}")
        return result
    return wrapper

@debug
def my_function(x, y):
    return x + y

my_function(1, 2) # prints "Calling my_function with args=(1, 2) kwargs={}" and "my_function returned 3"

The retry decorator

In data science projects and software development projects, there are so many instances where we depend on external systems. Things are not in our control all the time. When an unexpected event occurs, we might want our code to wait a while, allowing the external system to correct itself and rerun. I prefer to implement this retry logic inside a python decorator so that I can annotate any function to apply the retry behavior.

Here's the code for a retry decorator.

import requests
import time
from functools import wraps

def retry(max_tries=3, delay_seconds=1):
    def decorator_retry(func):
        @wraps(func)
        def wrapper_retry(*args, **kwargs):
            tries = 0
            while tries < max_tries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    tries += 1
                    if tries == max_tries:
                        raise e
                    time.sleep(delay_seconds)
        return wrapper_retry
    return decorator_retry


@retry(max_tries=5, delay_seconds=2)
def call_dummy_api():
    response = requests.get("https://jsonplaceholder.typicode.com/todos/1")
    return response

Timing decorator

Here's an example Python decorator that prints the running time of a function when it's called:

import time
from functools import wraps

def timing_decorator(func):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"Function {func.__name__} took {end_time - start_time} seconds to run.")
        return result
    return wrapper


@timing_decorator
def my_function():
    # some code here
    time.sleep(1)  # simulate some time-consuming operation
    return

my_function()

# output:
Function my_function took 1.0012552738189697 seconds to run.

Logging decorator

When you design your code in such a way, you'd also want to log the execution information of your functions. This is where logging decorators come in handy.

import logging
import functools

logging.basicConfig(level=logging.INFO)

def log_execution(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logging.info(f"Executing {func.__name__}")
        result = func(*args, **kwargs)
        logging.info(f"Finished executing {func.__name__}")
        return result
    return wrapper

@log_execution
def extract_data(source):
    # extract data from source
    data = ...

    return data

Email notification decorator

The following decorator sends an email whenever the execution of the inner function fails. It doesn't have to be an email notification in your case. You can configure it to send a Teams/slack notification.

import smtplib
import traceback
from email.mime.text import MIMEText
from functools import wraps

def email_on_failure(sender_email, password, recipient_email):
    def decorator(func):
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                # format the error message and traceback
                err_msg = f"Error: {str(e)}\n\nTraceback:\n{traceback.format_exc()}"
                
                # create the email message
                message = MIMEText(err_msg)
                message['Subject'] = f"{func.__name__} failed"
                message['From'] = sender_email
                message['To'] = recipient_email
                
                # send the email
                with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp:
                    smtp.login(sender_email, password)
                    smtp.sendmail(sender_email, recipient_email, message.as_string())
                    
                # re-raise the exception
                raise
                
        return wrapper
    
    return decorator

@email_on_failure(sender_email='your_email@gmail.com', password='your_password', recipient_email='recipient_email@gmail.com')
def my_function():
    # code that might fail

@jit

JIT is short for Just In Time compilation. Normally whenever we run some code in Python, the first thing that happens is compilation. This compilation creates a bit of overhead, as types are allocated memory and stored as unassigned but named aliases. With Just In Time compilation, we do most of this work at execution. In a lot of ways, we can think of this as something akin to parallel computing, where the Python interpreter is working on two things at once in order to save some time.

The Numba JIT compiler is famous for providing that very concept into Python. Similarly to the @lru_cache, this decorator can be called pretty easily with an immediate boost to performance in your code. The Numba package provides the jit decorator, which makes running more intensive software a lot easier without having to drop into C.

example:

from numba import jit
import random

@jit(nopython=True)
def monte_carlo_pi(nsamples):
    acc = 0
    for i in range(nsamples):
        x = random.random()
        y = random.random()
        if (x ** 2 + y ** 2) < 1.0:
            acc += 1
    return 4.0 * acc / nsamples

@use_unit

Decorator that might come in handy quite often for scientific computing is the self-made use_unit decorator. This can be useful for those who don’t want to add units of measurement to their data, but still want people to know what those units are.

import functools

def use_unit(unit):
    '''Have a function return a Quantity with given unit'''
    def decorator_use_unit(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            value = func(*args, **kwargs)
            return f'{value} {unit}'
        return wrapper
    return decorator_use_unit


@use_unit("meters per second")
def average_speed(distance, duration):
    return distance / duration

average_speed(100, 20)

# output:
# '5.0 meters per second'

@register

The register function comes from the module atexit . This decorator could have something to do with performing some action at termination. The register decorator names a function to be ran at termination. For example, this would work will with some software that needs to save whenever you exit.

>>> import atexit
>>> 
>>> @atexit.register
... def goodbye(name="Danny", adjective="nice"):
...     print(f'Goodbye {name}, it was {adjective} to meet you.')
... 
>>> # type CTRL + D for exit python shell 
Goodbye Danny, it was nice to meet you.

SUBSCRIBE FOR NEW ARTICLES

@
comments powered by Disqus