sqlalchemy，將ID列表轉換爲對象列表

Question

我有要檢索的ID序列。這很簡單：

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

有沒有更好的方法來做到這一點？

Answer 1

你的代碼是絕對沒問題的。

IN就像一堆X=Y加入OR，並且在當代數據庫中速度相當快。

但是，如果您的ID列表很長，您可以通過傳遞一個返回ID列表的子查詢來提高查詢效率。

Answer 2

我建議看看它產生的SQL。你可以打印str（查詢）來查看它。

我不知道使用標準SQL進行此操作的理想方式。

Answer 3

如果使用複合主鍵，你可以使用tuple_，如

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

請注意，這不是可在SQLite的（見doc）。

Answer 4

還有一種方法;如果期望所討論的對象已經加載到會話中是合理的，您在同一交易之前訪問了它們，你可以改爲做：

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

在這些對象已經存在的情況下，這會快很多，因爲不會有任何疑問，那些獲取對象;另一方面，如果加載的對象中有不止一小部分是而不是，它會慢很多，因爲它會導致每個缺少實例的查詢，而不是所有對象的單個查詢。

當您在執行joinedload()查詢之前，在達到上述步驟之前，這可能很有用，因此您可以確定它們已被加載。通常，您應該在默認情況下在問題中使用解決方案，並且只有在看到您一遍又一遍地查詢相同對象時才能探索此解決方案。

Answer 5

代碼是完全正常的。然而，有人要求我在做大型IN的兩種方法之間進行套期保值，而對個人ID使用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> 

這纔是真正的世界中使用案件。

但是爲了說明一些SQLAlchemy API，我們可以創建一個函數，用於記錄我們沒有的IN以及用於我們所做的記錄的本地獲取。下面是：

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