如何使用SQLAlchemy查询用户尚未浏览过的帖子

核心思路是通过NOT EXISTS子查询过滤掉用户已经产生浏览记录的帖子，这种写法比左连接判空的查询性能更好，是这类场景的标准实现。

• 性能优势：数据库查询优化器会对EXISTS子查询做短路优化，只要找到一条匹配的浏览记录就会停止匹配，帖子和浏览记录数据量越大，性能优势越明显。

假设你已经通过Flask-Login拿到了当前登录用户对象current_user，直接用下面的查询即可：

from sqlalchemy import exists

# 构造当前用户已浏览帖子的匹配子查询
has_viewed = exists().where(
    PostViewed.post_id == Post.id,
    PostViewed.author_id == current_user.id
)

# 取反得到所有未浏览的帖子
unread_posts = Post.query.filter(~has_viewed).all()

如果需要分页展示，直接替换末尾的.all()为分页方法即可：

# 按每页20条分页，传入当前页码current_page即可
unread_posts_page = Post.query.filter(~has_viewed).paginate(page=current_page, per_page=20)

复用优化 #

如果这个查询在多处用到，可以直接封装到Post模型的静态方法里，减少重复代码：

class Post(db.Model):
    __tablename__ = "posts"
    id = db.Column(db.String, unique=True, primary_key=True)
    title = db.Column(db.String(124), nullable=False)
    description = db.Column(db.Text, nullable=False)
    post_viewed = relationship("PostViewed", back_populates="post", cascade="all, delete-orphan")

    @staticmethod
    def get_unread_posts(user_id: int):
        viewed_match = exists().where(
            PostViewed.post_id == Post.id,
            PostViewed.author_id == user_id
        )
        return Post.query.filter(~viewed_match)

调用时直接传入用户ID即可，比如查询ID为5的用户的未读帖子：

unread_posts = Post.get_unread_posts(user_id=5).all()

注意：你的模型中PostViewed.author_id设置了nullable=True，匿名用户产生的浏览记录（author_id为NULL）不会干扰登录用户的未读筛选结果，不需要额外加过滤条件。

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