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

Question

我正在使用SQLAlchemy（查询MySQL）设计一个API，并希望强制所有查询都具有page_size（LIMIT）和page_number（OFFSET）参数。

是否有一种简洁的方法在SQLAlchemy中实现这一点？也许可以构建某种工厂来创建自定义查询对象？或者，也许可以使用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.

- Rob Crowell

---

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

Answer 1

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

# 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

- pydsigner

---

通过将其分成两个部分（获取查询对象，应用强制筛选器），我们依赖开发人员记得在所有查询上调用q()。如果有人在调用all()之前忘记调用q()，结果仍将生成。 - Rob Crowell

然后，您必须采用过滤字典而不是查询。但这会在一定程度上限制您。 - pydsigner

@RobCrowell 三年后，我很好奇这个问题是否已经解决了，或者你最终采取了其他措施（如果是这样，那么看到你提交关于你想出的答案会很不错）？ - pydsigner

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

@MarkWinterbottom，从函数定义中的 page=0 可以看出，我正在使用 0 索引。 - pydsigner

Answer 2

在此问题的时间点上不是一个选项，自版本1.0.0起，您可以利用查询事件，以确保limit和offset方法始终在用户使用q函数进行任何操作后，在编译query对象之前被调用：

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

- Daniele

---

在1.4版本之后的SQLAlchemy中，由于已经被弃用，你该如何执行此操作？请参考以下链接：https://docs.sqlalchemy.org/en/14/orm/events.html#sqlalchemy.orm.QueryEvents.before_compile - rickerp

Answer 3

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

- pie