在SQLAlchemy中使用OR

问题:在SQLAlchemy中使用OR

我看了看文档,似乎无法找出如何在SQLAlchemy中进行OR查询。我只想执行此查询。

SELECT address FROM addressbook WHERE city='boston' AND (lastname='bulger' OR firstname='whitey')

应该是这样的

addr = session.query(AddressBook).filter(City == "boston").filter(????)

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

回答 0

教程

from sqlalchemy import or_
filter(or_(User.name == 'ed', User.name == 'wendy'))

From the tutorial:

from sqlalchemy import or_
filter(or_(User.name == 'ed', User.name == 'wendy'))

回答 1

SQLAlchemy的重载位运算符&|~因此而不是丑陋和难以读取的前缀语法与or_()and_()(像巴斯蒂安的答案),你可以使用这些运营商:

.filter((AddressBook.lastname == 'bulger') | (AddressBook.firstname == 'whitey'))

请注意,由于按位运算符的优先级,括号不是可选的

因此,您的整个查询可能如下所示:

addr = session.query(AddressBook) \
    .filter(AddressBook.city == "boston") \
    .filter((AddressBook.lastname == 'bulger') | (AddressBook.firstname == 'whitey'))

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'))

回答 2

or_() 在未知数量的OR查询组件的情况下,该函数很有用。

例如,假设我们正在创建一个带有少量可选过滤器的REST服务,如果任何过滤器返回true,则该服务应返回记录。另一方面,如果未在请求中定义参数,则我们的查询不应更改。没有or_()功能,我们必须执行以下操作:

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

通过or_()功能可以将其重写为:

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

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

回答 3

这真的很有帮助。这是任何给定表的实现:

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)

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)