SQLAlchemy map_imperatively映射子查询返回结果不一致问题

问题基本信息 #

• 运行环境：Python 3.10 + SQLAlchemy 异步ORM
• 异常表现：
  • 对经map_imperatively命令式映射的SectorEntity类执行ORM查询，返回2条重复记录
  • 返回的对象上存在数据类未定义的total_amount_undrawn属性
  • 直接执行映射对应的sector_entity原生select语句可以返回2条正确的不同记录
  • 打印ORM生成的SQL直接在数据库执行，结果和原生查询完全一致

复现代码 #

1. types.py 数据类定义 #

from dataclass import dataclass
from typing import Optional

@dataclass
class SectorEntity:
    path: str
    year: int
    sector: str
    entity_count: int
    sector_value: Optional[float]

def __str__(self) -> str:
    return """
SECTOR ENTITY
{self.path=}
{self.year=}
{self.sector=}
{self.entity_count=}
{self.sector_value=}
"""

2. schema.py 表结构与映射配置 #

from sqlalchemy import Column, ForeignKey, Metadata, Table, func, select
from sqlalchemy.orm import registry, relationship
from sqlalchemy.types import DateTime, Float, Integer, String


from .types import SectorEntity

metadata = MetaData()

entity_last_updated = Table(
    "entity_last_updated",
    metadata,
    Column("entity_guid", String(), primary_key=True),
    Column("sector", String()),
    Column("last_updated", DateTime(), nullable=False),
)

entity_unit = Table(
    "entity_unit",
    metadata,
    Column("entity_guid",
        String(),
        ForeignKey("entity_last_updated.entity_guid", ondelete="CASCADE"),
        primary_key=True,
    Column("year", Integer(), primary_key=True),
    Column("path", String(), primary_key=True),
    Column("sector_value", Float(), nullable=True),
)

sector_entity = (
    select(
        entity_unit.c.path,
        entity_unit.c.year,
        entity_unit.c.sector,
        func.count(entity_unit.c.entity_guid).label("entity_count"),
        func.sum(entity_unit.c.sector_value).label("sector_value"),
    )
    .join(entity_unit)
    .group_by(entity_unit.c.sector, entity_unit.c.path, entity_unit.c.year)
)

def start_mappers() -> None:
    mapper_registry = registry()
    mapper_registry.map_imperatively(
        SectorEntity,
        sector_entity.subquery(),
    )

3. query.py 查询逻辑 #

from typing import List
from .types import SectorEntity
from .schema import sector_entity
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

async def select_sector_entity(
    *,
    session: AsyncSession,
) -> List[SectorEntity]:
    assert session.in_transaction()

    base_query = select(SectorEntity)
    result = await session.execute(base_query)
    cursor = result.scalars()
    output = list(cursor)
    cursor = (await session.execute(sector_entity)).fetchall()
    for obj in output:
        print(str(obj))
    for obj in cursor:
        print(str(obj))
    return output

运行输出 #

----------------------------------- Captured stdout call -----------------------------------
SECTOR ENTITY
    self.path='Gary'
    self.year=2016
    self.sector='709492'
    self.entity_count=2
    self.sector_value=4.5
    self.total_amount_undrawn=14365627661.01394


SECTOR ENTITY
    self.path='Gary'
    self.year=2016
    self.sector='709492'
    self.entity_count=2
    self.sector_value=4.5

('Gary', 2016, '709492', 2, 4.5)
('Gary', 2016, '935252', 1, 1.3)

根因分析 #

1. 核心原因：映射未指定主键，ORM身份映射逻辑失效
   SQLAlchemy ORM默认通过主键值实现*身份映射（Identity Map）*机制：同一会话中，相同主键值的数据库行会被映射为同一个Python对象，查询时如果发现会话中已存在对应主键的对象，会直接返回缓存的对象，不会重新构造。
   本次映射的目标是聚合查询生成的子查询，没有物理主键，代码中也未显式指定主键列，ORM会默认选取子查询返回的第一列（即path列）作为主键。测试数据中两条记录的path值均为Gary，被ORM识别为同一条记录，直接返回缓存的第一个对象，因此出现重复记录。
   而对象上莫名出现的total_amount_undrawn属性，是会话中其他查询生成的、主键值同为Gary的缓存对象的属性，被直接复用到了本次返回结果中。
2. 代码存在多处语法与逻辑错误，放大异常
   • types.py中dataclass导入错误，标准库模块名为dataclasses（带尾缀s）；__str__方法缩进错误，定义在类外部，未绑定到类上
   • schema.py中entity_unit表的entity_guid列定义缺少闭合括号，会导致列嵌套定义异常
   • sector_entity查询逻辑错误：未指定join主表、未明确join条件，会生成无意义的自连接SQL；sector列实际属于entity_last_updated表，代码中错误引用了entity_unit.c.sector

修复方案 #

1. 修复基础语法与逻辑错误
   • 修正导入与类方法缩进：将from dataclass import dataclass改为from dataclasses import dataclass，将__str__方法缩进至SectorEntity类内部
   • 补全entity_unit表entity_guid列的闭合括号
   • 修正sector_entity查询的join逻辑与列引用，明确主表和关联条件：

     sector_entity = (
         select(
             entity_unit.c.path,
             entity_unit.c.year,
             entity_last_updated.c.sector,
             func.count(entity_unit.c.entity_guid).label("entity_count"),
             func.sum(entity_unit.c.sector_value).label("sector_value"),
         )
         .join(entity_unit, entity_last_updated.c.entity_guid == entity_unit.c.entity_guid)
         .group_by(entity_last_updated.c.sector, entity_unit.c.path, entity_unit.c.year)
     )
     

2. 命令式映射时显式指定子查询的主键列
   通过primary_key参数指定可以唯一标识聚合结果每一行的列组合（即group by使用的三个字段），让ORM可以正确区分不同行：

def start_mappers() -> None:
    mapper_registry = registry()
    sector_subq = sector_entity.subquery()
    mapper_registry.map_imperatively(
        SectorEntity,
        sector_subq,
        primary_key=[
            sector_subq.c.path,
            sector_subq.c.year,
            sector_subq.c.sector
        ]
    )

3. （可选）查询时添加刷新缓存选项
   如果不需要复用会话缓存，可以在查询时添加populate_existing执行选项，强制ORM用数据库返回的最新数据覆盖缓存，避免脏数据影响：

base_query = select(SectorEntity).execution_options(populate_existing=True)

内容的提问来源于stack exchange，提问作者baron