我有一个名为“value”的函数，它会进行大量计算...
如果标识符的数据集未更改，则函数的结果始终相同。
一旦数据集的某些标识符发生更改，我想清除缓存，并让函数重新计算它。
看这段代码你可以更好地理解我：
from functools import cached_property
class Test:
identifiers = {}
dataset = an empty object of dataset type
def __init__(self, identifier, ...)
self.identifier = identifier
...
Test.identifiers[identifier] = self
...
@cached_property
def value(self):
result = None
# heavy calculate based on dataset
return result
@classmethod
def get(cls, identifier):
if identifier in cls.identifiers:
return cls.identifiers[identifier]
else:
return cls(identifier, ...)
@classmethod
def update(cls, dataset):
for block in dataset:
# assume there is block['identifier'] in each block
# here i want to clear the cache of value() function
instance = cls.get(block['identifier'])
# clear @cached_property of instance
cls.dataset.append(block)
您可以在 CPython 源代码中读到，Python 3.8 中
cached_property但是如果你只是想完成它而不考虑兼容性，你可以使用
del instance.value（@Blckknght 答案的补充）
如果您有一个 mutable 对象，并且需要 refresh 所有
@cached_property修改现有的 -mutable- 方法，例如，定义
add_element__setattr__每次调用这些方法之一时，都会删除缓存在
self.__dict__@cached_propertiesfrom functools import cached_property
class Test:
datalist: List[int]
@cached_property
def value(self):
result = None
# heavy calculate based on datalist
return result
def add_element(self, new:int)-> None:
# restore cache if calculated
self.__dict__.pop('value', None) # this will delete the cached val if already cached, otherwise do nothing
self.datalist.append(new)
或编辑
__setattr__from functools import cached_property
class Test:
datalist: List[int]
@cached_property
def value(self):
result = None
# heavy calculate based on datalist
return result
def __setattr__(self, name, val):
self.__dict__[name] = val
self.__dict__.pop('value', None)
我提供了一种替代方法，在某些情况下可能有用。 如果需要进行计算的数据集类型是可哈希的，则可以使用常规
functools.cachelru_cache这是我的意思的一个例子：
from functools import lru_cache
class MyClass():
def __init__(self, data):
self.data = data
@property
def slow_attribute(self):
return self._slow_attribute(self.data)
@staticmethod
@lru_cache
def _slow_attribute(data):
# long computation, using data,
# here is just an example
return sum(data)
这里不需要关心何时清除缓存：如果底层数据集发生变化，静态方法自动知道它不能再使用缓存的值。
这有一个额外的好处，如果数据集要恢复到以前使用的状态，查找可能仍然能够使用缓存的值。
这是上面代码的演示：
from time import perf_counter_ns
def print_time_and_value_of_computation(c):
t1 = perf_counter_ns()
val = c.slow_attribute
t2 = perf_counter_ns()
print(f'Time taken: {(t2 - t1)/1000} microseconds')
print(f'Value: {val}')
c = MyClass(range(10_000))
print_time_and_value_of_computation(c)
print_time_and_value_of_computation(c)
print('Changing the dataset!')
c.data = range(20_000)
print_time_and_value_of_computation(c)
print_time_and_value_of_computation(c)
print('Going back to the original dataset!')
c.data = range(10_000)
print_time_and_value_of_computation(c)
返回：
Time taken: 162.074 microseconds
Value: 49995000
Time taken: 2.152 microseconds
Value: 49995000
Changing the dataset!
Time taken: 264.121 microseconds
Value: 199990000
Time taken: 1.989 microseconds
Value: 199990000
Going back to the original dataset!
Time taken: 1.144 microseconds
Value: 49995000
我遇到了这个问题，并在尝试解决它时遇到了这个线程。在我的例子中，数据实际上是不可变的，除了在某些情况下该对象的设置涉及使用属性，并且属性在设置后已过期。 @Pablo 的回答很有帮助，但我希望该过程能够动态重置缓存的所有内容。
这是一个通用示例：
设置和损坏的东西：
from functools import cached_property
class BaseThing:
def __init__(self, *starting_numbers: int):
self.numbers = []
self.numbers.extend(starting_numbers)
@property
def numbers_as_strings(self) -> dict[int, str]:
"""This property method will be referenced repeatedly"""
def process_arbitrary_numbers(self, *arbitrary_numbers: int) -> list[str]:
return [self.numbers_as_strings.get(number) for number in arbitrary_numbers]
def extend_numbers(self, *additional_numbers: int):
self.numbers.extend(additional_numbers)
class BrokenThing(BaseThing):
@cached_property
def numbers_as_strings(self) -> dict[int, str]:
print("Working on:", " ".join(map(str, self.numbers)))
return {number: str(number) for number in self.numbers}
输出：
>>> thing = BrokenThing(1, 2, 3, 4)
>>> thing.process_arbitrary_numbers(1, 3) == ["1", "3"]
Working on: 1 2 3 4
True
>>> thing.extend_numbers(4, 5, 6)
>>> thing.process_arbitrary_numbers(5, 6) == ["5", "6"]
False
@cached_property@propertyclass InefficientThing(BaseThing):
@property
def numbers_as_strings(self) -> dict[int, str]:
print("Working on:", " ".join(map(str, self.numbers)))
return {number: str(number) for number in self.numbers}
输出：
>>> thing = InefficientThing(1, 2, 3)
>>> thing.process_arbitrary_numbers(1, 3) == ["1", "3"]
Working on: 1 2 3
Working on: 1 2 3
True
>>> thing.extend_numbers(4, 5, 6)
>>> thing.process_arbitrary_numbers(5, 6) == ["5", "6"]
Working on: 1 2 3 4 5 6
Working on: 1 2 3 4 5 6
True
解决方案：
class EfficientThing(BaseThing):
def _clear_cached_properties(self):
for name in dir(type(self)):
if isinstance(getattr(type(self), name), cached_property):
print(f"Clearing self.{name}")
vars(self).pop(name, None)
def extend_numbers(self, *additional_numbers: int):
self._clear_cached_properties()
return super().extend_numbers(*additional_numbers)
@cached_property
def numbers_as_strings(self) -> dict[int, str]:
print("Working on:", " ".join(map(str, self.numbers)))
return {number: str(number) for number in self.numbers}
输出：
>>> thing = EfficientThing(1, 2, 3, 4)
>>> thing.process_arbitrary_numbers(1, 3) == ["1", "3"]
Working on: 1 2 3 4
True
>>> thing.extend_numbers(4, 5, 6)
Clearing self.numbers_as_strings
>>> thing.process_arbitrary_numbers(5, 6) == ["5", "6"]
Working on: 1 2 3 4 4 5 6
True
这会循环访问对象父类的所有属性。如果属性的值是
cached_propertyNonepop