SQLAlchemy - 在WHERE子句中使用子查询

Question

我最近刚刚开始使用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版本。

- Chad Birch

Answer 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代表“列”

- sayap

Answer 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票。在当前的tip或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)
    )

- zzzeek

Answer 3

通常情况下，它的表达方式与实际SQL类似 - 您创建一个返回单个结果的子查询并将其与之比较。但是，有时如果您必须在子查询中使用正在查询或加入的表，则可能会非常麻烦。解决方案是创建一个别名版本的模型以在子查询中引用。因此，假设您已经在连接中操作，在那里您有一个现有的“帖子”模型和一些基本的查询准备好 - 现在，您想要查询每个用户的最新（单个）帖子列表，您会像这样过滤查询：

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，因为我认为它是简化版本，并且已经在其他答案中了。我认为这个例子对那些想要在同一张表中进行子查询的人很有用。

- jave.web

---

感谢您发布了这样一个现代化的答案，更容易地适应 select() 风格查询在 sqlalchemy 1.4/2.0 版本中。 - yurisich