如何在expiringdict元素过期删除前打印对应内容

原生expiringdict没有提供元素过期前的回调钩子，最简洁的实现方式是继承ExpiringDict自定义子类，插入过期回调逻辑：

步骤1：自定义带回调的过期字典类 #

from expiringdict import ExpiringDict

class CallbackExpiringDict(ExpiringDict):
    def __init__(self, max_len, max_age_seconds, expire_callback=None):
        super().__init__(max_len=max_len, max_age_seconds=max_age_seconds)
        # 接收自定义过期回调函数，入参为(key, value)
        self.expire_callback = expire_callback
    
    def _purge_expired(self):
        # 先筛选所有过期的key
        now = self._now()
        expired_keys = [
            key for key, (insert_time, _) in self._ExpiringDict__dict.items()
            if now - insert_time > self.max_age
        ]
        # 过期元素删除前先执行回调
        if self.expire_callback and expired_keys:
            for key in expired_keys:
                self.expire_callback(key, self[key])
        # 调用父类原有逻辑完成删除
        super()._purge_expired()

步骤2：初始化缓存并绑定打印逻辑 #

# 定义过期时执行的打印函数
def print_expired_trip(key, value):
    print(f"行程结束，行程标识：{key}，行程数据：{value}")

# 初始化缓存
self._assets_history = CallbackExpiringDict(
    max_len=100,
    max_age_seconds=10*60,
    expire_callback=print_expired_trip
)

注意事项 #

• expiringdict的过期清理逻辑默认仅在调用get/set等字典访问操作时触发，无后台主动轮询。如果需要准实时触发回调，可以额外启动一个低频率的后台线程，定时调用_assets_history._purge_expired()即可。
• 上述实现仅覆盖超时过期的场景，如果需要处理缓存达到max_len上限时被主动淘汰的元素，可进一步重写__setitem__方法，在淘汰旧元素前插入回调逻辑。

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