如何通过SQLAlchemy经典映射将整张backup_registry表映射为单个BackupRegistry实例的属性

要实现将backup_registry表的所有行映射到单个BackupRegistry实例（其中backups属性是Dict[Location, list[Location]]），我们需要跳出“每行对应一个实例”的默认经典映射逻辑，转而通过聚合查询+自定义属性映射来实现。以下是具体步骤：

1. 完善Location类的经典映射（添加关联关系） #

首先，给Location类添加与备份位置的关联关系，这样可以方便地通过Location实例直接获取对应的备份列表。在经典映射中，通过relationship参数定义：

from sqlalchemy import ForeignKey, Integer, String, Column, Table
from sqlalchemy.orm import mapper_registry, relationship

# 原有的Location类和表定义
class Location:
    datacenter_name: str
    path: str

locations = Table(
    "locations",
    mapper_registry.metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("path", String, nullable=False),
    Column("datacenter_name", String, nullable=False),
)

backup_registry = Table(
    "backup_registry",
    mapper_registry.metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("source_location_id", ForeignKey("locations.id"), nullable=False),
    Column("backup_location_id", ForeignKey("locations.id"), nullable=False),
)

# 完善Location的经典映射，添加备份关联
mapper_registry.map_imperatively(
    Location,
    locations,
    properties={
        # 当当前Location作为源时，对应的所有备份Location列表
        "backup_locations": relationship(
            Location,
            secondary=backup_registry,
            primaryjoin=locations.c.id == backup_registry.c.source_location_id,
            secondaryjoin=locations.c.id == backup_registry.c.backup_location_id,
            lazy="joined"  # 可选：预加载关联数据，避免N+1查询
        )
    }
)

2. 定义BackupRegistry类并实现需求 #

如果你的核心需求是单个实例包含所有备份关联，其实不需要把BackupRegistry直接映射到backup_registry表，而是利用已有的Location关联关系，动态构建目标字典。以下是最简洁的实现方式：

from typing import Dict, List
from sqlalchemy.orm import Session

class BackupRegistry:
    def __init__(self, backups: Dict[Location, List[Location]]):
        self.backups = backups

    @classmethod
    def load_from_session(cls, session: Session) -> "BackupRegistry":
        # 查询所有带有备份的源Location，直接通过关联关系获取对应的备份列表
        source_locations = session.query(Location).filter(Location.backup_locations.any()).all()
        backups_dict = {source: source.backup_locations for source in source_locations}
        return cls(backups_dict)

3. 扩展：严格基于经典映射的实现 #

如果必须通过经典映射将BackupRegistry与数据库结构绑定，我们可以把它映射到一个聚合查询结果，再通过属性转换为目标字典：

3.1 创建聚合查询Selectable #

from sqlalchemy import select, func

# 聚合每个源位置对应的备份位置ID列表
aggregated_backups = select(
    backup_registry.c.source_location_id,
    func.array_agg(backup_registry.c.backup_location_id).label("backup_ids")
).group_by(backup_registry.c.source_location_id).subquery()

# 关联locations表，获取源位置的完整信息+对应的备份ID列表
backup_selectable = select(
    locations.c.id.label("source_id"),
    locations.c.datacenter_name.label("source_datacenter"),
    locations.c.path.label("source_path"),
    aggregated_backups.c.backup_ids
).join(aggregated_backups, locations.c.id == aggregated_backups.c.source_location_id)

3.2 经典映射BackupRegistry #

from sqlalchemy.orm import column_property

class BackupRegistry:
    # 映射聚合查询返回的字段
    source_id: int
    source_datacenter: str
    source_path: str
    backup_ids: List[int]

    @property
    def backups(self) -> Dict[Location, List[Location]]:
        # 获取当前实例绑定的会话
        session = Session.object_session(self)
        if not session:
            raise ValueError("BackupRegistry实例必须绑定到SQLAlchemy会话")
        
        # 一次性查询所有涉及的Location实例
        all_location_ids = {self.source_id} | set(self.backup_ids)
        location_map = {loc.id: loc for loc in session.query(Location).filter(Location.id.in_(all_location_ids)).all()}

        # 构建目标字典
        source_loc = location_map[self.source_id]
        return {source_loc: [location_map[bid] for bid in self.backup_ids]}

# 完成经典映射
mapper_registry.map_imperatively(
    BackupRegistry,
    backup_selectable,
    properties={
        "source_id": column_property(backup_selectable.c.source_id),
        "source_datacenter": column_property(backup_selectable.c.source_datacenter),
        "source_path": column_property(backup_selectable.c.source_path),
        "backup_ids": column_property(backup_selectable.c.backup_ids)
    }
)

4. 使用方式 #

from sqlalchemy.orm import Session

# 创建会话
session = Session(engine)

# 方式1：推荐的简洁方式
backup_registry = BackupRegistry.load_from_session(session)
print(backup_registry.backups)

# 方式2：基于聚合映射的方式（需手动合并多个实例的结果）
backup_instances = session.query(BackupRegistry).all()
full_backups = {}
for inst in backup_instances:
    full_backups.update(inst.backups)
backup_registry = BackupRegistry(full_backups)

内容的提问来源于stack exchange，提问作者Sebastián Bevc