在 SQLAlchemy 模型参数中优化访问 n 个关系

Question

问题的标题可能令人困惑，但我相信我已经在下面描述了这个问题。

我有各种使用 SQLAlchemy 构建的声明性“测试类型”模型。每个“测试类型”模型都与位于“位置”上的一件“设备”相关联。因此，每个测试类型都有一个“location.name”参数，该参数位于远离实际测试模型的不同数量的关系中。我需要这个“location.name”值与每个“测试”对象相关联。

我现在设置它的方式在查询测试时会发出大量 SQL。我想这是因为关系级联。我将所有测试显示在一个表中，并且我希望能够根据测试关联的位置以某种方式过滤测试。

这个模型配置对我来说是新的，我不知道如何使用原始 SQL 与之交互，我只知道如何使用模型。

我想知道获取特定测试实例的位置名称参数的最佳方法是什么。请注意，每个测试的get_json() 函数都会返回location.name，尽管我希望有一种更好的方法可以在过滤所有测试类型时发出更少的SQL并且更简单。

希望下面的模型定义可以清楚地说明这一点

class Location(Model):
    __tablename__ = 'locations'
    id = Column(Integer, Sequence('location_id_seq'), primary_key=True)
    name = Column(String(50), unique=True)
    ...


class EquipmentFoo(Model):
    __tablename__ = 'equipmentfoos'
    id = Column(Integer, Sequence('equipmentfoo_id_seq'), primary_key=True)
    location_id = Column(Integer, ForeignKey('locations.id'))
    location = relationship('Location', backref='equipmentfoos')
    footests = relationship('FooTest', backref='equipmentfoo')
    ...


class EquipmentBar(Model):
    __tablename__ = 'equipmentbars'
    id = Column(Integer, Sequence('equipmentbar_id_seq'), primary_key=True)
    equipmentfoo_id = Column(Integer, ForeignKey('equipmentfoos.id'))
    equipmentfoo = relationship('EquipmentFoo', backref='equipmentbars')
    bartests = relationship('BarTest', backref='equipmentbar')
    ...


class EquipmentZab(Model):
    __tablename__ = 'equipmentzabs'
    id = Column(Integer, Sequence('equipmentzab_id_seq'), primary_key=True)
    equipmentbar_id = Column(Integer, ForeignKey('equipmentbars.id'))
    equipmentbar = relationship('EquipmentBar', backref='equipmentzabs')
    zabtests = relationship('ZabTest', backref='equipmentzab')
    ...

这里有设备模型以及它们与位置的关系。还包括它们与下面各自测试的关系。

class HasId(object):
    @declared_attr
    def id(cls):
        return Column('id', Integer, Sequence('test_id_seq'), primary_key=True)
    @declared_attr
    def status(cls):
        return Column('status', String(50))
    ...


class TestParent(HasId, Model):
    __tablename__ = 'tests'
    discriminator = Column(String(50))
    __mapper_args__ = {'polymorphic_on': discriminator}
    ...
    def parent_json(self):
        return {'id': self.id,
                'status': self.status,
                ...
                }         


class FooTest(TestParent):
    __tablename__ = 'footests'
    __mapper_args__ = {'polymorphic_identity': 'footests'}
    id = Column(Integer, ForeignKey('tests.id'), primary_key=True) cascade='save-update, merge')
    pressure_start = Column(Float)
    ...
    def get_json():
        my_json = {'location': self.equipmentfoo.location.name,
                   'pressure_start': self.pressure_start,
                   ...
                   }
        parent_json = super(FooTest, self).parent_json()
        my_json.update(parent_json)
        return my_json


class BarTest(TestParent):
    __tablename__ = 'bartests'
    __mapper_args__ = {'polymorphic_identity': 'bartests'}
    id = Column(Integer, ForeignKey('tests.id'), primary_key=True) cascade='save-update, merge')
    hatch_value = Column(Boolean)
    ...
    def get_json():
        my_json = {'location': self.equipmentbar.equipmentfoo.location.name,
                   'hatch_value': self.hatch_value,
                   ...
                   }
        parent_json = super(BarTest, self).parent_json()
        my_json.update(parent_json)
        return my_json


class ZabTest(TestParent):
    __tablename__ = 'zabtests'
    __mapper_args__ = {'polymorphic_identity': 'zabtests'}
    id = Column(Integer, ForeignKey('tests.id'), primary_key=True) cascade='save-update, merge')
    safety_check = Column(Boolean)
    ...
    def get_json():
        my_json = {'location': self.equipmentzab.equipmentbar.equipmentfoo.location.name,
                   'safety_check': self.hatch_value,
                   ...
                   }
        parent_json = super(ZabTest, self).parent_json()
        my_json.update(parent_json)
        return my_json

Answer 1

默认情况下，不加载关系。这意味着当您第一次访问关系时，会发出一个查询来获取数据，而这又不会加载 its 关系。对于您的方案，对于 n ZabTest 实例，它会发出 4n 查询以获取位置名称。

提高许多小型操作效率的一般方法是通过批处理，它将许多操作的开销合二为一。在 SQL 中执行此操作的方法是一次查询多行而不是一次查询一行。 SQLAlchemy 通过eager loading 公开了这种能力：

session.query(ZabTest).options(
    joinedload(ZabTest.equipmentzab)
    .joinedload(EquipmentZab.equipmentbar)
    .joinedload(EquipmentBar.equipmentfoo)
    .joinedload(EquipmentFoo.location))

这样做的目的是填充整个self.equipmentbar.equipmentfoo.location.name 链，包括ZabTest 本身，在单个查询中。

或者，您可以将 SQLAlchemy 配置为始终急切加载：

zabtests = relationship('ZabTest', backref=backref('equipmentzab', lazy="joined"))

预先警告：由于您还使用继承，急切加载语法可能会变得很麻烦。始终检查发出的 SQL 以确保您正确加入。