如何在SQLAlchemy中自动隐式为查询添加有效期过滤？

1. 快速实现：直接扩展Session类 #

最简单的方式是给SQLAlchemy的Session类新增一个query_latest方法，内部自动完成有效期过滤：

from sqlalchemy import func
from sqlalchemy.orm import Session

def query_latest(self, *entities, **kwargs):
    query = self.query(*entities, **kwargs)
    # 遍历查询的所有实体，自动为带valid_until字段的模型添加过滤条件
    for desc in query.column_descriptions:
        model = desc["type"]
        if hasattr(model, "valid_until"):
            query = query.filter(model.valid_until > func.getdate())
    return query

# 给Session类挂载这个自定义方法
Session.query_latest = query_latest

2. 进阶：自定义Query类（复用过滤逻辑） #

如果需要后续扩展更多有效期相关的查询方法（比如过滤历史数据），可以自定义Query子类：

from sqlalchemy.orm import Query

class ValidPeriodQuery(Query):
    def filter_latest(self):
        """自动过滤出当前有效的数据"""
        for desc in self.column_descriptions:
            model = desc["type"]
            if hasattr(model, "valid_until"):
                self = self.filter(model.valid_until > func.getdate())
        return self

# 若使用Flask-SQLAlchemy，可通过配置指定该Query类：
# app.config['SQLALCHEMY_QUERY_CLASS'] = ValidPeriodQuery
# 之后可直接调用：db.query(Foo).filter_latest()

3. 使用示例 #

# 单表查询
rs = db.query_latest(Foo)
# 等价于 db.query(Foo).filter(Foo.valid_until > func.getdate())

# 多表查询
rs = db.query_latest(Foo, Bar)
# 等价于 db.query(Foo, Bar).filter(Foo.valid_until > func.getdate(), Bar.valid_until > func.getdate())

关键优势 #

• 完全增量扩展，不修改原有模型定义，也不影响原有query()方法的正常使用
• 自动适配所有带valid_until字段的模型，无需手动为每个表编写过滤条件
• 代码量极少，实现逻辑简单，符合低耗时的需求
• 多表查询时会自动为每个涉及的表添加过滤规则，省去手动拼接条件的麻烦

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