Using OR in SQLAlchemy
I've looked through the docs and I cant seem to find out how to do an OR query in SQLAlchemy. I just want to do this query.
SELECT address FROM addressbook WHERE city='boston' AND (lastname='bulger' OR firstname='whitey')
Should be something like
addr = session.query(AddressBook).filter(City == "boston").filter(????)
From the tutorial:
from sqlalchemy import or_
filter(or_(User.name == 'ed', User.name == 'wendy'))
SQLAlchemy overloads the bitwise operators &
, |
and ~
so instead of the ugly and hard-to-read prefix syntax with or_()
and and_()
(like in Bastien's answer) you can use these operators:
.filter((AddressBook.lastname == 'bulger') | (AddressBook.firstname == 'whitey'))
Note that the parentheses are not optional due to the precedence of the bitwise operators.
So your whole query could look like this:
addr = session.query(AddressBook) \
.filter(AddressBook.city == "boston") \
.filter((AddressBook.lastname == 'bulger') | (AddressBook.firstname == 'whitey'))
or_()
function can be useful in case of unknown number of OR query components.
For example, let's assume that we are creating a REST service with few optional filters, that should return record if any of filters return true. On the other side, if parameter was not defined in a request, our query shouldn't change. Without or_()
function we must do something like this:
query = Book.query
if filter.title and filter.author:
query = query.filter((Book.title.ilike(filter.title))|(Book.author.ilike(filter.author)))
else if filter.title:
query = query.filter(Book.title.ilike(filter.title))
else if filter.author:
query = query.filter(Book.author.ilike(filter.author))
With or_()
function it can be rewritten to:
query = Book.query
not_null_filters = []
if filter.title:
not_null_filters.append(Book.title.ilike(filter.title))
if filter.author:
not_null_filters.append(Book.author.ilike(filter.author))
if len(not_null_filters) > 0:
query = query.filter(or_(*not_null_filters))
For SQLAlchemy ORM 2.0 both |
and or_
are accepted.
Documentation
from sqlalchemy.future import select
from sqlalchemy.sql import or_
query = select(User).where(or_(User.name == 'ed', User.name == 'wendy'))
print(query)
# also possible:
query = select(User).where((User.name == 'ed') | (User.name == 'wendy'))
print(query)
This has been really helpful. Here is my implementation for any given table:
def sql_replace(self, tableobject, dictargs):
#missing check of table object is valid
primarykeys = [key.name for key in inspect(tableobject).primary_key]
filterargs = []
for primkeys in primarykeys:
if dictargs[primkeys] is not None:
filterargs.append(getattr(db.RT_eqmtvsdata, primkeys) == dictargs[primkeys])
else:
return
query = select([db.RT_eqmtvsdata]).where(and_(*filterargs))
if self.r_ExecuteAndErrorChk2(query)[primarykeys[0]] is not None:
# update
filter = and_(*filterargs)
query = tableobject.__table__.update().values(dictargs).where(filter)
return self.w_ExecuteAndErrorChk2(query)
else:
query = tableobject.__table__.insert().values(dictargs)
return self.w_ExecuteAndErrorChk2(query)
# example usage
inrow = {'eqmtvs_id': eqmtvsid, 'datetime': dtime, 'param_id': paramid}
self.sql_replace(tableobject=db.RT_eqmtvsdata, dictargs=inrow)