sqlalchemy, en tournant une liste d'Id d'une liste d'objets

Le code est tout à fait acceptable. Cependant, quelqu'un me demande un système de couverture entre les deux approches, de faire un grand DANS contre l'utilisation de get() pour chaque Id.

Si quelqu'un est vraiment en essayant d'éviter le SÉLECTIONNER, puis la meilleure façon de le faire est de configurer les objets dont vous avez besoin dans la mémoire à l'avance. Tels que, vous travaillez sur un grand tableau d'éléments. Briser le travail en morceaux, tels que, afin de l'ensemble des travaux par clé primaire, ou par plage de date, que ce soit, alors la charge de tout pour que le morceau localement en 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>

C'est le monde réel de cas d'utilisation.

Mais aussi d'illustrer certains de SQLAlchemy API, nous pouvons faire une fonction qui fait le DANS les pour les registres, nous n'avons pas et un local à obtenir pour ceux que nous ne. Ici, c'est 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

Voici une démonstration:

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

Je vous recommande de prendre un regard sur SQL qu'il produit. Vous pouvez simplement imprimer str(requête) pour le voir.

Je ne suis pas au courant de la façon idéale de le faire avec le standard SQL.