SQLAlchemy 2.0+（asyncpg驱动）如何实现类似Django exists()的功能？

在SQLAlchemy 2.0+的异步模式（搭配asyncpg）中，有几种高效的方式实现类似Django queryset.exists()的功能，不需要获取并反序列化整个对象：

• 使用exists()构造查询（推荐）
  这是最接近Django exists()的实现，数据库仅需检查是否存在匹配行，不会返回实际数据，性能最优。示例代码：

  from sqlalchemy import select, exists
  from sqlalchemy.ext.asyncio import AsyncSession
  from your_models_module import User  # 替换为你的模型导入路径
  
  async def user_exists(session: AsyncSession) -> bool:
      # 构造exists查询，匹配指定条件
      exists_query = exists().where(
          User.id.in_([1, 2, 3, 4, 5]),
          User.username == 'test'
      )
      # 执行查询并获取布尔结果
      result = await session.execute(select(exists_query))
      return result.scalar_one()
  

  scalar_one()会直接返回查询的布尔结果，因为exists()查询的返回值就是单个布尔值。

• 查询单个字段并限制行数
  也可以通过查询一个常量值（比如1）并限制返回1行，再判断是否有结果：

  from sqlalchemy import select
  from sqlalchemy.ext.asyncio import AsyncSession
  from your_models_module import User
  
  async def user_exists(session: AsyncSession) -> bool:
      query = select(1).where(
          User.id.in_([1, 2, 3, 4, 5]),
          User.username == 'test'
      ).limit(1)
      result = await session.execute(query)
      # 如果有匹配行，scalar()返回1，否则返回None
      return result.scalar() is not None
  

需要注意的是，await session.get()是根据主键获取单个对象的方法，不适合多条件过滤的场景，而且会完整获取并反序列化对象，远不如上面两种方法高效。

内容的提问来源于stack exchange，提问作者Aleksei Khatkevich