python:可能更改的缓存属性中的簿记依赖项

Question

我有一个有三个属性a,b,c的A类,其中a是从b和c计算的(但这很贵).此外,属性b和c可能会随着时间而变化.我想确保:

1. a一旦计算完就被缓存,然后从缓存中再现
2. 如果b或c发生变化,则下次需要时,必须重新计算以反映变化

以下代码似乎有效:

class A():

    def __init__(self, b, c):
        self._a = None
        self._b = b
        self._c = c

    @property
    def a(self):
        if is None:
            self.update_a()
        return self._a

    def update_a(self):
        """
        compute a from b and c
        """
        print('this is expensive')
        self._a = self.b + 2*self.c

    @property
    def b(self):
        return self._b

    @b.setter
    def b(self, value):
        self._b = value
        self._a = None #make sure a is recalculated before its next use

    @property
    def c(self):
        return self._c

    @c.setter
    def c(self, value):
        self._c = value
        self._a = None #make sure a is recalculated before its next use

然而,由于许多原因,这种方法看起来并不是很好:

1. b和c的制定者需要知道a
2. 如果依赖树变大,编写和维护就变得一团糟
3. 它在update_a的代码中可能并不明显,它的依赖关系是什么
4. 它会导致很多代码重复

是否有一种抽象的方法来实现这一目标,并不要求我自己完成所有的簿记工作？理想情况下,我想有一些装饰器告诉属性它的依赖是什么,以便所有的簿记发生在引擎盖下.

我想写:

@cached_property_depends_on('b', 'c')
def a(self):
    return self.b+2*self.c

或类似的东西.

编辑:我希望解决方案不要求分配给a,b,c的值是不可变的.我最感兴趣的是np.arrays和list,但我希望代码可以在许多不同的情况下重用,而不必担心可变性问题.

Answer 1

你可以使用functools.lru_cache：

from functools import lru_cache
from operator import attrgetter

def cached_property_depends_on(*args):
    attrs = attrgetter(*args)
    def decorator(func):
        _cache = lru_cache(maxsize=None)(lambda self, _: func(self))
        def _with_tracked(self):
            return _cache(self, attrs(self))
        return property(_with_tracked, doc=func.__doc__)
    return decorator

这个想法是在每次访问属性时检索跟踪属性的值，将它们传递给记忆可调用，但在实际调用期间忽略它们。

给出该类的最小实现：

class A:

    def __init__(self, b, c):
        self._b = b
        self._c = c

    @property
    def b(self):
        return self._b

    @b.setter
    def b(self, value):
        self._b = value

    @property
    def c(self):
        return self._c

    @c.setter
    def c(self, value):
        self._c = value

    @cached_property_depends_on('b', 'c')
    def a(self):
        print('Recomputing a')
        return self.b + 2 * self.c

a = A(1, 1)
print(a.a)
print(a.a)
a.b = 3
print(a.a)
print(a.a)
a.c = 4
print(a.a)
print(a.a)

输出

from functools import lru_cache
from operator import attrgetter

def cached_property_depends_on(*args):
    attrs = attrgetter(*args)
    def decorator(func):
        _cache = lru_cache(maxsize=None)(lambda self, _: func(self))
        def _with_tracked(self):
            return _cache(self, attrs(self))
        return property(_with_tracked, doc=func.__doc__)
    return decorator

Answer 2

幸运的是，如果您熟悉描述符和元类，这样的依赖管理系统很容易实现。

我们的实施需要四件事：

1. 一种新类型property知道哪些其他属性依赖于它。当该属性的值发生变化时，它将通知所有依赖该属性的属性必须重新计算其值。我们将这个类称为DependencyProperty.
2. 另一种类型DependencyProperty缓存由其 getter 函数计算的值。我们称之为DependentProperty.
3. DependencyMeta将所有 DependentProperties 连接到正确的 DependencyProperties 的元类。
4. 一个函数装饰器@cached_dependent_property，将 getter 函数转换为DependentProperty.

这是实现：

_sentinel = object()


class DependencyProperty(property):
    """
    A property that invalidates its dependencies' values when its value changes
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.dependent_properties = set()

    def __set__(self, instance, value):
        # if the value stayed the same, do nothing
        try:
            if self.__get__(instance) is value:
                return
        except AttributeError:
            pass

        # set the new value
        super().__set__(instance, value)

        # invalidate all dependencies' values
        for prop in self.dependent_properties:
            prop.cached_value = _sentinel

    @classmethod
    def new_for_name(cls, name):
        name = '_{}'.format(name)

        def getter(instance, owner=None):
            return getattr(instance, name)

        def setter(instance, value):
            setattr(instance, name, value)

        return cls(getter, setter)


class DependentProperty(DependencyProperty):
    """
    A property whose getter function depends on the values of other properties and
    caches the value computed by the (expensive) getter function.
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.cached_value = _sentinel

    def __get__(self, instance, owner=None):
        if self.cached_value is _sentinel:
            self.cached_value = super().__get__(instance, owner)

        return self.cached_value


def cached_dependent_property(*dependencies):
    """
    Method decorator that creates a DependentProperty
    """
    def deco(func):
        prop = DependentProperty(func)
        # we'll temporarily store the names of the dependencies.
        # The metaclass will fix this later.
        prop.dependent_properties = dependencies
        return prop
    return deco


class DependencyMeta(type):
    def __new__(mcls, *args, **kwargs):
        cls = super().__new__(mcls, *args, **kwargs)

        # first, find all dependencies. At this point, we only know their names.
        dependency_map = {}
        dependencies = set()
        for attr_name, attr in vars(cls).items():
            if isinstance(attr, DependencyProperty):
                dependency_map[attr] = attr.dependent_properties
                dependencies.update(attr.dependent_properties)
                attr.dependent_properties = set()

        # now convert all of them to DependencyProperties, if they aren't
        for prop_name in dependencies:
            prop = getattr(cls, prop_name, None)
            if not isinstance(prop, DependencyProperty):
                if prop is None:
                    # it's not even a property, just a normal instance attribute
                    prop = DependencyProperty.new_for_name(prop_name)
                else:
                    # it's a normal property
                    prop = DependencyProperty(prop.fget, prop.fset, prop.fdel)
                setattr(cls, prop_name, prop)

        # finally, inject the property objects into each other's dependent_properties attribute
        for prop, dependency_names in dependency_map.items():
            for dependency_name in dependency_names:
                dependency = getattr(cls, dependency_name)
                dependency.dependent_properties.add(prop)

        return cls

最后，一些证明它确实有效的证据：

class A(metaclass=DependencyMeta):
    def __init__(self, b, c):
        self.b = b
        self.c = c

    @property
    def b(self):
        return self._b

    @b.setter
    def b(self, value):
        self._b = value + 10

    @cached_dependent_property('b', 'c')
    def a(self):
        print('doing expensive calculations')
        return self.b + 2*self.c


obj = A(1, 4)
print('b = {}, c = {}'.format(obj.b, obj.c))
print('a =', obj.a)
print('a =', obj.a) # this shouldn't print "doing expensive calculations"
obj.b = 0
print('b = {}, c = {}'.format(obj.b, obj.c))
print('a =', obj.a) # this should print "doing expensive calculations"