如何使用 SQLAlchemy 获取与三元组条件列表匹配的行

Question

拥有 3 元组列表：

[(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)

How can I create such a request using SQLAlchemy ？就我而言，三元组列表将包含数百个元素，我正在寻找最佳的可扩展解决方案。

感谢您的帮助，

Answer 1

最简单的方法是使用 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)]

Answer 2

我怀疑可以很好地扩展的一种不太传统的方法是创建所有元组的临时表，然后加入该表：

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

Answer 3

有人会考虑在原始表中创建一个额外的键吗？
即用“1”-“2”-“3”创建一个新列而不是另一个表并检查唯一性。