Applying LIMIT and OFFSET to all queries in SQLAlchemy

Question

I'm designing an API with SQLAlchemy (querying MySQL) and I would like to force all my queries to have page_size (LIMIT) and page_number (OFFSET) parameters.

Is there a clean way of doing this with SQLAlchemy? Perhaps building a factory of some sort to create a custom Query object? Or maybe there is a good way to do this with a mixin class?

I tried the obvious thing and it didn't work because .limit() and .offset() must be called after all filter conditions have been applied:

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

When I try using this, I get the exception:

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

Try adding a first, required argument, which must be a group of query filters. Thus,

# 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

or,

# 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

Not an option at the time of this question, since version 1.0.0 you can take advantage of Query events to ensure limit and offset methods are always called just before your query object is compiled, after any manipulation is performed by the users of your q function:

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