如何复用SQLAlchemy过滤器？多方法多模型场景的实现需求

我有一个类通过多种方法与SQLAlchemy对象交互，执行各类查询任务。部分查询使用相同的过滤器，示例如下：

def get_all_expired(self,my_model):
    q = session.query(my_model).filter(
        my_model.status == 'OK',
        my_model.is_validated == True,
        my_model.expires_at <= plum_dt.now(),
    )
    return q.all()

def update_all_expired(self,my_model):
    session.query(my_model).filter(
        my_model.status == 'OK',
        my_model.is_validated == True,
        my_model.expires_at <= plum_dt.now(),
    ).update({'status':'EXPIRED'})

上述两个方法的查询过滤器完全相同，且该过滤器还在其他方法中使用。我希望将其应用于不同模型，请问能否创建单一可复用的过滤器，统一管理并在需要的查询中复用？

当然可以，以下两种简洁方式可实现可复用的SQLAlchemy过滤器，统一管理过滤逻辑：

方式一：独立过滤函数（最灵活） #

创建单独函数接收模型类作为参数，返回组合好的过滤条件。这种方式无需修改模型定义，适用于任何包含对应字段的模型：

from sqlalchemy import and_
import plum_dt

def expired_filters(model):
    return and_(
        model.status == 'OK',
        model.is_validated == True,
        model.expires_at <= plum_dt.now()
    )

使用示例： #

def get_all_expired(self, my_model):
    q = session.query(my_model).filter(expired_filters(my_model))
    return q.all()

def update_all_expired(self, my_model):
    session.query(my_model).filter(expired_filters(my_model)).update({'status':'EXPIRED'})

方式二：模型Mixin类（面向对象风格） #

如果多个模型都包含status、is_validated、expires_at字段，可创建Mixin类封装字段定义和过滤逻辑，让模型继承后直接复用：

from sqlalchemy import Column, String, Boolean, DateTime, and_
import plum_dt

class ExpirableMixin:
    # 统一定义通用字段
    status = Column(String)
    is_validated = Column(Boolean)
    expires_at = Column(DateTime)

    @classmethod
    def get_expired_filters(cls):
        return and_(
            cls.status == 'OK',
            cls.is_validated == True,
            cls.expires_at <= plum_dt.now()
        )

模型继承示例： #

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class MyModel(Base, ExpirableMixin):
    __tablename__ = 'my_model'
    id = Column(Integer, primary_key=True)
    # 其他自定义字段...

使用示例： #

def get_all_expired(self, my_model):
    q = session.query(my_model).filter(my_model.get_expired_filters())
    return q.all()

注意事项 #

• 确保所有应用该过滤器的模型都包含对应字段，否则会抛出属性不存在的错误。
• 若不同模型字段名有差异，可在过滤函数中添加参数适配（比如传入字段名映射），但建议尽量统一模型字段命名以简化维护。

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