sqlalchemy, eine Liste von IDs in eine Liste von Objekten umwandeln

Question

Ich habe eine Folge von IDs, die ich abrufen möchte. Es ist einfach:

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

Gibt es einen besseren Weg, es zu tun?

Answer 1

Ihr Code ist absolut in Ordnung.

IN ist wie ein Bündel von X=Y mit OR verbunden und ist ziemlich schnell in modernen Datenbanken.

Wenn Ihre ID-Liste jedoch lang ist, können Sie die Abfrage etwas effizienter gestalten, indem Sie eine Unterabfrage übergeben, die die ID-Liste zurückgibt.

Answer 2

Ich würde empfehlen, einen Blick auf die SQL zu werfen, die es produziert. Sie können einfach str (query) drucken, um es zu sehen.

Mir ist keine ideale Möglichkeit bekannt, dies mit Standard-SQL zu tun.

Answer 3

Wenn Sie Composite Primärschlüssel verwenden, können Sie tuple_, wie in

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

Hinweis verwenden, die dieses auf SQLite nicht verfügbar ist (siehe doc).

Answer 4

Es gibt einen anderen Weg; Wenn es vernünftigerweise zu erwarten ist, dass die fraglichen Objekte bereits in die Sitzung geladen sind; Sie sie, bevor sie in der gleichen Transaktion zugegriffen haben, können Sie stattdessen tun:

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

In dem Fall, wo diese Objekte bereits vorhanden sind, so viel schneller sein wird, da es keine Anfragen werden an diejenigen abrufen Objekte; Auf der anderen Seite, wenn mehr als eine winzige Anzahl dieser Objekte nicht geladen werden, wird es viel, viel langsamer, da es eine Abfrage pro fehlende Instanz statt einer einzigen Abfrage für alle Objekte verursachen wird.

Dies kann nützlich sein, wenn Sie joinedload() Abfragen ausführen, bevor Sie den obigen Schritt erreichen, so dass Sie sicher sein können, dass sie bereits geladen wurden. Im Allgemeinen sollten Sie die Lösung in der Frage standardmäßig verwenden und diese Lösung nur untersuchen, wenn Sie festgestellt haben, dass Sie immer wieder nach den gleichen Objekten suchen.

Answer 5

Der Code wie er ist, ist völlig in Ordnung. Allerdings bittet mich jemand um ein Hedging-System zwischen den beiden Ansätzen, einen großen IN zu machen, indem man get() für einzelne IDs verwendet.

Wenn jemand wirklich versucht, SELECT zu vermeiden, dann sollten Sie die Objekte, die Sie benötigen, im Voraus einrichten. So arbeiten Sie an einer großen Tabelle von Elementen. Brechen Sie die Arbeit in Stücke, wie der vollständigen Satz von Arbeit durch einen Primärschlüssel oder nach Datumsbereich, was auch immer, dann laden Sie alles für diese Brocken lokal in einen 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> 

Das ist die reale Welt Verwendung Fall.

Aber auch um einige SQLAlchemy API zu veranschaulichen, können wir eine Funktion machen, die das IN für Datensätze, die wir nicht haben, und ein lokales get für diejenigen, die wir tun.Hier ist, dass:

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 

hier eine Demonstration ist:

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