将 LIMIT 和 OFFSET 应用于 SQLAlchemy 中的所有查询

Posted 2023-02-19

【中文标题】将 LIMIT 和 OFFSET 应用于 SQLAlchemy 中的所有查询

【英文标题】：Applying LIMIT and OFFSET to all queries in SQLAlchemy

【发布时间】：2012-10-26 20:56:11

【问题描述】：

我正在使用 SQLAlchemy 设计一个 API（查询 mysql），我想强制我的所有查询都具有 page_size (LIMIT) 和 page_number (OFFSET) 参数。

有没有用 SQLAlchemy 做到这一点的干净方法？也许建立某种工厂来创建自定义 Query 对象？或者也许有一个使用 mixin 类的好方法？

我尝试了明显的方法，但没有成功，因为必须在应用所有过滤条件后调用 .limit() 和 .offset()：

def q(page=0, page_size=None):
    q = session.query(...)
    if page_size: q = q.limit(page_size)
    if page: q = q.offset(page*page_size)
    return q

当我尝试使用它时，我得到了异常：

sqlalchemy.exc.InvalidRequestError: Query.filter() being called on a Query which already has LIMIT or OFFSET applied. To modify the row-limited results of a  Query, call from_self() first.  Otherwise, call filter() before limit() or offset() are applied.

【问题讨论】：

请不要在您的问题中编辑解决方案。相反，请将其作为单独的答案发布在下面。

【参考方案1】：

您可以调用query.limit(None). 来移除之前应用的限制或偏移量。

【参考方案2】：

在这个问题的时候不是一个选项，因为版本 1.0.0 你可以利用 Query events 来确保 limit 和 offset 方法总是在你的 query 对象被编译之前被调用，之后任何操作都由q 函数的用户执行：

from sqlalchemy.event import listen


def q(page=0, page_size=None):
    query = session.query()
    listen(query, 'before_compile', apply_limit(page, page_size), retval=True)
    return query

def apply_limit(page, page_size):
    def wrapped(query):
        if page_size:
            query = query.limit(page_size)
            if page:
                query = query.offset(page * page_size)
        return query
    return wrapped

【讨论】：

如何在 1.4 版 sqlalchemy 中执行此操作，因为它已被弃用 docs.sqlalchemy.org/en/14/orm/…

【参考方案3】：

尝试添加第一个必需参数，该参数必须是一组查询过滤器。因此，

# q('id': 5, 2, 50)
def q(filters, page=0, page_size=None):
    query = session.query(...).filter_by(**filters)
    if page_size:
        query = query.limit(page_size)
    if page: 
        query = query.offset(page*page_size)
    return query

或者，

# q(Model.id == 5, 2, 50)
def q(filter, page=0, page_size=None):
    query = session.query(...).filter(filter)
    if page_size:
        query = query.limit(page_size)
    if page: 
        query = query.offset(page*page_size)
    return query

【讨论】：

通过将其分成两部分（获取查询对象，应用强制过滤器），我们依靠开发人员记住在他的所有查询上调用 q()。如果有人在调用 all() 之前忘记调用 q()，仍然会生成结果。

那么您必须使用过滤字典，而不是查询。但这会在一定程度上限制你。

@RobCrowell 三年后，我很好奇这是否解决了您的问题，或者您是否最终做了其他事情（在这种情况下，很高兴看到您提交有关您来的问题的答案跟上）？

不应该是query = query.offset((page*page_size) - page_size)吗？

@MarkWinterbottom 从函数定义中的page=0 可以看出，我使用的是0-indexing。