我有一系列要检索的 ID。很简单：
session.query(Record).filter(Record.id.in_(seq)).all()
有没有更好的方法？
【问题讨论】：
seq 变得足够长，您可能会收到“SQL 变量过多”异常，因为 IN 子句已参数化且参数过多。
标签： python select sqlalchemy
您的代码绝对没问题。
IN 就像一堆X=Y 与OR 结合在一起，在当代数据库中非常快。
但是，如果您的 ID 列表很长，您可以通过传递返回 ID 列表的子查询来提高查询效率。
【讨论】：
代码完全没问题。但是，有人问我在进行大 IN 与使用 get() 处理个人 ID 的两种方法之间建立某种对冲系统。
如果有人真的想避免 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 并为我们有的记录执行本地 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
【讨论】：
如果使用复合主键，可以使用tuple_，如
from sqlalchemy import tuple_
session.query(Record).filter(tuple_(Record.id1, Record.id2).in_(seq)).all()
请注意，这在 SQLite 上不可用（请参阅 doc）。
【讨论】：
我建议看一下它生成的 SQL。你可以打印 str(query) 来查看它。
我不知道使用标准 SQL 的理想方法。
【讨论】：
还有另一种方法；如果期望相关对象已经加载到会话中是合理的；你以前在同一个事务中访问过它们，你可以这样做：
map(session.query(Record).get, seq)
在这些对象已经存在的情况下，这会更快，因为不会有任何查询来检索这些对象；另一方面，如果没有加载这些对象中的一小部分，它会慢得多，因为它会导致每个缺失实例的查询，而不是对所有的单个查询对象。
当您在到达上述步骤之前执行joinedload() 查询时，这可能很有用，因此您可以确定它们已经被加载。一般情况下，您应该默认使用问题中的解决方案，并且只有在您看到您一遍又一遍地查询相同的对象时才探索此解决方案。
【讨论】：