有一个三元组列表：
[(a, b, c), (d, e, f)]
我想从 3 列与元组匹配的表中检索所有行。对于这个例子，查询 WHERE 子句可能是这样的：
(column_X = a AND column_Y = b AND column_Z = c)
OR (column_X = d AND column_Y = e AND column_Z = f)
如何使用 SQLAlchemy 创建这样的请求？在我的例子中，三元组列表将包含数百个元素，我正在寻找最佳的可扩展解决方案。
感谢您的帮助，
【问题讨论】：
标签： python sql sqlalchemy
最简单的方法是使用 SQLAlchemy 提供的 tuple_ 函数：
from sqlalchemy import tuple_
session.query(Foo).filter(tuple_(Foo.a, Foo.b, Foo.c).in_(items))
这适用于 PostgreSQL，但不适用于 SQLite。不确定其他数据库引擎。
幸运的是，有一种解决方法应该适用于所有数据库。
首先使用and_ 表达式映射出所有项目：
conditions = (and_(c1=x, c2=y, c3=z) for (x, y, z) in items)
然后创建一个包含所有条件的or_ 过滤器：
q.filter(or_(*conditions))
这是一个简单的例子：
#/usr/bin/env python
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer
from sqlalchemy.sql import and_, or_
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite:///')
session = sessionmaker(bind=engine)()
Base = declarative_base()
class Foo(Base):
__tablename__ = 'foo'
id = Column(Integer, primary_key=True)
a = Column(Integer)
b = Column(Integer)
c = Column(Integer)
def __init__(self, a, b, c):
self.a = a
self.b = b
self.c = c
def __repr__(self):
return '(%d %d %d)' % (self.a, self.b, self.c)
Base.metadata.create_all(engine)
session.add_all([Foo(1, 2, 3), Foo(3, 2, 1), Foo(3, 3, 3), Foo(1, 3, 4)])
session.commit()
items = ((1, 2, 3), (3, 3, 3))
conditions = (and_(Foo.a==x, Foo.b==y, Foo.c==z) for (x, y, z) in items)
q = session.query(Foo)
print q.all()
q = q.filter(or_(*conditions))
print q
print q.all()
哪些输出：
$ python test.py
[(1 2 3), (3 2 1), (3 3 3), (1 3 4)]
SELECT foo.id AS foo_id, foo.a AS foo_a, foo.b AS foo_b, foo.c AS foo_c
FROM foo
WHERE foo.a = :a_1 AND foo.b = :b_1 AND foo.c = :c_1 OR foo.a = :a_2 AND foo.b = :b_2 AND foo.c = :c_2
[(1 2 3), (3 3 3)]
【讨论】：
我怀疑可以很好扩展的一种不太传统的方法是创建一个包含所有元组的临时表，然后加入该表：
import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, Table
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
engine = sqlalchemy.create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=engine)
session = Session()
class Triple(Base):
__tablename__ = 'triple'
id = Column(Integer(), primary_key=True)
x = Column(Integer())
y = Column(Integer())
z = Column(Integer())
ws_table = Table('where_sets', Base.metadata,
Column('x', Integer()),
Column('y', Integer()),
Column('z', Integer()),
prefixes = ['temporary']
)
Base.metadata.create_all(engine)
...
where_sets = [(1, 2, 3), (3, 2, 1), (1, 1, 1)]
ws_table.create(engine, checkfirst=True)
session.execute(ws_table.insert(), [dict(zip('xyz', s)) for s in where_sets])
matches = session.query(Triple).join(ws_table, (Triple.x==ws_table.c.x) & (Triple.y==ws_table.c.y) & (Triple.z==ws_table.c.z)).all()
这样执行 SQL：
INSERT INTO triple (x, y, z) VALUES (?, ?, ?)
(1, 2, 3)
INSERT INTO triple (x, y, z) VALUES (?, ?, ?)
(3, 1, 2)
INSERT INTO triple (x, y, z) VALUES (?, ?, ?)
(1, 1, 1)
SELECT triple.id AS triple_id, triple.x AS triple_x, triple.y AS triple_y, triple.z AS triple_z
FROM triple JOIN where_sets ON triple.x = where_sets.x AND triple.y = where_sets.y AND triple.z = where_sets.z
【讨论】：
有人会考虑在原始表中创建一个额外的键吗？ 即用“1”-“2”-“3”而不是另一个表创建一个新列并检查唯一性。
【讨论】：