在初始查询 sqlalchemy 中限制子集合

Question

我正在构建一个 api，如果用户请求它可以返回资源的子级。例如，user有messages. 我希望查询能够限制message返回的对象数量。

我发现了一个有用的技巧aboutl imiting在子集合对象的数量在这里。基本上，它表示以下流程：

class User(...):
    # ...
    messages = relationship('Messages', order_by='desc(Messages.date)', lazy='dynamic')

user = User.query.one()
users.messages.limit(10)

我的用例涉及有时会返回大量用户。

如果我要遵循该链接中的建议并使用，.limit()那么我将需要遍历调用.limit()每个用户的整个用户集合。这比LIMIT在创建集合的原始 sql 表达式中使用效率低得多。

我的问题是，是否有可能使用声明来有效地（N+0）加载大量对象，同时使用 sqlalchemy 限制其子集合中的子集合的数量？

更新

需要明确的是，以下是我试图避免的。

users = User.query.all()
messages = {}
for user in users:
    messages[user.id] = user.messages.limit(10).all()

我想做一些更像：

users = User.query.option(User.messages.limit(10)).all()

Answer 1

这个答案来自sqlalchemy google group的 Mike Bayer 。我将其发布在这里是为了帮助人们： TLDR： 我使用version 1Mike 的答案来解决我的问题，因为在这种情况下，我没有参与此关系的外键，因此无法使用LATERAL. 版本 1 效果很好，但一定要注意offset. 它在测试过程中让我困惑了一段时间，因为我没有注意到它被设置为0.

版本 1 的代码块：

subq = s.query(Messages.date).\
    filter(Messages.user_id == User.id).\
    order_by(Messages.date.desc()).\
    limit(1).offset(10).correlate(User).as_scalar()

q = s.query(User).join(
    Messages,
    and_(User.id == Messages.user_id, Messages.date > subq)
).options(contains_eager(User.messages))

Mike的回答 所以你应该忽略它是否使用“声明式”，这与查询无关，事实上首先也忽略查询，因为首先这是一个SQL问题。您需要一个 SQL 语句来执行此操作。SQL 中的哪个查询会从主表加载大量行，并连接到每个主表的辅助表的前十行？

LIMIT 很棘手，因为它实际上不是通常的“关系代数”计算的一部分。它不在其中，因为它是对行的人为限制。例如，我对如何执行此操作的第一个想法是错误的：

    select * from users left outer join (select * from messages limit 10) as anon_1 on users.id = anon_1.user_id

这是错误的，因为它只获取聚合中的前十条消息，而不考虑用户。我们希望获取每个用户的前十条消息，这意味着我们需要为每个用户单独执行“从消息限制 10 条中选择”操作。也就是说，我们需要以某种方式关联起来。虽然相关子查询通常不允许作为 FROM 元素，而只允许作为 SQL 表达式，但它只能返回单列和单行；我们通常无法在普通 SQL 中 JOIN 到相关子查询。然而，我们可以在 JOIN 的 ON 子句内部进行关联，从而在普通 SQL 中实现这一点。

但首先，如果我们使用的是现代 Postgresql 版本，我们可以打破通常的关联规则并使用名为 LATERAL 的关键字，它允许在 FROM 子句中进行关联。LATERAL 仅受现代 Postgresql 版本支持，它使这变得简单：

    select * from users left outer join lateral
    (select * from message where message.user_id = users.id order by messages.date desc limit 10) as anon1 on users.id = anon_1.user_id

我们支持 LATERAL 关键字。上面的查询看起来像这样：

subq = s.query(Messages).\
    filter(Messages.user_id == User.id).\
    order_by(Messages.date.desc()).limit(10).subquery().lateral()

q = s.query(User).outerjoin(subq).\
     options(contains_eager(User.messages, alias=subq))

请注意，上面，为了选择用户和消息并将它们生成到 User.messages 集合中，必须使用“contains_eager()”选项，并且为此“动态”必须消失。这不是唯一的选择，例如，您可以为没有“动态”的 User.messages 构建第二个关系，或者您可以单独从查询（用户，消息）加载并根据需要组织结果元组。

如果您不使用 Postgresql 或不支持 LATERAL 的 Postgresql 版本，则必须将相关性纳入联接的 ON 子句中。SQL 看起来像：

select * from users left outer join messages on
users.id = messages.user_id and messages.date > (select date from messages where messages.user_id = users.id order by date desc limit 1 offset 10)

在这里，为了将 LIMIT 塞在那里，我们实际上是使用 OFFSET 逐步遍历前 10 行，然后执行 LIMIT 1 来获取代表我们想要为每个用户提供的下限日期的日期。然后我们必须在比较该日期时加入，如果该列没有索引，这可能会很昂贵，而且如果存在重复的日期，这也可能不准确。

这个查询看起来像：

subq = s.query(Messages.date).\
    filter(Messages.user_id == User.id).\
    order_by(Messages.date.desc()).\
    limit(1).offset(10).correlate(User).as_scalar()

q = s.query(User).join(
    Messages,
    and_(User.id == Messages.user_id, Messages.date >= subq)
).options(contains_eager(User.messages))

如果没有经过良好的测试，我不会信任此类查询，因此下面的 POC 包括两个版本，其中包括健全性检查。

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
import datetime

Base = declarative_base()


class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    messages = relationship(
        'Messages', order_by='desc(Messages.date)')

class Messages(Base):
    __tablename__ = 'message'
    id = Column(Integer, primary_key=True)
    user_id = Column(ForeignKey('user.id'))
    date = Column(Date)

e = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
Base.metadata.drop_all(e)
Base.metadata.create_all(e)

s = Session(e)

s.add_all([
    User(id=i, messages=[
        Messages(id=(i * 20) + j, date=datetime.date(2017, 3, j))
        for j in range(1, 20)
    ]) for i in range(1, 51)
])

s.commit()

top_ten_dates = set(datetime.date(2017, 3, j) for j in range(10, 20))


def run_test(q):
    all_u = q.all()
    assert len(all_u) == 50
    for u in all_u:

        messages = u.messages
        assert len(messages) == 10

        for m in messages:
            assert m.user_id == u.id

        received = set(m.date for m in messages)

        assert received == top_ten_dates

# version 1.   no LATERAL

s.close()

subq = s.query(Messages.date).\
    filter(Messages.user_id == User.id).\
    order_by(Messages.date.desc()).\
    limit(1).offset(10).correlate(User).as_scalar()

q = s.query(User).join(
    Messages,
    and_(User.id == Messages.user_id, Messages.date > subq)
).options(contains_eager(User.messages))

run_test(q)

# version 2.  LATERAL

s.close()

subq = s.query(Messages).\
    filter(Messages.user_id == User.id).\
    order_by(Messages.date.desc()).limit(10).subquery().lateral()

q = s.query(User).outerjoin(subq).\
    options(contains_eager(User.messages, alias=subq))

run_test(q)