如何在 Flask-SQLAlchemy 中为 db.session 连接查询分页？

Question

说，我们有以下关系：

• 一个人可以有多个电子邮件地址
• 电子邮件服务提供商可以（显然）提供多个电子邮件地址

所以，这是多对多的关系。我有三个表：电子邮件、提供商和用户。电子邮件有两个用于提供者和用户的外国 ID。

现在，给定一个特定的人，我想打印所有电子邮件提供商及其为该人托管的电子邮件地址（如果存在）。 （如果此人在 Gmail 中没有电子邮件，我仍然希望 Gmail 出现在结果中。我相信否则我只需要左内连接即可解决此问题。）

我想出了如何使用以下子查询来做到这一点（遵循 sqlalchemy 教程）：

email_subq = db.session.query(Emails).\
                filter(Emails.user_id==current_user.id).\
                subquery()

provider_and_email = db.session.query(Provider, email_subq).\
                outerjoin(email_subq, Provider.emails).\
                all()

这没问题（它返回一个 (Provider, user_id, provider_id, email_address) 的 4 元组，我想要的所有信息），但我后来发现这是 不是 使用 Flask BaseQuery 类，所以Flask-SQLAlchemy 提供的pagination 不起作用。显然db.session.query() 不是 Flask-SQLAlchemy Query 实例。

我尝试做Emails.query.outerjoin[...]，但它只返回电子邮件表中的列，尽管我想要提供者信息和电子邮件。

我的问题：我怎样才能用 Flask-SQLAlchemy 做同样的事情，这样我就不必重新实现已经存在的分页？

---

我想此时最简单的选择是实现我自己的分页功能，但我很想知道是否有另一种正确的方法可以做到这一点。

Answer 1

我不确定这是否最终会成为长期解决方案，它并没有直接解决我对不使用 Flask-SQLAlchemy 的 BaseQuery 的担忧，而是完成我想要的最简单的方法是重新实现分页功能。

而且，事实上，使用原始的 Flask-SQLAlchemy 例程很容易做到这一点：

def paginate(query, page, per_page=20, error_out=True):
    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 Pagination(query, page, per_page, total, items)

从第 376 行附近的 paginate 函数修改：https://github.com/mitsuhiko/flask-sqlalchemy/blob/master/flask_sqlalchemy.py

Answer 2

您的问题是如何将 Flask-SQLAlchemy 的分页与常规 SQLAlchemy 查询一起使用。

由于 Flask-SQLAlchemy 的 BaseQuery 对象没有自己的状态，并且是从 SQLAlchemy 的 Query 派生的，并且实际上只是方法的容器，因此您可以使用此 hack：

from flask.ext.sqlalchemy import BaseQuery
def paginate(sa_query, page, per_page=20, error_out=True):
  sa_query.__class__ = BaseQuery
  # We can now use BaseQuery methods like .paginate on our SA query
  return sa_query.paginate(page, per_page, error_out)

使用方法：

@route(...)
def provider_and_email_view(page):
  provider_and_email = db.session.query(...) # any SQLAlchemy query
  paginated_results = paginate(provider_and_email, page)
  return render_template('...', paginated_results=paginated_results)

*编辑：

请小心执行此操作。如另一个答案所示，这实际上只是一种避免复制/粘贴 paginate 函数的方法。请注意，BaseQuery 没有 __init__ 方法。见How dangerous is setting self.__class__ to something else?。

*编辑2：

如果 BaseQuery 有一个 __init__，您可以使用 SA 查询对象构造一个，而不是破解 .__class__。

Answer 3

嘿，我在这里找到了一个快速解决方法：

provider_and_email = Provider.query.with_entities(email_subq).\
            outerjoin(email_subq, Provider.emails).paginate(page, POST_PER_PAGE_LONG, False)

Answer 4

我目前正在使用这种方法：

query = BaseQuery([Provider, email_subq], db.session())

创建我自己的BaseQuery。 db 是 SqlAlchemy 实例。

更新：@afilbert suggests 你也可以这样做：

query = BaseQuery(provider_and_email.subquery(), db.session())

Answer 5

您如何使用 SQLAlchemy 初始化您的应用程序？

可能您当前的 SQLAlchemy 连接与 flask.ext.sqalchemy 无关，而您使用的是原始 sqlalchemy

查看本教程并检查您的导入，它们确实来自 flask.ext.sqlalchemy

http://pythonhosted.org/Flask-SQLAlchemy/quickstart.html#a-minimal-application

Answer 6

您可以尝试使用结果对列表进行分页。

my_list = [my_list[i:i + per_page] for i in range(0, len(my_list), per_page)][page]

Answer 7

我可能是错的，但我认为你的问题可能是 .all()。通过使用它，您将获得一个列表，而不是一个查询对象。

尝试将其关闭，然后像这样将您的查询传递给分页方法（为了清楚起见，我省略了所有子查询详细信息）：

email_query = db.session.query(Emails).filter(**filters)
email_query.paginate(page, per_page)