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 es absolutamente correcto.

IN es como un grupo de X=Y junto con 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 2

Recomendaría echarle 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 3

Si utiliza claves primarias compuestas, puede utilizar 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 es disponible en SQLite (ver doc).

Answer 4

Hay otra forma; Si es razonable esperar que los objetos en cuestión ya estén cargados en la sesión; se les ha accedido 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á ningún consultas para recuperar los objetos; Por otro lado, si se carga más de un pequeño número de esos objetos no, 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 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 5

El código tal como está es completamente correcto. 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 trozos, tales como, ordenar el conjunto completo de trabajo por parte clave principal o por rango de fechas, lo que, a continuación, cargar todo lo necesario para que el trozo localmente en una memoria 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 uso del mundo real caso.

Pero para ilustrar también algunas API de SQLAlchemy, podemos hacer una función que haga el IN para registros que no tenemos y un local para aquellos que hacemos.Aquí es que:

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í es 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