如何通过 SQLAlchemy 中的自定义函数进行排序

Question

所以我有一个 SQLALchemy 模型，如下所示

from sqlalchemy import (create_engine, Column, BigInteger, String, 

                        DateTime)

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.ext.hybrid import hybrid_property

Base = declarative_base()

class Trades(Base):

    __tablename__ = 'trades'

    row_id = Column(BigInteger, primary_key=True, autoincrement=True)

    order_id = Column(String)

    time = Column(DateTime)

    event_type = Column(String)

    @hybrid_property

    def event_type_to_integer(self):

        return dict(received=0, open=1, done=2)[self.event_type]

    @event_type_to_integer.expression

    def event_type_to_integer(self):

        pass

我希望能够在timethen by 之前先订购查询event_type。按时间排序很容易，因为日期时间具有自然排序。然而通过排序event_type是有点棘手，因为event_type可以利用的价值received，open和done。我希望我的所有查询都按照event_type上述指定的顺序对查询进行排序。似乎我需要使用混合属性，这是我在上面开始做的，但是为了使该order_by功能正常工作，我似乎还需要编写

    @event_type_to_integer.expression

    def event_type_to_integer(self):

        pass

功能。这是我画一个空白的地方。有没有人有关于如何编写这个函数来完成上述操作的建议。我试过阅读文档和类似的 StackOverflow 帖子。还是有问题。以供参考。这是我试图开始工作的查询

    sess = Session()

    orders = (

        sess

        .query(Trades)

        .order_by(Trades.time.asc(), Trades.event_type_to_integer.asc())

        .all()

        )

    sess.close()

它抛出一个

KeyError: <sqlalchemy.orm.attributes.InstrumentedAttribute object at 0x7fcb11861048>

Answer 1

您可以使用SQL 中的CASE表达式来实现您的查找：from sqlalchemy import case_event_type_lookup = dict(received=0, open=1, done=2)class Trades(Base):&nbsp; &nbsp; ...&nbsp; &nbsp; @hybrid_property&nbsp; &nbsp; def event_type_to_integer(self):&nbsp; &nbsp; &nbsp; &nbsp; return _event_type_lookup[self.event_type]&nbsp; &nbsp; @event_type_to_integer.expression&nbsp; &nbsp; def event_type_to_integer(cls):&nbsp; &nbsp; &nbsp; &nbsp; return case(_event_type_lookup, value=cls.event_type)这使用构造的value简写case()来生成一个表达式，该表达式将给定的列表达式与字典中传递的键进行比较，从而产生映射值作为结果。