如何使用SQLAlchemy获取符合3元组条件的行

Question

我有一个包含三元组的列表：

[(a, b, c), (d, e, f)]

我想从一个表中获取所有行，这些行的三列要和这些元组匹配。对于这个例子，查询的 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 来创建这样的请求呢？在我的情况下，这个三元组的列表会有几百个元素，我在寻找一个最佳的可扩展解决方案。

谢谢你的帮助，

Answer 1

有人会考虑在原来的表格里增加一个额外的键吗？也就是说，创建一个新的列，用“1”-“2”-“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

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