如何让自定义查询方法复用SQLAlchemy内置的relationship缓存机制

方案1：优先使用自定义Relationship（最推荐） #

如果你的查询逻辑可以通过Relationship的参数表达，直接自定义带过滤/排序规则的Relationship即可，原生就兼容SQLAlchemy的所有缓存、过期、刷新机制，不需要额外写缓存逻辑：

from sqlalchemy import and_
from sqlalchemy.orm import relationship

class User(Model):
    id = ...
    # 示例：只关联已上架、未删除的书籍，按创建时间倒序
    books = relationship(
        'Book',
        primaryjoin=and_(
            User.id == Book.user_id,
            Book.is_published == True,
            Book.is_deleted == False
        ),
        order_by=Book.created_at.desc(),
        lazy='select' # 和默认Relationship加载策略一致，可自行修改
    )

这种写法的行为和你默认定义的普通Relationship完全一致：首次访问user.books触发查询，后续访问直接返回缓存结果，当实例被expire/refresh/flush操作标记为过期后，下一次访问会自动重新查询数据库。

方案2：复杂查询逻辑用属性集成SA状态管理 #

如果你的查询逻辑非常复杂，无法用Relationship参数实现（比如需要多表关联、动态计算过滤条件等），可以把查询结果存到SQLAlchemy托管的实例字典中，SA的生命周期操作会自动清理缓存：

from sqlalchemy.orm import object_session

class User(Model):
    id = ...

    @property
    def books(self):
        # 检查缓存是否存在且未被SA标记为过期
        cache_key = '_cached_books'
        if (
            cache_key in self.__dict__
            and cache_key not in self._sa_instance_state.expired_attributes
        ):
            return self.__dict__[cache_key]
        # 执行自定义查询
        session = object_session(self)
        res = session.query(Book).filter(Book.user_id == self.id).all()
        # 存入SA托管的实例字典，过期时会被自动清理
        self.__dict__[cache_key] = res
        return res

注意不要用独立的自定义私有变量存缓存，SA的expire、refresh等操作不会追踪你自定义的私有变量，只有存在__dict__中的属性才会被SA的状态管理机制识别，过期时自动清空触发重新查询。

方案3：SQLAlchemy 1.4+ 官方query_provider装饰器 #

1.4及以上版本可以用官方提供的query_provider装饰器，完全集成到SA的属性加载体系，自动处理缓存、过期逻辑：

from sqlalchemy.orm import query_provider

class User(Model):
    id = ...

    @query_provider
    def books(cls):
        def load(context):
            return context.session.query(Book).filter(
                Book.user_id == context.identity[0]
            )
        return load

内容的提问来源于stack exchange，提问作者Cyril N.