SQLAlchemy выбрать гибридное выражение свойства в виде столбца

У меня есть свойство SQLAlchemy hyrbid, похожее на то, что официальные документы описывают как гибрид взаимосвязанных подзапросов .

Их пример кода выглядит так:

class SavingsAccount(Base):
    __tablename__ = "account"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("user.id"), nullable=False)
    balance = Column(Numeric(15, 5))


class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)

    accounts = relationship("SavingsAccount", backref="owner")

    @hybrid_property
    def balance(self):
        # I don't want to use this method
        return sum(acc.balance for acc in self.accounts)

    @balance.expression
    def balance(cls):
        # This is how I want to calculate the total balance
        return (
            select([func.sum(SavingsAccount.balance)])
            .where(SavingsAccount.user_id == cls.id)
            .label("total_balance")
        )

Если я User.balance список из нескольких пользователей, я бы хотел заполнить значение User.balance основе User.balance чтобы агрегация выполнялась в SQL, а не в Python. Однако я не могу найти способ сделать это.

Самое близкое, что я смог придумать, это:

>>> User.query.add_columns(User.balance).all()
[(<User 1>, Decimal(ཆ.00')), (<User 2>, Decimal(ཋ.00')), (<User 3>, Decimal(ཆ.00'))]

Есть ли другой способ сделать запрос агрегации, который заполняет свойство в моей модели (вместо того, чтобы просто возвращать кортеж? Я посмотрел на column_property но у него меньше примеров и он кажется недостаточно мощным для всех моих column_property использования).

Всего 1 ответ


Я понял это, благодаря этому сообщению групп Google . По сути, мне нужно использовать column_property , но из-за всей логики мне нужно установить этот column_property способом. В идеале я мог бы использовать свойство @declared_attr , но, похоже, SQLAlchemy в данный момент не поддерживается.

В результате нам нужно определить и присоединить его к классу User используя функцию с прослушивателем событий.

class SavingsAccount(ext.db.Model):
    __tablename__ = "test_account"
    id = ext.db.Column(ext.db.Integer, primary_key=True)
    user_id = ext.db.Column(ext.db.Integer, ext.db.ForeignKey("test_user.id"), nullable=False)
    balance = ext.db.Column(ext.db.Numeric(15, 5))


class User(ext.db.Model):
    __tablename__ = "test_user"
    id = ext.db.Column(ext.db.Integer, primary_key=True)
    name = ext.db.Column(ext.db.String(100), nullable=False)
    accounts = ext.db.relationship("SavingsAccount", backref="owner")


@event.listens_for(mapper, "mapper_configured")
def set_thread_count(mapper, cls) -> None:
    if not issubclass(cls, User):
        return

    total_balance_subquery = (
        select([func.sum(SavingsAccount.balance)])
        .where(SavingsAccount.user_id == cls.id)
        .label("total_balance")
    )
    cls.total_balance = column_property(total_balance_subquery, deferred=True)

Уродливо, но это работает. Все равно был бы очень рад услышать более чистое альтернативное решение!