Sqlalchemy, превращая список идентификаторов в список объектов

Ответ 1

Ваш код безупречен.

IN похож на пучок X=Y, соединенный с OR и довольно быстро работает в современных базах данных.

Однако, если ваш список идентификаторов длинный, вы можете сделать запрос более эффективным, передав подзапрос, возвращающий список идентификаторов.

Ответ 2

Код как есть полностью. Тем не менее, кто-то спрашивает меня о какой-то системе хеджирования между двумя подходами к выполнению большого IN против использования get() для отдельных идентификаторов.

Если кто-то действительно пытается избежать SELECT, лучший способ сделать это - настроить объекты, которые вам нужны в памяти раньше времени. Например, вы работаете над большой таблицей элементов. Разбивайте работу на куски, например, заказывайте полный набор работ по первичному ключу или по дате, независимо от того, загрузите ли вы все для этого фрагмента локально в кеш:

 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>

Это реальный случай использования.

Но для иллюстрации некоторых API SQLAlchemy мы можем сделать функцию, которая делает IN для записей, которых у нас нет, и локальный get для тех, которые мы делаем. Вот что:

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

Вот демонстрация:

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

Ответ 3

Если вы используете составные первичные ключи, вы можете использовать tuple_, как в

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

Обратите внимание, что это недоступно в SQLite (см. doc).

Ответ 4

Я бы рекомендовал взглянуть на SQL, который он производит. Вы можете просто напечатать str (query), чтобы увидеть его.

Я не знаю идеального способа сделать это со стандартным SQL.

Ответ 5

Есть еще один способ; Если разумно ожидать, что объекты, о которых идет речь, уже загружены в сеанс; вы получили доступ к ним ранее в одной транзакции, вы можете:

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

В случае, когда эти объекты уже присутствуют, это будет намного быстрее, так как не будет никаких запросов для извлечения этих объектов; С другой стороны, если более чем небольшое количество этих объектов не загружено, это будет намного, намного медленнее, поскольку это вызовет запрос на один недостающий экземпляр вместо одного запроса для всех объектов.

Это может быть полезно, когда вы делаете запросы joinedload(), прежде чем достигнуть вышеуказанного шага, чтобы вы могли быть уверены, что они уже загружены. В общем, вы должны использовать решение в вопросе по умолчанию и только изучать это решение, когда увидите, что вы запрашиваете одни и те же объекты снова и снова.