代码之家 › 专栏 › 技术社区 › nimaxin

如何在哪里或过滤方法中使用SQLAlchemy hybrid_property?

orm sqlalchemy python

nimaxin · 技术社区 · 1 年前

我使用的是SQLAlchemy v2,有两个模型: User 和 Transaction 在这里 交易 指的是用户之间的货币交易,而不是数据库交易。我在中定义了一个混合属性 使用者 该模型根据每个用户的入金和出金交易来计算他们的余额。然而,当我尝试在中使用此混合属性时遇到了问题 where 或 filter 用于根据余额查询用户。

以下是我如何定义我的模型和混合属性:

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")
            )
        )

以下是我运行的代码,用于查询余额大于1的用户(未按预期工作):

engine = create_engine("sqlite:///db.db")
Session = sessionmaker(engine)

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

运行上述cd后,遇到以下错误:

Traceback (most recent call last):
  File "c:\Users\Nima\Desktop\userbalance\main.py", line 111, in <module>
    stmt = select(User).where(User.balance > 1)
                              ^^^^^^^^^^^^^^^^
  File "C:\Users\Nima\AppData\Local\Programs\Python\Python311\Lib\site-packages\sqlalchemy\sql\operators.py", line 629, in __gt__
    return self.operate(gt, other)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Nima\AppData\Local\Programs\Python\Python311\Lib\site-packages\sqlalchemy\orm\attributes.py", line 453, in operate
    return op(self.comparator, *other, **kwargs)  # type: ignore[no-any-return]  # noqa: E501
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Nima\AppData\Local\Programs\Python\Python311\Lib\site-packages\sqlalchemy\sql\operators.py", line 629, in __gt__
    return self.operate(gt, other)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Nima\AppData\Local\Programs\Python\Python311\Lib\site-packages\sqlalchemy\ext\hybrid.py", line 1509, in operate
    return op(self.expression, *other, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: '>' not supported between instances of 'Select' and 'int'

1 回复 | 直到 1 年前

Parman M. Alizadeh 1 年前

这个 User.balance 混合属性返回 Select 中定义的对象 _balance_expression 作用 选择 是一个子查询,而SQLAlchemy不知道如何将该子查询与整数值进行比较( 1 在您的情况下)。

要解决此问题,您需要修改 _balance_expression 方法返回标量值,而不是子查询。你可以这样做:

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