SQLAlchemy：如何根据计算值使用 order_by？

Question

所以我有以下数据库模型：

class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    nickname = db.Column(db.String(64), nullable=False, unique=True)
    ...
    tasks = db.relationship('Task', backref='author', lazy='dynamic')

    def number_of_tries(self):
        return Task.query.filter_by(user_id=self.id).count()

    def correct_answers(self):
        return Task.query.filter_by(user_id=self.id).filter_by(correct=True).count()

    def percentage(self):
        try:
            return '{:.2%}'.format(self.correct_answers()/self.number_of_tries())
        except ZeroDivisionError:
            return None

    def __repr__(self):
        return '<User %r>' % (self.nickname)


class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    task_id = db.Column(db.Integer)
    correct = db.Column(db.Boolean)
    timestamp = db.Column(db.DateTime)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))

    def __repr__(self):
        return '<%r %s>' % (self.task_id, self.correct)

两个表，第一个存储用户数据，第二个与第一个相关联，存储用户回答问题的数据。

我想根据成功率对用户进行排序，所以我会这样做：

User.query.order_by(User.percentage).all()

但这行不通。我究竟做错了什么？基于这样的方法甚至可以使用 order_by 吗？

Answer 1

我认为 "Hybrid Attributes" 可以解决您的问题。

http://docs.sqlalchemy.org/en/latest/orm/extensions/hybrid.html#module-sqlalchemy.ext.hybrid

更新：

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey

Base = declarative_base()


class User(Base):
    __tablename__ = 'user'

    id = Column(Integer, primary_key=True)
    nickname = Column(String(64), nullable=False, unique=True)
    tasks = relationship('Task', backref='author', lazy='dynamic')

    def number_of_tries(self):
        return self.tasks.count()

    def correct_answers(self):
        return self.tasks.filter_by(correct=True).count()

    def percentage(self):
        try:
            return '{:.2%}'\
                .format(float(self.correct_answers())/self.number_of_tries())
        except ZeroDivisionError:
            return None

    def __repr__(self):
        return '<User %r>' % (self.nickname)


class Task(Base):
    __tablename__ = 'task'

    id = Column(Integer, primary_key=True)
    task_id = Column(Integer)
    correct = Column(Boolean)
    timestamp = Column(DateTime)
    user_id = Column(Integer, ForeignKey('user.id'))

    def __repr__(self):
        return '<%r %s>' % (self.task_id, self.correct)


from sqlalchemy import create_engine
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(bind=engine)

from sqlalchemy.orm import sessionmaker
# create a configured "Session" class
Session = sessionmaker(bind=engine)

# create a Session
session = Session()

from datetime import datetime
task1 = Task(task_id=1, correct=True, timestamp=datetime.now())
task2 = Task(task_id=2, correct=False, timestamp=datetime.now())
task3 = Task(task_id=3, correct=False, timestamp=datetime.now())
task4 = Task(task_id=4, correct=False, timestamp=datetime.now())
task5 = Task(task_id=5, correct=True, timestamp=datetime.now())
task6 = Task(task_id=6, correct=True, timestamp=datetime.now())
user1 = User(nickname="Lekha", tasks=[task1, task2, task3])
user2 = User(nickname="Vanya", tasks=[task4, task5, task6])
session.add(user1, user2)
session.commit()

print(user1.number_of_tries())
print(user1.correct_answers())
print(user1.percentage())

Answer 2

即使使用 SqlAlchemy，您也必须考虑对象集及其值。您想要的查询涉及三个不同的集合：用户、他们的正确答案和他们的总答案。

你想要的是这样的查询（警告，这只是一个示例，你可以写得更好）

select userid, cor_count/ans_count from users
inner join (select userid, count(*) cor_count from answers where correct=true group by userid) as correct_answers on users.userid=correct_answers.userid
inner join (select userid, count(*) as ans_count from answers group by userid) as total_answers on total_answers.userid=users.userid
where users.userid='xxxx'
order by 2

所以，您必须（以某种方式）在 SqlAlchemy 中制定这个。指导方针会产生这样的效果：

ans_q = session.query(Task.user_id, func.count(task.id).label('cnt')).group_by(Task.user_id)
corr_ans_q = session.query(Task.user_id, func.count(task.id).label('cnt')).filter(Task.correct).group_by(Task.user_id)

ans_q = alias(ans_q.selectable)
corr_ans_q = alias(corr_ans_q.selectable)

q = session.query(User).join(ans_q,ans_q.c.user_id==User.id).join(corr_ans_q.c.user_id==User.id).order_by(corr_ans_q.c.cnt/ans_q.c.cnt)

SQLAlchemy：如何根据计算值使用 order_by？

SQLAlchemy: How to use order_by based on a calculated value?

python

sqlalchemy

flask-sqlalchemy