python - tables - sqlalchemy multiple rows were found for one()

Aplicando LIMIT y OFFSET a todas las consultas en SQLAlchemy

Question

Estoy diseñando una API con SQLAlchemy (consultando MySQL) y me gustaría forzar que todas mis consultas tengan los parámetros page_size (LIMIT) y page_number (OFFSET).

¿Hay una manera limpia de hacer esto con SQLAlchemy? ¿Tal vez construir una fábrica de algún tipo para crear un objeto de consulta personalizado? ¿O tal vez hay una buena manera de hacer esto con una clase mixta?

Intenté lo obvio y no funcionó porque se debe llamar a .limit () y .offset () después de que se hayan aplicado todas las condiciones de filtro:

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

Cuando intento usar esto, obtengo la excepción:

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

Intente agregar un primer argumento obligatorio, que debe ser un grupo de filtros de consulta. Así,

# 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

o,

# 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

No es una opción en el momento de esta pregunta, ya que la versión 1.0.0 puede aprovechar los eventos de Consulta para garantizar que offset métodos de limit y offset siempre se llamen justo antes de que se compile su objeto de query , después de que los usuarios de su q realizan la manipulación. función:

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