python - queries - sqlalchemy, convirtiendo una lista de ID en una lista de objetos

sqlalchemy documentation

Question

Tengo una secuencia de ID que quiero recuperar. Es sencillo:

session.query(Record).filter(Record.id.in_(seq)).all()

Hay una mejor manera de hacerlo?

Answer 1

Recomiendo echar un vistazo al SQL que produce. Simplemente puede imprimir str (consulta) para verlo.

No estoy al tanto de una forma ideal de hacerlo con SQL estándar.

Answer 2

Tu código está absolutamente bien.

IN es como un grupo de X=Y unido a OR y es bastante rápido en las bases de datos contemporáneas.

Sin embargo, si su lista de ID es larga, puede hacer que la consulta sea un poco más eficiente al pasar una sub consulta que devuelve la lista de ID.

Answer 3

Hay otra forma; Si es razonable esperar que los objetos en cuestión ya estén cargados en la sesión; ya accedió a ellos en la misma transacción, puede hacer lo siguiente:

map(session.query(Record).get, seq)

En el caso de que esos objetos ya estén presentes, esto será mucho más rápido, ya que no habrá consultas para recuperar esos objetos; Por otro lado, si no se carga más de un pequeño número de esos objetos, será mucho, mucho más lento, ya que causará una consulta por instancia faltante, en lugar de una única consulta para todos los objetos.

Esto puede ser útil cuando realiza consultas de joinedload() antes de llegar al paso anterior, por lo que puede estar seguro de que ya se han cargado. En general, debe usar la solución en la pregunta de manera predeterminada, y solo explorar esta solución cuando haya visto que está consultando los mismos objetos una y otra vez.

Answer 4

Si usa claves primarias compuestas, puede usar tuple_ , como en

from sqlalchemy import tuple_
session.query(Record).filter(tuple_(Record.id1, Record.id2).in_(seq)).all()

Tenga en cuenta que esto no está disponible en SQLite (ver documento ).

Answer 5

El código tal como está completamente bien. Sin embargo, alguien me está pidiendo algún sistema de cobertura entre los dos enfoques de hacer una gran IN vs. usar get () para identificaciones individuales.

Si alguien realmente está tratando de evitar el SELECCIONAR, la mejor manera de hacerlo es configurar los objetos que necesita en la memoria con anticipación. Por ejemplo, estás trabajando en una gran tabla de elementos. Divida el trabajo en fragmentos, como ordenar el conjunto completo de trabajo por clave principal o por rango de fechas, lo que sea, luego cargue todo para ese fragmento localmente en un caché:

 all_ids = [<huge list of ids>]

 all_ids.sort()
 while all_ids:
     chunk = all_ids[0:1000]

     # bonus exercise!  Throw each chunk into a multiprocessing.pool()!
     all_ids = all_ids[1000:]

     my_cache = dict(
           Session.query(Record.id, Record).filter(
                 Record.id.between(chunk[0], chunk[-1]))
     )

     for id_ in chunk:
         my_obj = my_cache[id_]
         <work on my_obj>

Ese es el caso de uso del mundo real.

Pero para ilustrar también algunas API de SQLAlchemy, podemos hacer una función que haga la RI para registros que no tenemos y un local para aquellos que hacemos. Aquí está eso:

from sqlalchemy import inspect


def get_all(session, cls, seq):
    mapper = inspect(cls)
    lookup = set()
    for ident in seq:
        key = mapper.identity_key_from_primary_key((ident, ))
        if key in session.identity_map:
            yield session.identity_map[key]
        else:
            lookup.add(ident)
    if lookup:
        for obj in session.query(cls).filter(cls.id.in_(lookup)):
            yield obj

Aquí hay una demostración:

from sqlalchemy import Column, Integer, create_engine, String
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import declarative_base
import random

Base = declarative_base()


class A(Base):
    __tablename__ = ''a''
    id = Column(Integer, primary_key=True)
    data = Column(String)

e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

ids = range(1, 50)

s = Session(e)
s.add_all([A(id=i, data=''a%d'' % i) for i in ids])
s.commit()
s.close()

already_loaded = s.query(A).filter(A.id.in_(random.sample(ids, 10))).all()

assert len(s.identity_map) == 10

to_load = set(random.sample(ids, 25))
all_ = list(get_all(s, A, to_load))

assert set(x.id for x in all_) == to_load