Python缓存变量/函数依赖追踪的优化方案问询

1. 更Pythonic的通用解决方案 #

可以实现一个带依赖声明的缓存装饰器，结合可观察的数据源，自动管理缓存失效，无需手动追踪依赖：

实现思路 #

• 定义ObservableData基类，所有被缓存函数依赖的数据源都继承它，内部维护订阅者列表（即依赖该数据的缓存清理方法）。
• 自定义cached_with_deps装饰器，让缓存函数显式声明依赖的ObservableData实例，装饰器自动将函数的缓存清理方法注册为数据源的订阅者。
• 数据源更新时，自动通知所有订阅的缓存函数执行cache_clear()。

代码示例 #

from functools import wraps, lru_cache

class ObservableData:
    def __init__(self):
        self._subscribers = set()

    def subscribe(self, callback):
        self._subscribers.add(callback)

    def unsubscribe(self, callback):
        self._subscribers.discard(callback)

    def notify_update(self):
        for callback in self._subscribers:
            callback()

# 示例：模拟数据库查询结果类
class UserData(ObservableData):
    def __init__(self):
        super().__init__()
        self._data = {}

    def fetch(self, user_id):
        # 模拟慢SQL查询
        return self._data.get(user_id)

    def update(self, user_id, new_data):
        self._data[user_id] = new_data
        # 更新后自动通知清理缓存
        self.notify_update()

def cached_with_deps(*dependencies):
    def decorator(func):
        cached_func = lru_cache(maxsize=None)(func)
        # 注册缓存清理函数为依赖的订阅者
        for dep in dependencies:
            dep.subscribe(cached_func.cache_clear)

        @wraps(func)
        def wrapper(*args, **kwargs):
            return cached_func(*args, **kwargs)

        # 可选：添加取消订阅的方法，避免内存泄漏
        def unsubscribe():
            for dep in dependencies:
                dep.unsubscribe(cached_func.cache_clear)
        wrapper.unsubscribe = unsubscribe

        return wrapper
    return decorator

# 使用示例
user_data = UserData()

@cached_with_deps(user_data)
def get_user_full_name(user_id):
    # 模拟慢计算/查询逻辑
    user = user_data.fetch(user_id)
    return f"{user['first_name']} {user['last_name']}"

# 更新数据时，缓存会自动被清理
user_data.update(1, {"first_name": "Alice", "last_name": "Smith"})
# 此时调用get_user_full_name(1)会重新计算，无需手动清理缓存

这种方案遵循Python“显式优于隐式”的设计原则，依赖关系清晰，完全基于内置特性实现，不需要第三方库。

2. 用随对象变更的引用替代字符串标识 #

直接用对象引用无法解决更新后匹配问题，但可以通过给数据对象绑定唯一不变标识+弱引用管理来实现类似效果：

实现方式 #

• 给每个数据对象分配一个全局唯一的_uid（比如用uuid生成），该标识在对象创建后永不改变，即使内部数据更新。
• 用弱引用字典管理对象与缓存清理函数的关联，避免内存泄漏，同时保证对象更新时能稳定找到对应的缓存。

代码示例（基于ObservableData改造） #

import uuid
import weakref
from functools import wraps, lru_cache

class TrackableData:
    # 弱引用字典：对象 -> 缓存清理函数集合，对象被回收时自动清理条目
    _cache_subscribers = weakref.WeakKeyDictionary()

    def __init__(self):
        self._uid = uuid.uuid4().hex
        self._cache_subscribers[self] = set()

    @property
    def uid(self):
        return self._uid

    def subscribe(self, callback):
        self._cache_subscribers[self].add(callback)

    def unsubscribe(self, callback):
        self._cache_subscribers[self].discard(callback)
        if not self._cache_subscribers[self]:
            del self._cache_subscribers[self]

    def notify_update(self):
        for callback in self._cache_subscribers.get(self, set()):
            callback()

def cached_with_trackable_deps(*dependencies):
    def decorator(func):
        cached_func = lru_cache(maxsize=None)(func)
        for dep in dependencies:
            dep.subscribe(cached_func.cache_clear)

        @wraps(func)
        def wrapper(*args, **kwargs):
            return cached_func(*args, **kwargs)

        def unsubscribe():
            for dep in dependencies:
                dep.unsubscribe(cached_func.cache_clear)
        wrapper.unsubscribe = unsubscribe

        return wrapper
    return decorator

这里的_uid是对象的唯一身份标识，即使对象内部数据更新，标识也不会改变，能稳定关联缓存清理逻辑；弱引用字典则避免了对象被回收后仍占用内存的问题。如果需要替换数据对象（比如重新从数据库加载），只需确保新对象的_uid与旧对象一致，就能继续关联到原有缓存的清理逻辑。

内容的提问来源于stack exchange，提问作者dogAwakeCat