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

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

我找到了一个有用的提示,用于模仿子集合here中的对象数量.基本上,它表示以下流程：

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

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

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

如果我按照该链接中的建议并使用.limit(),那么我需要遍历在每个用户上调用.limit()的整个用户集合.例如,在创建集合的原始sql表达式中使用LIMIT,效率要低得多.

我的问题是,是否可以使用声明来有效地(N 0)加载大量对象,同时使用sqlalchemy限制子集合中的子节点数量？

UPDATE

需要说明的是,以下是我要避免的内容.

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()

解决方法:

这个答案来自Mike Bayer在sqlalchemy google group.我在这里张贴它来帮助大家：
TLDR：
我使用Mike的答案版本1来解决我的问题,因为在这种情况下,我没有涉及此关系的外键,因此无法使用LATERAL.版本1工作得很好,但一定要注意偏移的影响.它在测试期间让我离开了一段时间,因为我没有注意到它被设置为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))

迈克的回答
所以你应该忽略它是否使用“声明”,这与查询无关,事实上最初忽略查询,因为首先这是一个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的ON子句内部进行关联,以便在vanilla 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构建第二个关系,或者您可以单独从查询(User,Message)加载并根据需要组织结果元组.

如果您不使用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)

标签：python,limit,sqlalchemy,flask-sqlalchemy,declarative
来源： https://codeday.me/bug/20190705/1391598.html