flask - 如何在 Flask-SqlAlchemy 中使用子查询？

Question

query_1 = db\
    .Query([UserModel, func.count(FriendModel.friend_id)])\
    .select_from(UserModel)\
    .outerjoin(FriendModel, and_(UserModel.id==FriendModel.user_id))\
    .group_by(FriendModel.user_id)
  s_1 = query_1.subquery('s_1')
  print s_1.c.id
  query_2 = db\
    .Query(FriendModel)\
    .select_from(FriendModel)\
    .outerjoin(s_1, FriendModel.user_id==s_1.c.id)

帮助将两个查询放在一起。
https://gist.github.com/vlikin/17d53440eeef7f4147b2

I receive such errors:
InvalidRequestError: SQL expression, column, or mapped entity expected - got '[<sqlalchemy.orm.attributes.InstrumentedAttribute object at 0x7f2e28e1b810>, <sqlalchemy.sql.functions.count at 0x7f2e236c3ed0; count>]'

或者

AttributeError: Neither ‘count’ object nor ‘Comparator’ object has an attribute ‘_autoflush’

我也有问题，因为 db.Query 而不是 db.session.query。我想使用 db.Query 因为它有分页:)

谢谢你。

Answer 1

我遇到了完全相同的问题。

我对这个问题的解决方案是使用 db.session.query直接创建 flask_sqlalchemy.Pagination 的新子类.该子类包含一个方法，允许我创建一个 Pagination来自任何查询的对象。

from flask import abort
from flask_sqlalchemy import Pagination as _Pagination

class Pagination(_Pagination):
    """Extend Pagination class to let us construct it from any BaseQuery."""
    @classmethod
    def from_query(cls, query, page, per_page, error_out=True):
        """Returns `per_page` items from page `page`.  By default it will
        abort with 404 if no items were found and the page was larger than
        1.  This behavor can be disabled by setting `error_out` to `False`.

        This is basically a copy of flask_sqlalchemy.Query.paginate()

        Returns an :class:`Pagination` object.
        """
        if error_out and page < 1:
            abort(404)
        items = query.limit(per_page).offset((page - 1) * per_page).all()
        if not items and page != 1 and error_out:
            abort(404)

        # No need to count if we're on the first page and there are fewer
        # items than we expected.
        if page == 1 and len(items) < per_page:
            total = len(items)
        else:
            total = query.order_by(None).count()

        return cls(query, page, per_page, total, items)

现在根据以下代码段使用它:

query = db.session.query(…)
pagination = Pagination.from_query(query, page, 20)

首先的问题是创建 Pagination的主要逻辑-来自 db.Query 的对象不是 Pagination 类的一部分，而是在 db.Query.paginate() 中实现.