【问题标题】:Limit child collections in initial query sqlalchemy在初始查询 sqlalchemy 中限制子集合
【发布时间】:2017-05-01 22:11:47
【问题描述】:

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

我发现了一个有用的技巧,关于限制子集合中的对象数量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 限制子集合中的子对象数量?

更新

明确地说,以下是我试图避免的。

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

【问题讨论】:

    标签: python sqlalchemy limit flask-sqlalchemy declarative


    【解决方案1】:

    这个答案来自 sqlalchemy google group 上的 Mike Bayer。我在这里发布它以帮助人们: TLDR: 我使用 Mike 的回答中的version 1 来解决我的问题,因为在这种情况下,我没有涉及此关系的外键,因此无法使用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))
    

    迈克的回答 所以你应该忽略它是否使用“声明式”,这与查询无关,实际上一开始也忽略查询,因为首先这是一个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 条消息,这意味着我们需要为每个用户单独执行此“从消息限制 10 中选择”。也就是说,我们需要以某种方式关联。关联子查询虽然通常不允许作为 FROM 元素,只允许作为 SQL 表达式,但它只能返回单列单行;我们通常不能在普通的普通 SQL 中加入相关的子查询。但是,我们可以在 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)
    

    【讨论】:

    • 这太棒了,我现在在我的代码中解决这个问题。我确实觉得它实际上可能不正确。这是否仅因为您的消息日期在所有用户中都相同而起作用?子查询看起来像是从 DESC 列表中提供了 1 个日期回溯偏移量 10。您的所有消息都适合该块,因此从外部查询返回。我仍在尝试混合结果
    • 我最好的猜测是,如果您有在日期范围内的消息,它将包括所有匹配的用户/消息。如果没有比最新消息更新的消息,则它不会返回任何消息,或者返回所需的 50 条以下(因为不会有那么多要检索)。
    • 好的 .. 这是一些复杂的 shazzz .. 再次感谢您。似乎有一些魔法可以根据每个用户的消息日期进行外观。棒极了。如果偏移量大于该用户的消息数量,它会绊倒我,它似乎是空的。可能有一种方法可以做一个 or 子句之类的......有趣的星期六
    【解决方案2】:

    如果你应用限制,然后在其上调用.all(),你将获得所有对象一次,它不会一个一个地获得对象,导致你提到的性能问题。

    只需应用限制并获取所有对象。

    users = User.query.limit(50).all()
    print(len(users))
    >>50
    

    或者对于子对象/关系

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

    【讨论】:

    • 很抱歉给您带来了困惑。我不想限制users 的数量。我想限制 user.messages 中messages 的数量(users 的集合子)。
    • 同样适用于惰性子集合/关系
    • 为了更清楚,我想获得all() 用户,并且在返回的每个单独对象上,我希望消息数限制为 10(例如)。
    • 很抱歉给您带来了混乱。我已经更新了我的问题,以便更清楚。
    • 您不需要一一获取所有用户对象来对子项目应用限制。只需获取所有用户,然后对所有消息应用限制。您无法通过用户查询获得 10 条消息。你必须迭代
    【解决方案3】:

    所以,我认为您需要在第二个查询中加载消息,然后以某种方式与您的用户关联。 以下是依赖于数据库的;正如discussed in this question,mysql 不支持有限制的查询,但sqlite 至少会解析查询。我没有看计划,看它是否做得很好。 以下代码将找到您关心的所有消息对象。然后,您需要将它们与用户相关联。
    我已经对此进行了测试,以确认它产生了一个 sqlite 可以解析的查询;我还没有确认 sqlite 或任何其他数据库对这个查询是否正确。 我不得不作弊并使用文本原语来引用 select 中的外部 user.id 列,因为 SQLAlchemy 一直希望在内部 select 子查询中包含对用户的附加连接。

    from sqlalchemy import Column, Integer, String, ForeignKey, alias
    from sqlalchemy.sql import text
    
    from sqlalchemy.orm import Session
    from sqlalchemy.ext.declarative import declarative_base
    
    Base = declarative_base()
    
    class User(Base):
        __tablename__ = 'users'
        id = Column(Integer, primary_key = True)
        name = Column(String)
    
    class Message(Base):
        __tablename__ = 'messages'
        user_id = Column(Integer, ForeignKey(User.id), nullable = False)
        id = Column(Integer, primary_key = True)
    
    
    s = Session()
    m1 = alias(Message.__table__)
    
    user_query = s.query(User) # add any user filtering you want
    inner_query = s.query(m1.c.id).filter(m1.c.user_id == text('users.id')).limit(10)
    all_messages_you_want = s.query(Message).join(User).filter(Message.id.in_(inner_query))
    

    要将消息与用户关联,您可以执行以下操作,假设您的消息具有用户关系并且您的用户对象有一个 got_child_message 方法,可以为您做任何您喜欢的事情

    users_resulting = user_query.all() #load objects into session and hold a reference
    for m in all_messages_you_want: m.user.got_child_message(m)
    

    因为您已经在会话中拥有用户,并且因为关系在用户的主键上,所以 m.user 解析为针对身份映射的 query.get。 我希望这可以帮助您到达某个地方。

    【讨论】:

      【解决方案4】:

      @melchoirs 答案是最好的。我基本上把这个放在这里给未来自己

      我尝试了上述答案,并且它有效,我更需要它来限制在传递到棉花糖序列化器之前返回的关联数量。

      需要澄清的一些问题:

      • 子查询根据关联运行,因此它会找到相应的date 以正确定位
      • 将限制/偏移量视为从下一个 X(偏移量)开始给我 1 条(限制)记录。因此,第 X 个最旧的记录是什么,然后在主查询中,它会从中返回所有内容。它该死的聪明
      • 看来,如果关联的记录少于 X 条,则它不会返回任何内容,因为偏移量已超过记录,因此主查询不会返回记录。

      使用以上作为模板,我想出了以下答案。初始查询/计数保护是由于如果关联记录小于偏移量,则找不到任何内容。此外,我还需要在没有关联的情况下添加一个外连接。

      最后,我发现这个查询有点像 ORM 巫毒,不想走那条路。相反,我从设备序列化程序中排除了histories,并需要使用device ID 进行第二次history 查找。该集合可以分页,并使所有内容都更清洁。

      这两种方法都有效,只是归结为why,您需要做一个查询而不是几个查询。在上面,可能有商业原因通过单个查询更有效地恢复所有内容。对于我的用例,可读性和惯例胜过巫术

      @classmethod
          def get_limited_histories(cls, uuid, limit=10):
      
              count = DeviceHistory.query.filter(DeviceHistory.device_id == uuid).count()
      
              if count > limit:
                  sq = db.session.query(DeviceHistory.created_at) \
                      .filter(DeviceHistory.device_id == Device.uuid) \
                      .order_by(DeviceHistory.created_at.desc()) \
                      .limit(1).offset(limit).correlate(Device)
      
      
              return db.session.query(Device).filter(Device.uuid == uuid) \
                      .outerjoin(DeviceHistory,
                          and_(DeviceHistory.device_id == Device.uuid, DeviceHistory.created_at > sq)) \
                      .options(contains_eager(Device.device_histories)).all()[0]
      
      

      然后它的行为类似于Device.query.get(id)Device.get_limited_histories(id)

      • 享受

      【讨论】:

        猜你喜欢
        • 2014-02-15
        • 2012-06-28
        • 1970-01-01
        • 2014-02-01
        • 2021-09-21
        • 1970-01-01
        • 2014-11-02
        • 1970-01-01
        • 1970-01-01
        相关资源
        最近更新 更多