Caching that resets itself under certain circumstances: Reduce boiler plate code

Question

In my machine learning programs I have the typical situation where I need to perform certain operations on data, which results in the following graph:

• data --op A--> intermediate_state --op B--> final result type 1
• data --op A--> intermediate_state --op C--> final result type 2
• data --op A--> intermediate_state --op D--> final result type 3

So in order to not have to repeat operation A all the time I provide a "fit" method, which implements operation A. Then, I implement transformation methods that implement the other operations. However, since I usually only need some of the final results in the graph leaves, I want to calculate those on the one hand lazily, but then cache the result.

Since the operations require a parameter, I provide this one in the constructor. I also need to pass a parametrized object with the same parametrization to different parts of the the program, where the data of operation A changes, without the parametrization changing. That is the reason why fit is it's own method and not part of the constructor. Also, because of that I cannot simply cache the results of the operations that are not A, without resetting them after invoking a new fit.

Here is my result in Python:

class MyClass:       

    def __init__(self, parameter):
        self.parameter = parameter

    # expensive machine learning fitting
    def fit(self, data) -> MyClass:
        self.intermediate_data_ = data + self.parameter + 2
        self._complex_transform1 = None
        self._complex_transform2 = None
        return self

    # expensive machine learning operation version 1
    @property
    def complex_transform1(self):
        if self._complex_transform1 is None:
            self._complex_transform1 = self.intermediate_data_ / self.parameter / 2
        return self._complex_transform1

    # expensive machine learning operation version 2
    @property
    def complex_transform2(self):
        if self._complex_transform2 is None:
            self._complex_transform2 = self.intermediate_data_ / self.parameter / 5
        return self._complex_transform2

The problem I have with this approach is that I am repeating quite a bit of boiler plate code. (Note, that I have many more operations that are not A.) Namely:

1. The variables self._complex_transform1 and self._complex_transform2

2. The constructs if self._complex_transform1 is None: could be reduced from

        if self._complex_transform1 is None:
            self._complex_transform1 = self.intermediate_data_ / self.parameter / 2
        return self._complex_transform1
   

to

        return self.intermediate_data_ / self.parameter / 2

and I would save a nesting.

I was thinking of defining a container class that contains the results and is reset by fit. But I do not see how to implement this without basically introducing the same about of boilerplate code into the transform methods.

Some decorator-approach might be great, but I am not sure how to approach this.

Answer 1

This might be a job for @functools.cached_property. It works a lot like @property, except it remembers the value between calls. The value can then be cleared using del in the fit function, or elsewhere if needed.

Unfortunately, the properties can't be deled until they've been set, which means we have to re-add the boiledplate in fit in the form of try: del x; except AttributeError: pass. Maybe that's an improvement since it's at least centralized, but it's still not ideal.

There are workarounds like

def fit(self, data):
    self.intermediate_data_ = ...

    for prop in ["complex_transform1", "complex_transform2"]:
        try:
            delattr(self, prop)
        except AttributeError:
            pass

    return self

Or perhaps something like

def fit(self, data):
    self.intermediate_data_ = ...

    for prop in self.reset_on_fit:
        delattr(self, prop)

    self.reset_on_fit = []
    return self

with the properties looking like

@cached_property
def complex_transform1(self):
    self.reset_on_fit.append("complex_transform1")
    return self.intermediate_data_ / self.parameter / 2

But putting that data into a hardcoded string is not very refactor-friendly. Perhaps we can define a decorator to figure that out for us?

Well, I haven't tested it very thoroughly, but I think we can:

from functools import cached_property, wraps

def resets_on_fit(fn):
    @wraps(fn)
    def wrapper(self):
        self.reset_on_fit.append(fn.__name__)
        return fn(self)

    return wrapper

class MyClass:
    # ...

    def fit(self, data):
        self.intermediate_data_ = data + self.parameter + 2
        
        for prop in self.reset_on_fit:
            delattr(self, prop)
        
        self.reset_on_fit = []
        return self

    @cached_property
    @resets_on_fit
    def complex_transform1(self):
        return self.intermediate_data_ / self.parameter / 2

Answer 2

One option is to use the code in the following. Not only should the code be very convenient, but also pretty fast after fitting. It was partly inspired by https://github.com/pydanny/cached-property, which also inspired the @cached_property of standard python's functools.

The exciting additional part (and the part that I came up with :-)) is that we do not only decorate the method that is obviously decorated, but also the methods that are given as arguments.

import functools
import types
from typing import Union, Iterable


class cached_resetable_property(object):

    def __init__(self, resetting_methods: Union[str, Iterable[str]]):
        self.resetting_methods = list(resetting_methods) if isinstance(resetting_methods, Iterable) else [resetting_methods]
        self.reset_methods_that_are_already_wrapped = []

    def __call__(self, function):
        self.__doc__ = getattr(function, "__doc__")
        self.function = function
        return self

    def __get__(self, obj, cls):
        
        decorated_function_name = self.function.__name__

        for reset_method_name in self.resetting_methods:
            if reset_method_name not in self.reset_methods_that_are_already_wrapped:
                self.reset_methods_that_are_already_wrapped.append(reset_method_name)
                reset_method = getattr(obj, reset_method_name)
                
                @functools.wraps(reset_method)
                def wrapper(self_wrapper, *args, **kwargs):
                    try:
                        delattr(self_wrapper, decorated_function_name)
                    except AttributeError:
                        print('Twice fit')
                    return reset_method(*args, **kwargs)  # still bound zu obj, that is why not reset_method(self_wrapper, *args, **kwargs) 
                
                setattr(obj, reset_method_name, types.MethodType(wrapper, obj))

        value = obj.__dict__[decorated_function_name] = self.function(obj)
        return value


class MyClass:

    def __init__(self, parameter):
        self.parameter = parameter

    def fit(self, data):
        print('fit')
        self.intermediate_data_ = data + self.parameter + 2
        print(self.__dict__.keys())
        return self

    @cached_resetable_property(['fit'])
    def trans1(self):
        print('trans 1 - calc')
        return self.intermediate_data_ / self.parameter / 2

    @cached_resetable_property(['fit'])
    def trans2(self):
        print('trans 2 - calc')
        return self.intermediate_data_ / self.parameter / 5


myclass = MyClass(2)
myclass.fit(10)
t1_a_f1 = myclass.trans1
t1_b_f1 = myclass.trans1
t2_a_f1 = myclass.trans2
assert t1_a_f1 == 3.5
assert t1_b_f1 == 3.5
assert t2_a_f1 == 1.4

myclass.fit(20)
t1_a_f2 = myclass.trans1
t1_b_f2 = myclass.trans1
t2_a_f2 = myclass.trans2
assert t1_a_f2 == 6
assert t1_b_f2 == 6
assert t2_a_f2 == 2.4

myclass2 = MyClass(3)
assert myclass2.parameter == 3
myclass2.fit(15)
assert myclass2.intermediate_data_ == 15 + 3 + 2
t1_a_f3 = myclass2.trans1
t1_b_f3 = myclass2.trans1
assert t1_a_f3 ==  myclass2.intermediate_data_ / 3 / 2
assert t1_b_f3 ==  myclass2.intermediate_data_ / 3 / 2

t1_c_f2 = myclass.trans1
assert t1_a_f2 == t1_c_f2