将 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.

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

Answer 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

Answer 3

您可以调用

query.limit(None).

来删除之前应用的限制或偏移。

Answer 4

谷歌搜索返回这个问题，以解决应用带有偏移量和限制的 order_by 问题，因此我将在这里回答以解决该主题的谷歌问题

使用排序方法创建自定义 order_by 方法

def order_by(data=[], sqlasc=False, key=''):
    if len(data) > 0:
        arugs = {
          'reverse': sqlasc,
          'key': lambda db_obj: getattr(db_obj, key)
        }
        data.sort(**arugs)
        
        return data
    else:
        return []

print(order_by(data=x, sqlasc=False, key='product'))

False

表示升序

True

表示描述