Wrapping around wrappers: a primer on Python decorators
Mar 19, 2024
Several methods exist for applying decorators to functions, which can lead to confusion. This article aims to clarify this topic.
Uses
Simple form
At its core, a decorator modifies a function’s behavior prior to its definition. Nothing stops you from defining a class method as follows:
def foo(self):
pass
foo = classmethod(foo)
# instead of the normal
@classmethod
def foo(self):
pass
In the simplest form, a decorator looks like:
def uppercase(f):
def new_f():
return f().upper()
return new_f
#decorator form
@uppercase
def say_hi():
return 'hello there'
#function form
say_hi = uppercase(say_hi)
Nested decorators
The following two methods are equivalent, keeping in mind the inside-out order of application for decorators consistent with the functional form:
# decorator syntax
@dec2
@dec1
def func(arg1, arg2, ...):
pass
# function syntax
def func(arg1, arg2, ...):
pass
func = dec2(dec1(func))
It’s already getting a bit confusing to see a decorator with its own parameters – here decomaker(argA, argB, ...) returns a temporary function that acts on func.
# decorator syntax
@decomaker(argA, argB, ...)
def func(arg1, arg2, ...):
pass
# function syntax
func = decomaker(argA, argB, ...)(func)
A more elaborated example from PEP 318.
def accepts(*types):
def check_accepts(f):
assert len(types) == f.func_code.co_argcount
def new_f(*args, **kwds):
for (a, t) in zip(args, types):
assert isinstance(a, t), \
"arg %r does not match %s" % (a,t)
return f(*args, **kwds)
new_f.func_name = f.func_name
return new_f
return check_accepts
def returns(rtype):
def check_returns(f):
def new_f(*args, **kwds):
result = f(*args, **kwds)
assert isinstance(result, rtype), \
"return value %r does not match %s" % (result,rtype)
return result
new_f.func_name = f.func_name
return new_f
return check_returns
@accepts(int, (int,float))
@returns((int,float))
def func(arg1, arg2):
return arg1 * arg2
This is not difficult to rationalize from the perspective of currying: our job would have been done with check_returns(f; rtype) if Python allows us to parametrize a decorator in this way. However, this requires some new syntax for this purpose so that’s not good. Instead we are wrapping it around by returns(rtype) and stuck with the slightly awkward fact that the decorator parameter (int,float) for @returns((int,float)) skips the pro forma decorator check_returns and ends up in the scope for the innerest function, new_f. Realistically, you won’t need more than 3 layers of functions in your implementation of decorators.
def returns(rtype): # returns the rtype-decorator
def check_returns(f): # decorator syntax - temp function
def new_f(*args, **kwds): # new function
result = f(*args, **kwds)
assert isinstance(result, rtype), \
"return value %r does not match %s" % (result,rtype)
return result
return new_f
return check_returns
Two further observations. Firstly, employing a consistent decorator pattern across your code base could be beneficial for uniformity. The names given to these inner functions lack significant semantic value – you should be able to call the middle layer function _tmp or _decorator, the innest function new_f, and the wrapped function f, so that new contributors can quickly get used to the pattern and focus on the actual code. Second, even if your outer function comes with a default paremeter, you’d still need to invoke the decorator by @return() instead of @return. This abstraction enforces separation of concerns for decorators and wrapped functions.
Class decorators
You can replace a class with its instance with this example of a class decorator:
def singleton(cls):
instances = {}
def getinstance():
if cls not in instances:
instances[cls] = cls()
return instances[cls]
return getinstance
@singleton
class MyClass:
...
@wraps
Update a wrapper function to look like the wrapped function, essentially forcing the wrapper function to take on the identity (__name__, __doc__ and so on) of the wrapped function. Use it around the wrapped function, e.g.,
def returns(rtype):
def check_returns(f):
@wraps(f)
def new_f(*args, **kwds):
result = f(*args, **kwds)
assert isinstance(result, rtype), \
"return value %r does not match %s" % (result,rtype)
return result
new_f.func_name = f.func_name
return new_f
return check_returns
Class as decorator
Naturally follows above from the equivalence Timer(some_function)(delay). In general, avoid, since functional forms generally do.
from time import time
from time import sleep
class Timer:
def __init__(self, func):
self.function = func
def __call__(self, *args, **kwargs):
start_time = time()
result = self.function(*args, **kwargs)
end_time = time()
print("Execution took {} seconds".format(end_time-start_time))
return result
@Timer
def some_function(delay):
sleep(delay)
some_function(3)
Adjoint decorators
A practical approach involves employing decorators to modify the “basis” of a function. Suppose you have a collection of data providers featuring getters that accept requests in a specific symbology. Rather than altering all the getters to adapt to a different symbology – considering your systems may need to integrate various symbologies – decorators can be utilized to seamlessly adapt these getters. This adaptation allows them to accept and return data in the new symbology as required, instead of the original one.
If you take \(g\) as the old to the new symbology map, \(g^{-1}\) is the inverse map from the new to the old symbology. For any function \(f\), you essentially induces an “adjoint” function \(f \mapsto g f g^{-1}\), taking inspiration from an adjoint action.
@staticmethod
def ensure_new_basis(
basis_change: DataFrame[BasisSchema], col_basis: str = "basis_id"
):
forward_map = map_constructor(basis_change)
backward_map = forward_map.inverse()
def decorator(func):
@wraps(func)
def new_f(new_basis, *args, **kwargs):
results = func(backward_map(basis), *args, **kwargs)
results[col_basis] = forward_map(results[col_basis])
return results
return new_f
return decorator
#decorator
@ensure_new_basis(basis_change)
SomeClass.some_func(old_basis, params)
#functional
ensure_new_basis(basis_change)(SomeClass.some_func)(new_basis, params)
You may encounter various data sources deriving from a common base class. For instance, if you aim to switch between different pricing sources without affecting the rest of the system, it’s beneficial to enforce a uniform interface using ABC (Abstract Base Classes) and compel different data sources to adhere to the same abstract class. This approach ensures that all data sources present a consistent interface, facilitating seamless interchangeability.
class BaseClass(ABC):
@abstractmethod
"""Interface for some_func
"""
def some_func(instrument, *args, **kwargs): ...
class SomeClass(BaseClass):
"""A dervied class with implementation of some_func
"""
def some_func(self, instrument, *args, **kwargs):
return complicated_operations(instrument, *args, **kwargs)
class OtherClass(BaseClass):
"""A derived class with some dummy function for some_func
"""
def some_func(self, instrument, *args, **kwargs):
return other_complicated_operations(instrument, *args, **kwargs)
In some context, ensure_new_basis should cause no side effect. You could adopt a similar pattern on the adjoint decorators:
class BaseBasisChange(ABC):
@abstractmethod
"""Interface for ensure_new_basis
"""
def ensure_new_basis(self, basis_change: DataFrame[BasisSchema], col_basis: str = "basis_id"): ...
class NoBasisChange(BaseBasisChange):
"""A dervied class with implementation of ensure_new_basis
"""
def ensure_new_basis(self, basis_change: DataFrame[BasisSchema], col_basis: str = "basis_id"):
return lambda f: f
class ContextualBasisChange(BaseBasisChange):
"""A derived class with some dummy function for ensure_new_basis
"""
def ensure_new_basis(self, basis_change: DataFrame[BasisSchema], col_basis: str = "basis_id"):
...
Here are some valuable identities worth noting:
If \(g\) is trivial, \(f \mapsto g f g^{-1} = 1 f 1 = f\): when NoBasisChange().ensure_new_basis(SomeClass().some_func) is applied, you get the unmutated function back.
If \(f\) is trivial, \(f \mapsto g f g^{-1} = g 1 g^{-1} = 1\): when ContextualBasisChange().ensure_new_basis() is applied to some placeholder class NoOpClass().some_func, you would still get a placeholder payload back without any side effect.
Decorator hygiene
From GOF where Python’s decorator takes the inspiration from, a decorator has the following properties:
Intent
- Attach additional responsibilities or alternate processing to an object dynamically. Decorators provide a flexible alternative to subclassing for extending functionality.
Applicability
Use decorator:
- to add responsibilities to individual objects dynamically and transparently, that is, without affecting other objects.
- for responsibilities that can be withdrawn.
- when extension by subclassing is impractical. Sometimes a large number of independent extensions are possible and would produce an explosion of subclasses to support every combination. Or a class definition may be hidden or otherwise unavailable for subclassing.
Consequences
- Allows building flexible combinations and variations of responsibilities due to its strong basis on 1:1 Recursive Connection.
- Client classes need not have any knowledge of a decorator’s specific operation to use it (Liskov Substitution).
- Separates public interface from implementation and moves code down and across a shallow inheritance hierarchy (Dependency Inversion)
- Properties may be added more than once, which may be bad or good, depending on requirements
- If clients need to use a specific decorator’s interface, downcasting may be required, thus violating Liskov Substitution
- Often results in systems with large numbers of small classes which seem to look alike. This can make it harder to find specific classes. Good documentation is required to explain the design (the design pattern simplifies this chore!)
# ticker
def apply_ticker_change(func, col_key: str = "instrument_id"):
@wraps(func)
def wrapper(*args, **kwargs):
results = func(*args, **kwargs)
ticker_changes = kwargs.get("ticker_changes")
results[col_key] = results[col_key].replace(ticker_changes)
return results
return wrapper
# risk
@apply_ticker_change
def get_factor_exposures(
self,
date: str,
tickers: list[str],
ticker_changes: pd.DataFrame,
) -> pd.DataFrame:
return self.risk_reader.get_factor_exposures(date=date, tickers=tickers)
This code runs. But there are several problems with this use of decorators - some small, some more significant. First, if you don’t care about the identity of the wrapper function, we could avoid @wraps. Second, wrapper is actually the new wrapped function. Third, your code will break if you actually pass a parameter to the decorator @apply_ticker_change(col_key="ticker"). Fourth, and more importantly, this breaks Liskov Substitution - the client function should not know the implementation of the decorator (and vice versa). It is better to define your getter as if your decorator doesn’t exist:
# ticker
def apply_ticker_change(ticker_changes:pd.DataFrame, col_key: str = "instrument_id")
def _decorator(func):
def new_f(*args, **kwargs):
results = func(*args, **kwargs)
results[col_key] = results[col_key].replace(ticker_changes)
return results
return new_f
return _decorator
# risk
@apply_ticker_change(ticker_changes)
def get_factor_exposures(
self,
date: str,
tickers: list[str],
) -> DataFrame:
return self.risk_reader.get_factor_exposures(date=date, tickers=tickers)
And if this is not possible within the risk class because it doesn’t have access to ticker_changes. The caller can simply use the same decorator to mutate the function: instead of calling risk.get_factor_exposures(date, tickers), it is calling the mutated version
apply_ticker_change(ticker_changes)(risk.get_factor_exposures)(date, tickers)
Code Challenge
A common pattern to serve dynamically generated datasets is to put them behind a REST api (say on fastapi), where each endpoint returns some dataset asynchronously in a serialized format say JSON.
@router.get("/holdings")
async def _get_holdings():
df = data_provider.get_holdings()
df = some_transformations(df)
return df.to_json()
It’d be nice if we could call get_holdings synchronously in another function in its native return format (say a dataframe) instead, This way you can compose functions like:
@router.get("/weights")
async def _get_weights()
df = _get_holdings()
df = more_transformations(df)
return df.to_json()
We can achieve this by implementing decorators like:
def _async(f):
@wraps(f)
async def async_func(*args, **kwargs):
return f(*args, **kwargs)
return async_func
def converters(df:pd.DataFrame, to_type: str):
if to_type == "json":
return df.to_json()
elif to_type == "html":
return HTMLResponse(df.to_html())
else:
return df
def returns(rtype="json"):
def check_returns(f):
@wraps(f)
def new_f(*args, **kwargs):
result = f(*args, **kwargs)
return converters(result, to_type=rtype)
return new_f
and the original async serialized endpoint can be equivalently written as:
@router.get("/holdings")
@_async
@returns("html")
def portfolio_holdings()
...
Question
Can you write a decorator that combines @_async and @returns("html") into a single decorator @async_returns("html")? (You could combine all three decorators into a single decorator that looks like @_router.get("/holdings", returns="html") but this is an overkill.)
#solution
async_returns = lambda rtype="json": lambda f: _async(returns(rtype)(f))
The decorator overrides the original function. If you want to retain the original function, you’d probably use async_returns as a functional map.
Acknowledgement. Thanks to Nick Irwin for correcting a coding error above and discussions on adjoint decoractors.