sqlalchemy, convirtiendo una lista de ID en una lista de objetos

Question

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

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

¿Hay una mejor manera de hacerlo?

Answer 1

Su 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 subconsulta que devuelve la lista de ID.

Answer 2

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

Si alguien realmente está tratando de evitar SELECT, entonces 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, por ejemplo, ordene 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 el IN para los registros que no tenemos y un local para los que tenemos. 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

Answer 3

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 (consulte doc ).

Answer 4

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

No conozco una forma ideal de hacerlo con SQL estándar.

Answer 5

Hay otra forma; Si es razonable esperar que los objetos en cuestión ya estén cargados en la sesión; ha accedido a ellos antes en la misma transacción, en su lugar puede hacer:

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á ninguna consulta para recuperar esos objetos; Por otro lado, si se carga no más de un pequeño número de esos objetos, será mucho, mucho más lento, ya que provocará una consulta por instancia faltante, en lugar de una sola consulta para todos objetos.

Esto puede ser útil cuando realiza consultas joinload () 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 forma predeterminada, y solo explorar esta solución cuando haya visto que está consultando los mismos objetos una y otra vez.