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)
If maxsize is set to None, the LRU
feature is disabled and the cache can grow without bound.
Since a dictionary is used to cache results, the positional and keyword arguments to the function must be hashable. If user_function is specified, it must be a callable. This allows the lru_cache decorator to be applied directly to a user function, leaving the maxsize at its default value of 128.
The cache keeps references to the arguments and return values until they age out of the cache or until the cache is cleared.
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)
decorator to multithreading any function
Decorators can be used to add multithreading to functions in a more elegant and reusable way. A decorator is a function that takes another function and extends its behavior without explicitly modifying it.
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
# Decorator to add multithreading
def multithreaded(max_workers=5):
def decorator(func):
def wrapper(*args, **kwargs):
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_args = {executor.submit(func, arg): arg for arg in args[0]}
results = []
for future in as_completed(future_to_args):
arg = future_to_args[future]
try:
result = future.result()
except Exception as exc:
print(f'{arg} generated an exception: {exc}')
else:
results.append(result)
return results
return wrapper
return decorator
usage:
# Function to square a number
@multithreaded(max_workers=5)
def square_number(number):
time.sleep(1) # Simulate a time-consuming task
return number * number
# List of numbers to process
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# Using the decorated function
start_time = time.time()
squared_numbers = square_number(numbers)
end_time = time.time()
print("Squared numbers:", squared_numbers)
print("Time taken:", end_time - start_time, "seconds")
Using decorators to handle multithreading not only simplifies the code but also makes it more reusable and cleaner. You can easily apply the @multithreaded decorator to any function that needs to be executed in parallel, providing a flexible and powerful way to optimize your Python code.
@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.
