SQLAlchemy - WHERE 子句中的子查询

Posted 2023-02-16

【中文标题】SQLAlchemy - WHERE 子句中的子查询

【英文标题】：SQLAlchemy - subquery in a WHERE clause

【发布时间】：2011-09-06 14:07:48

【问题描述】：

我最近才开始使用 SQLAlchemy，但仍然无法理解一些概念。

归结为基本元素，我有两个这样的表（这是通过 Flask-SQLAlchemy）：

class User(db.Model):
    __tablename__ = 'users'
    user_id = db.Column(db.Integer, primary_key=True)

class Posts(db.Model):
    __tablename__ = 'posts'
    post_id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('users.user_id'))
    post_time = db.Column(db.DateTime)

    user = db.relationship('User', backref='posts')

我将如何查询用户列表及其最新帖子（不包括没有帖子的用户）。如果我使用 SQL，我会这样做：

SELECT [whatever]
FROM posts AS p
    LEFT JOIN users AS u ON u.user_id = p.user_id
WHERE p.post_time = (SELECT MAX(post_time) FROM posts WHERE user_id = u.user_id)

所以我确切地知道“想要的”SQL 以获得我想要的效果，但不知道如何在 SQLAlchemy 中“正确地”表达它。

编辑：如果它很重要，我使用的是 SQLAlchemy 0.6.6。

【参考方案1】：

这应该可行（不同的 SQL，相同的结果）：

t = Session.query(
    Posts.user_id,
    func.max(Posts.post_time).label('max_post_time'),
).group_by(Posts.user_id).subquery('t')

query = Session.query(User, Posts).filter(and_(
    User.user_id == Posts.user_id,
    User.user_id == t.c.user_id,
    Posts.post_time == t.c.max_post_time,
))

for user, post in query:
    print user.user_id, post.post_id

其中 c 代表“列”

【参考方案2】：

前面的答案有效，但您要求的确切 sql 也与实际语句一样写：

print s.query(User, Posts).\
    outerjoin(Posts.user).\
    filter(Posts.post_time==\
        s.query(
            func.max(Posts.post_time)
        ).
        filter(Posts.user_id==User.user_id).
        correlate(User).
        as_scalar()
    )

我猜这个不一定明显的“概念”是当前需要 as_scalar() 来将子查询建立为“标量”（它可能应该从上下文中假设 ==）。

编辑：确认，这是错误行为，已完成票证 #2190。在当前的提示或版本 0.7.2 中，自动调用了 as_scalar()，上面的查询可以是：

print s.query(User, Posts).\
    outerjoin(Posts.user).\
    filter(Posts.post_time==\
        s.query(
            func.max(Posts.post_time)
        ).
        filter(Posts.user_id==User.user_id).
        correlate(User)
    )

【参考方案3】：

它的表达方式通常与实际 SQL 类似 - 您创建一个返回单个结果的子查询并与之进行比较 - 但是有时真正的痛苦是如果您必须在子查询中使用您已经在查询或加入的表。

解决方案是创建模型的别名版本以在子查询中引用。

假设您已经在一个连接中运行，其中您有一个现有的 Posts model 和一些基本的 query 准备好了 - 现在，您想要查询最新（单个）帖子的列表每个用户，你会像这样过滤查询：

from sqlalchemy.orm import aliased
posts2 = aliased(Posts) # create aliased version

query = query.filter(
    model.post_id
    ==
    Posts.query # create query directly from model, NOT from the aliased version!
        .with_entities(posts2.post_id) # only select column "post_id"
        .filter(
            posts2.user_id == model.user_id
        )
        .order_by(posts2.post_id.desc()) # assume higher id == newer post
        .limit(1) # we must limit to a single row so we only get 1 value
)

我故意不使用func.max，因为我认为这是一个更简单的版本，并且它已经在其他答案中，我认为这个例子对通常发现这个问题的人有用，因为他们正在寻找解决方案子查询同一个表。