SQLAlchemy查询中有exists关键字

Question

如何检查一个查询中的数据是否存在？

例如：

users_query = User.query.filter_by(email='[e-mail]')

我该如何检查是否存在使用该电子邮件的用户？

我可以使用以下方式进行检查：

users_query.count()

但如何使用 exists 进行检查呢？

- lestat

Answer 1

以下解决方案稍微简单一些：

from sqlalchemy.sql import exists

print session.query(exists().where(User.email == '...')).scalar()

- Cito

---

注意，这对于多态类型会产生错误的结果。当同时过滤父类和子类属性时，得到的查询将从表的笛卡尔积（外连接）中选择。为了修复它，您应该通过 select_from 手动设置 FROM 子句:e = exists(select([1]).select_from(User).where(and_(User.email == '...', ...))).select() - aikoven

@aikoven：您的建议导致“每个导出表必须具有自己的别名”错误。添加 exists(…).select().alias('foo') 解决了问题（请参见 alias() 函数）。 - Jens

Answer 2

对我来说最可接受和易读的选择是

session.query(<Exists instance>).scalar()

喜欢

session.query(User.query.filter(User.id == 1).exists()).scalar()

该函数返回True或False。

- Eugene Kovalev

---

另外，请注意，某些数据库（如SQL Server）不允许在SELECT的列子句中存在EXISTS表达式。要基于WHERE中的exists选择简单的布尔值，请使用sqlalchemy.literal：session.query(literal(True)).filter(q.exists()).scalar()。 - kaka

这会抛出以下异常：AttributeError: 类型对象 'User' 没有属性 'query' - Fusion

.query 属性是 特定于 Flask-SQLAlchemy 的。您可以使用 session.query(session.query(User).filter(User.id == 1).exists()).scalar()。 - Martijn Pieters

Answer 3

我不知道有没有办法使用ORM查询API来实现这一点。但是你可以降低到更底层的水平，并使用sqlalchemy.sql.expression中的exists：

from sqlalchemy.sql.expression import select, exists

users_exists_select = select((exists(users_query.statement),)) 
print engine.execute(users_exists_select).scalar()

- Gary van der Merwe

Answer 4

SqlAlchemy 1.4的答案

避免使用.query，而是直接使用链接.exists的select，如下所示：

from sqlalchemy import select

stmt = select(User).where(User.email=="[e-mail]").exists()

来源: https://docs.sqlalchemy.org/en/14/core/selectable.html?highlight=exists#sqlalchemy.sql.expression.exists

- pcko1

---

这是 SqlAlchemy 1.4。 - kolypto

尝试执行此操作将导致错误，因为Exists是一个子句而不是语句。您需要将其包装在select()中。 - Theron Luhn

Answer 5

对于SQL Server，我需要这样做：

from sqlalchemy.sql.expression import literal

result = session.query(literal(True)).filter(
    session.query(User)
    .filter_by(email='...')
    .exists()
).scalar()
print(result is not None)

# Resulting query:
# SELECT 1
# WHERE EXISTS (SELECT 1 
# FROM User
# WHERE User.email = '...')

但是如果没有EXISTS，那就简单多了：

result = (
    session.query(literal(True))
    .filter(User.email == '...')
    .first()
)
print(result is not None)

# Resulting query:
# SELECT TOP 1 1
# FROM User
# WHERE User.email = '...'

- MarredCheese

---

同样适用于 Oracle 12c。感谢您提供的解决方案。 - Moritz Ringler

session.query(cls).filter_by(**kwargs).first() is not None 也可以工作，无需导入 literal，至少对于 MySQL 8.0。 - ATH

Answer 6

使用新的统一API，在SQLAlchemy 1.4/2.0中最简单的方法：

from sqlalchemy import exists


session.scalar(
    exists()
    .where(User.email == '[e-mail]')
    .select()
)

- Theron Luhn

Answer 7

可以做到：

from sqlalchemy import select

user = session.scalars(
    select(User).where(User.email=="[e-mail]")
).first()

if user:
    pass
else:
    pass

- Danniel Little