sqlalchemy, trasformando un elenco di ID in un elenco di oggetti

Question

Ho una sequenza di ID che voglio recuperare. È semplice:

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

C'è un modo migliore per farlo?

Answer 1

Il tuo codice è assolutamente perfetto.

IN è come un gruppo di X = Y unito a OR ed è piuttosto veloce nei database contemporanei.

Tuttavia, se l'elenco di ID è lungo, è possibile rendere la query un po 'più efficiente passando una sottoquery che restituisce l'elenco di ID.

Answer 2

Il codice così com'è va benissimo. Tuttavia, qualcuno mi sta chiedendo un sistema di copertura tra i due approcci di fare un grande IN rispetto all'uso di get () per singoli ID.

Se qualcuno sta davvero cercando di evitare SELECT, il modo migliore per farlo è quello di impostare gli oggetti necessari in memoria in anticipo. Ad esempio, stai lavorando su una grande tabella di elementi. Suddividere il lavoro in blocchi, ad esempio, ordinare l'intero set di lavori in base alla chiave primaria o in base all'intervallo di date, quindi caricare tutto per quel blocco in locale in una 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>

Questo è il caso d'uso del mondo reale.

Ma per illustrare anche alcune API SQLAlchemy, possiamo creare una funzione che esegue l'IN per i record che non abbiamo e un local get per quelli che facciamo. Ecco quello:

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

Ecco una dimostrazione:

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 usi chiavi primarie composite, puoi usare tuple_ , come in

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

Nota che questo non è disponibile su SQLite (vedi doc ).

Answer 4

Consiglio di dare un'occhiata all'SQL che produce. Puoi semplicemente stampare str (query) per vederlo.

Non sono a conoscenza di un modo ideale per farlo con SQL standard.

Answer 5

C'è un altro modo; Se è ragionevole aspettarsi che gli oggetti in questione siano già caricati nella sessione; li hai già raggiunti nella stessa transazione, puoi invece fare:

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

Nel caso in cui tali oggetti siano già presenti, questo sarà molto più veloce, poiché non ci saranno domande per recuperare quegli oggetti; D'altra parte, se più di un numero esiguo di quegli oggetti non sono non caricati, sarà molto, molto più lento, poiché causerà una query per istanza mancante, anziché una singola query per tutti oggetti.

Ciò può essere utile quando si eseguono query joinload () prima di raggiungere il passaggio precedente, in modo da poter essere certi che siano già state caricate. In generale, dovresti usare la soluzione nella domanda per impostazione predefinita ed esplorare questa soluzione solo quando hai visto che stai interrogando ripetutamente gli stessi oggetti.