How to get rows which match a list of 3-tuples conditions with SQLAlchemy
Easiest way would be using SQLAlchemy-provided tuple_ function:
from sqlalchemy import tuple_
session.query(Foo).filter(tuple_(Foo.a, Foo.b, Foo.c).in_(items))
This works with PostgreSQL, but breaks with SQLite. Not sure about other database engines.
Fortunately there's a workaround that should work on all databases.
Start by mapping out all the items with the and_
expression:
conditions = (and_(c1=x, c2=y, c3=z) for (x, y, z) in items)
And then create an or_
filter that encloses all the conditions:
q.filter(or_(*conditions))
Here's a simple example:
#/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()
Which outputs:
$ 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)]
A less conventional approach that I suspect would scale well would be to create a temporary table of all your tuples and then join on that:
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()
which executes SQL like this:
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