python - 在 SQLAlchemy ORM 中，如何获取在 session.query().group_by() 中使用的主键列？

Question

问题

如何获取模型类的所有主键列及其多态层次结构中所有父级的列表，以便我可以在 session.query().group_by 中使用它？

详细信息

在 SQLAlchemy ORM 中，如果我查询的类是多态层次结构的一部分，并且我想要对其主键进行 GROUP BY，我还必须 GROUP BY多态层次结构中其父级的所有主键。

想象一下受 Mapping Class Inheritance Hierarchies 启发的以下设置sqlalchemy 文档的部分:

class Employee(Base):
    __tablename__ = 'employee'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    type = Column(String(50))

    __mapper_args__ = {
        'polymorphic_identity':'employee',
        'polymorphic_on':type
    }

class Engineer(Employee):
    __tablename__ = 'engineer'
    id = Column(Integer, ForeignKey('employee.id'), primary_key=True)
    engineer_name = Column(String(30))

    __mapper_args__ = {
        'polymorphic_identity':'engineer',
    }

class EngineeringTask(Base):
    __tablename__ = 'engineering_task'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))

    assigned_engineer_id = Column(ForeignKey(Engineer.id))
    assigned_engineer = relationship(Engineer, backref=backref("assigned_tasks", lazy="dynamic"))

如果我想做这样的查询

session.query(
    Engineer,
).join(
    Engineer.assigned_tasks,
).add_columns(
    func.count(EngineeringTask.id).label('assigned_task_count'),
).group_by(Engineer.id, Employee.id)

这样的查询，从 Engineer 和 Employee 中选择所有列，但不聚合它们，在 PostgreSQL because 中是可能的。 (强调我的):

When GROUP BY is present (...) it is not valid for the SELECT list expressions to refer to ungrouped columns except within aggregate functions or when the ungrouped column is functionally dependent on the grouped columns, since there would otherwise be more than one possible value to return for an ungrouped column. A functional dependency exists if the grouped columns (or a subset thereof) are the primary key of the table containing the ungrouped column.

但它要求我知道/关心/记住我选择的映射类的所有主键列及其多态层次结构中的所有父级(Engineer.id 和 Employee .id 在本例中)

如何动态获取多态层次结构中 Engineer 的所有主键列和所有父级列的列表？

Answer 1

到目前为止我能想到的最好的就是这个函数:

from sqlalchemy import inspect

def polymorphic_primary_keys(cls): 
    mapper = inspect(cls)
    yield from (column for column in mapper.columns if column.primary_key) 
    if mapper.inherits is not None: 
        yield from polymorphic_primary_keys(mapper.inherits.entity)

像这样使用它:

query = session.query(
    Engineer,
).join(
    Engineer.assigned_tasks,
).add_columns(
    func.count(EngineeringTask.id).label('assigned_task_count'),
).group_by(*polymorphic_primary_keys(Engineer))

其中inspect(Engineer)返回Engineer.__mapper__，它是Mapper包含:

• Mapper.columns ，映射模型列的迭代器/字典/属性访问器
• Mapper.inherits ，这是父实体的映射器(如果有)。

由于以下几个原因，它令人不满意:

• 有Mapper.primary_key它应该是主键上的迭代器，但我不能使用它，因为它返回最顶层父级的主键，而不是当前映射的实体，所以我必须迭代 Mapper.columns 并检查 primary_key 属性。

• 有一个 Mapper.polymorphic_iterator()方法和Mapper.self_and_descendants属性，它们都返回一个包含当前映射器和当前实体的所有“下游”子类的映射器的迭代器。为什么“上游”父类(super class)没有等效的迭代器？

但是它完成了工作......