SQLAlchemy v2：如何在where/filter中使用hybrid_property查询用户余额？

问题场景 #

使用SQLAlchemy v2定义User和Transaction模型，通过hybrid_property计算用户余额，但尝试在where/filter中用该属性查询时触发错误：

TypeError: '>' not supported between instances of 'Select' and 'int'

原模型与查询代码如下：

from __future__ import annotations

from decimal import Decimal
from typing import List

from sqlalchemy import ForeignKey, SQLColumnExpression, create_engine, func, select
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    relationship,
    sessionmaker,
)

class Base(DeclarativeBase):
    pass

class Transaction(Base):
    __tablename__ = "transactions"
    id: Mapped[int] = mapped_column(primary_key=True)

    amount: Mapped[float]
    sender_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    recipient_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    sender: Mapped[User] = relationship(foreign_keys=[sender_id])
    recipient: Mapped[User] = relationship(foreign_keys=[recipient_id])

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)

    sent_transactions: Mapped[List[Transaction]] = relationship(
        foreign_keys=[Transaction.sender_id], overlaps="sender"
    )
    received_transactions: Mapped[List[Transaction]] = relationship(
        foreign_keys=[Transaction.recipient_id], overlaps="recipient"
    )

    @hybrid_property
    def balance(self) -> Decimal:
        incoming = sum(txn.amount for txn in self.received_transactions)
        outgoing = sum(txn.amount for txn in self.sent_transactions)
        balance = incoming - outgoing
        return balance

    @balance.inplace.expression
    @classmethod
    def _balance_expression(cls) -> SQLColumnExpression[Decimal]:
        return select(
            (
                func.coalesce(
                    select(func.sum(Transaction.amount))
                    .where(Transaction.recipient_id == 1)
                    .scalar_subquery(),
                    0,
                )
                - func.coalesce(
                    select(func.sum(Transaction.amount))
                    .where(Transaction.sender_id == 1)
                    .scalar_subquery(),
                    0,
                ).label("balance")
            )
        )

# 查询代码
engine = create_engine("sqlite:///db.db")
Session = sessionmaker(engine)

with Session() as session:
    stmt = select(User).where(User.balance > 1)
    session.execute(stmt)

错误原因 #

1. 硬编码ID：表达式中固定使用Transaction.recipient_id == 1和Transaction.sender_id == 1，无法动态关联当前查询的用户ID，不符合通用查询需求。
2. 外层包裹select：返回的是select()对象而非可直接比较的列表达式，导致无法与数值执行>等比较操作。

修正方案 #

修改_balance_expression方法，去掉外层select()包裹，并将硬编码ID替换为cls.id关联当前用户：

class User(Base):
    # ... 其他代码保持不变 ...

    @balance.inplace.expression
    @classmethod
    def _balance_expression(cls) -> SQLColumnExpression[Decimal]:
        return (
            func.coalesce(
                select(func.sum(Transaction.amount))
                .where(Transaction.recipient_id == cls.id)
                .scalar_subquery(),
                0,
            )
            - func.coalesce(
                select(func.sum(Transaction.amount))
                .where(Transaction.sender_id == cls.id)
                .scalar_subquery(),
                0,
            )
        ).label("balance")

验证查询 #

修正后重新执行查询代码，即可正常筛选余额大于1的用户：

with Session() as session:
    stmt = select(User).where(User.balance > 1)
    results = session.execute(stmt).scalars().all()
    for user in results:
        print(f"用户ID: {user.id}, 余额: {user.balance}")

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