判断数据是否存在
下面给你 PostgreSQL + SQLAlchemy 2.0 下常用的 “判断数据是否存在” 的最佳实践，分别包含：
- ✅ 原生 SQL 查询是否存在
- ✅ ORM 查询是否存在
- ✅ 推荐写法（性能最优）
- ❌ 低效写法（避免）
SELECT EXISTS (
SELECT 1 FROM users WHERE email = 'test@example.com'
);
返回：
| exists |
|---|
| true |
SELECT 1
FROM users
WHERE email = 'test@example.com'
LIMIT 1;
- 如果返回行 -> 存在
- 如果无返回 -> 不存在
from sqlalchemy import select, exists
from models import User
stmt = select(
exists().where(User.email == "test@example.com")
)
result = await session.scalar(stmt)
print(result) # True / False
特点：
- 不返回实体，不扫描全表
- 性能最好
- PostgreSQL 会用 EXISTS(…) 执行
stmt = select(1).where(User.email == "test@example.com").limit(1)
exists_ = await session.scalar(stmt) is not None
如果需要「判断存在且获取具体行」，用：
stmt = select(User).where(User.email == "test@example.com")
row = await session.scalar(stmt)
if row:
print("存在", row)
else:
print("不存在")
更严格的写法：
row = await session.scalar(select(User).where(User.id == 1))
if row is None:
print("不存在")
else:
print("存在")
stmt = select(func.count()).where(User.email == "test@example.com")
count = await session.scalar(stmt)
exists = count > 0
⚠️ 性能差，因为 COUNT 会扫描满足条件的全部行。
- PostgreSQL 中判断是否存在，用 EXISTS 是性能最优
- SQLAlchemy 2.0 中用 select(exists().where(…))
支持 ORM model 与任意 where 条件。
from sqlalchemy import exists, select
from sqlalchemy.ext.asyncio import AsyncSession
async def exists_row(session: AsyncSession, model, *conditions) -> bool:
stmt = select(exists().where(*conditions))
return await session.scalar(stmt)
使用：
exists_ = await exists_row(
session,
User,
User.email == "test@example.com"
)
print(exists_) # True / False