如何Hook变量引用自动执行代码，实现Python映射数据每次访问自动刷新

你可以通过伪装成普通字典的动态映射对象实现需求，仅需要修改models.py中原有映射的定义代码，所有引用映射的业务代码完全不需要改动。

核心思路是自定义类模拟字典的全部访问行为，在访问时自动触发数据库查询拉取最新数据，对外暴露的调用接口和原生字典完全一致。

实现方案1：每次访问实时查库（强一致性） #

适合映射对应的表数据量不大（比如代理商表记录数低于1万条）的场景，能保证每次拿到的都是数据库最新数据，兼容跨实例修改、同实例新增数据的实时同步需求。

首先在models.py中替换原有映射的定义：

import db
from sqlalchemy.orm.exc import NoResultFound

class DynamicIdMap:
    def __init__(self, model_cls, key_col="name", value_col="id"):
        self.model_cls = model_cls
        self.key_col = key_col
        self.value_col = value_col

    # 兼容字典[]取值语法
    def __getitem__(self, key):
        try:
            return db.session.query(getattr(self.model_cls, self.value_col)).filter(
                getattr(self.model_cls, self.key_col) == key
            ).scalar()
        except NoResultFound:
            raise KeyError(key)

    # 兼容字典get方法
    def get(self, key, default=None):
        try:
            return self[key]
        except KeyError:
            return default

    # 兼容字典in判断
    def __contains__(self, key):
        return db.session.query(
            db.session.query(self.model_cls).filter(
                getattr(self.model_cls, self.key_col) == key
            ).exists()
        ).scalar()

    # 兼容字典keys()/items()方法，用于填充下拉框等场景
    def keys(self):
        return [k for k, _ in self.items()]

    def items(self):
        return db.session.query(
            getattr(self.model_cls, self.key_col),
            getattr(self.model_cls, self.value_col)
        ).all()

# 替换原有agents_id_map的定义，不需要修改任何其他业务代码
agents_id_map = DynamicIdMap(db.Agent)
# 其他映射可以同理快速定义
# customer_id_map = DynamicIdMap(db.Customer)
# product_id_map = DynamicIdMap(db.Product)

实现方案2：带缓存的动态映射（兼顾性能和一致性） #

如果表数据量较大，每次访问都查库会有性能损耗，可以增加本地缓存和过期机制，同时支持主动刷新缓存：

import time
import db

class CachedDynamicIdMap:
    def __init__(self, model_cls, key_col="name", value_col="id", cache_ttl=30):
        self.model_cls = model_cls
        self.key_col = key_col
        self.value_col = value_col
        self.cache_ttl = cache_ttl  # 缓存有效期，单位秒，设为0则每次都查库
        self._cache = None
        self._cache_update_time = 0

    def _refresh_cache(self):
        # 缓存过期或为空时重新拉取全量数据
        now = time.time()
        if self._cache is None or now - self._cache_update_time > self.cache_ttl:
            # 拉取前清空会话一级缓存，确保拿到最新数据
            db.session.expire_all()
            self._cache = {
                k: v for k, v in db.session.query(
                    getattr(self.model_cls, self.key_col),
                    getattr(self.model_cls, self.value_col)
                ).all()
            }
            self._cache_update_time = now

    def __getitem__(self, key):
        self._refresh_cache()
        return self._cache[key]

    def get(self, key, default=None):
        self._refresh_cache()
        return self._cache.get(key, default)

    def __contains__(self, key):
        self._refresh_cache()
        return key in self._cache

    def items(self):
        self._refresh_cache()
        return self._cache.items()

    # 主动刷新缓存的接口，同实例新增/修改/删除数据后调用可立刻同步
    def flush(self):
        self._cache = None
        self._cache_update_time = 0

# 定义映射
agents_id_map = CachedDynamicIdMap(db.Agent, cache_ttl=60)

同实例内新增代理商后，只需要调用agents_id_map.flush()即可立刻刷新缓存，不需要等待过期时间。

注意事项 #

• 确保SQLAlchemy会话的expire_on_commit参数保持默认的True，避免会话返回旧缓存数据
• 如果对跨实例数据同步的实时性要求极高，建议将cache_ttl设为5秒以内，或者用数据库通知机制触发缓存刷新

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