Sqlalchemy, transformando uma lista de IDs para uma lista de objetos

Question

Eu tenho sequência de IDs que quero recuperar. É simples:

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

Existe uma maneira melhor de fazer isso?

Answer 1

Seu código é absolutamente bom.

IN é como um monte de X=Y juntou -se a OR e é muito rápido nos bancos de dados contemporâneos.

No entanto, se sua lista de IDs for longa, você poderá tornar a consulta um pouco mais eficiente ao passar por um sub-quadro retornando a lista de IDs.

Answer 2

O código como está é completamente bom. No entanto, alguém está me pedindo algum sistema de cobertura entre as duas abordagens de fazer um grande em versus usando get () para identificações individuais.

Se alguém está realmente tentando evitar a seleção, a melhor maneira de fazer isso é configurar os objetos necessários na memória com antecedência. Como, você está trabalhando em uma grande tabela de elementos. Divida o trabalho em pedaços, como, peça o conjunto completo de trabalho por chave primária, ou por intervalo, seja o que for, então carregue tudo para esse pedaço localmente em um cache:

 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>

Esse é o caso de uso do mundo real.

Mas, para ilustrar também algumas APIs da SQLalChemy, podemos fazer uma função que faz os registros que não temos e uma obtenção local para aqueles que fazemos. Aqui está isso:

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

Aqui está uma demonstração:

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

Se você usar teclas primárias compostas, pode usar tuple_, como em

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

Observe que isso não está disponível no sqlite (ver Doc).

Answer 4

Eu recomendo dar uma olhada no SQL que produz. Você pode simplesmente imprimir str (consulta) para vê -lo.

Não estou ciente de uma maneira ideal de fazê -lo com SQL padrão.

Answer 5

Há uma outra maneira; Se for razoável esperar que os objetos em questão já estejam carregados na sessão; Você os acessou antes na mesma transação, você pode fazer:

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

No caso em que esses objetos já estão presentes, isso será muito mais rápido, pois não haverá dúvidas para recuperar esses objetos; Por outro lado, se mais de um pequeno número desses objetos forem não Carregado, será muito, muito mais lento, pois causará uma consulta por instância ausente, em vez de uma única consulta para todos os objetos.

Isso pode ser útil quando você estiver fazendo joinedload() Consultas antes de atingir a etapa acima, para que você possa ter certeza de que elas já foram carregadas. Em geral, você deve usar a solução na pergunta por padrão e explorar apenas esta solução quando viu que está consultando os mesmos objetos repetidamente.