如何用单条SqlAlchemy查询统计用户未读通知数量？

你当前的实现需要循环查询每条通知的阅读状态，效率较低，完全可以通过单条SQL查询结合query.count()来实现统计，避免N+1查询问题。

实现代码 #

from sqlalchemy import or_

unread_count = (
    query
    .outerjoin(UserReads, (UserReads.nid == Notifs.id) & (UserReads.uid == current_user.id))
    .filter(
        or_(
            Notifs.tid == current_user.id,
            Notifs.target_groups.any(Role.id.in_([r.id for r in current_user.roles]))
        ),
        or_(UserReads.id.is_(None), UserReads.read == False)
    )
    .distinct(Notifs.id)
    .count()
)

逻辑说明 #

1. 左连接阅读记录表：通过outerjoin关联当前用户的UserReads记录，确保没有阅读记录的通知也能被纳入统计
2. 双重筛选条件：
   • 保留原有的通知可见性判断：要么通知目标是当前用户，要么用户属于通知的目标组
   • 新增未读判断：要么没有对应的阅读记录，要么有记录但未标记为已读
3. 去重处理：用distinct(Notifs.id)确保同一条通知不会被重复统计（逻辑上不会出现重复，但加上更严谨）
4. 直接统计数量：调用count()即可一次性得到未读通知的总数

这种方式把多次查询合并为一次数据库请求，在通知数量较多时性能提升明显。

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