如何在SQLAlchemy 2.0模型中实现类似Rails ActiveRecord的查询构建器？

当然可以实现类似Rails ActiveRecord的链式查询功能，SQLAlchemy 2.0提供了多种方式在模型内封装查询片段，以下是两种常用方案：

方案一：带参数的类方法（灵活叠加条件） #

通过给类方法传入可选的查询对象参数，实现条件的叠加组合：

from sqlalchemy import select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Shirt(DeclarativeBase):
    __tablename__ = "shirts"
    
    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    color: Mapped[str]
    dry_clean_only: Mapped[bool]

    @classmethod
    def red(cls, query=None):
        # 若传入已有查询则基于其叠加条件，否则创建基础查询
        base_query = query if query is not None else select(cls)
        return base_query.where(cls.color == "red")

    @classmethod
    def dry_clean_only(cls, query=None):
        base_query = query if query is not None else select(cls)
        return base_query.where(cls.dry_clean_only == True)

使用方式 #

# 查询所有红色衬衫
red_shirts = session.execute(Shirt.red()).scalars().all()

# 查询所有需要干洗的红色衬衫（叠加条件）
dry_clean_red_shirts = session.execute(Shirt.dry_clean_only(Shirt.red())).scalars().all()

方案二：自定义查询类（贴近Rails链式语法） #

通过继承Select类并添加自定义条件方法，实现和Rails几乎一致的链式调用体验：

from sqlalchemy import Select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# 自定义查询类，封装Shirt的查询条件方法
class ShirtQuery(Select):
    def red(self):
        return self.where(Shirt.color == "red")
    
    def dry_clean_only(self):
        return self.where(Shirt.dry_clean_only == True)

class Shirt(DeclarativeBase):
    __tablename__ = "shirts"
    
    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    color: Mapped[str]
    dry_clean_only: Mapped[bool]

    @classmethod
    def query(cls):
        # 返回自定义的查询对象
        return ShirtQuery(cls)

使用方式 #

# 链式调用查询条件，完全贴近Rails的写法
results = session.execute(Shirt.query().red().dry_clean_only()).scalars().all()

注意事项 #

• SQLAlchemy的查询对象是不可变的，每次链式调用都会返回一个新的查询实例，这和Rails ActiveRecord的scope行为一致。
• 最终执行查询需要通过session.execute()，再通过scalars()或all()获取结果集，这是SQLAlchemy和Rails的核心差异点。

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