sqlalchemy 查询以获取外键（父）组合的最新数据答案

【问题标题】：sqlalchemy query to fetch latest data for a foreign key (parent) combinationsqlalchemy 查询以获取外键（父）组合的最新数据
【发布时间】：2020-09-15 05:14:02
【问题描述】：

我有一些模型有一些父类，其中有一个名为 Measurement 的叶节点类，它每秒都会用新数据更新。

我想从另一个数据库连接中获取新数据并定期更新显示。

目前我查询根父级，然后遍历 ORM 关系列表（测量是动态加载的）。当我开始测量时，我尝试使用value = parent.measurements[-1].value，但这很慢。我假设在切片之前所有内容都已读入列表。

我将代码更改为显式查询，使其更快。例如value = parent.measurements.order_by( Measurement.id.desc() ).first().value

使用 SQLAlchemy，如何通过一个 SQL 事务获取每个父类组合的所有最新测量值？

我假设某种连接/过滤器（或者可能是视图）是答案？

下面是一个示例类模型。假设我在数据库中填充了 2 个收集器，每个收集器 3 个设备，每个设备有 4 个 ADC。

使用 SQLAlchemy，如何通过一次 SQL 事务获取每个收集器的每个设备的每个 adc 的所有最新测量值？

class Collector( Model ) :
    __tablename__   = 'collector'
    id              = Column( Integer, primary_key=True )
    #! --- Relationships ---
    devices         = relationship( 'Device', back_populates='collector' )


class Device( Model ) :
    __tablename__   = 'device'
    id              = Column( Integer, primary_key=True )
    collector_id    = Column( Integer, ForeignKey( 'collector.id' ) )
    #! --- Relationships ---
    collector       = relationship( 'Collector', back_populates='devices', uselist=False )
    adcs            = relationship( 'ADC', back_populates='device' )


class ADC( Model ) :
    __tablename__   = 'adc'
    id              = Column( Integer, primary_key=True )
    channel         = Column( Integer )
    device_id       = Column( Integer, ForeignKey( 'device.id' ) )
    #! --- Relationships ---
    device          = relationship( 'Device', back_populates='adcs', uselist=False )
    measurements    = relationship( 'Measurement', back_populates='adc', lazy='dynamic' )


class Measurement( Model ) :
    __tablename__   = 'measurement'
    id              = Column( Integer, primary_key=True )
    timestamp       = Column( DateTime, nullable=False )
    value           = Column( Integer, nullable=False )
    adc_id          = Column( Integer, ForeignKey( 'adc.id' ) )
    #! --- Relationships ---
    adc             = relationship( 'ADC', back_populates='measurements', uselist=False )

【问题讨论】：

你想要什么结果？最新的Measurement 对象列表是否足够？
是的 - 让我们忽略收集器 - 我想做两件事。 (1) 获取最新测量的列表 - 对于每个设备的每个 ADC 通道。 (2) 获取指定设备的每个 ADC 的最新测量值列表。一旦我能够实现这两个结果，我可能可以计算出任何其他变化。一个好处是如何使它成为数据库“视图”

标签： python sqlalchemy

【解决方案1】：

我找到了使用混合属性解决问题的方法。

class ADC( Model ) :
   ...
    @hybrid_property
    def latest_measurement( self ) -> float :
        measurement = self.measurements.order_by( Measurement.id.desc() ).first()
        return measurement

然后是这样的：

    collector = session.query( Collector )
    for device in collector.devices :
        for adc in device.adcs :
            measurement = adc.latest_measurement
            do_something( measurement )

可能还有另一种方法可以做到这一点（也许在测量模型上使用混合方法并将匹配标准作为参数传递并执行一些.filter() 调用？）

【讨论】：