SQLAlchemy报错ObjectDereferencedError：无法触发ProjectDTO.statuses变更事件

背景代码 #

DTO类定义 #

class ProjectDTO(DB):
    __tablename__ = 'projects'
    __table_args__ = {'schema': 'postgres'}

    id = Column(Integer, primary_key=True)
    serial = Column(UUID(as_uuid=True), unique=True, nullable=False, server_default=text('(uuid_generate_v4())'))
    team_engineer_serial = Column(UUID(as_uuid=True), ForeignKey('postgres.team.serial'))
    team_tester_serial = Column(UUID(as_uuid=True), ForeignKey('postgres.team.serial'))

    statuses = relationship('ProjectStatusDTO', back_populates='project', foreign_keys='[ProjectStatusDTO.project_serial]')
    team_engineer = relationship('TeamDTO', backref='project_as_engineer', foreign_keys=[team_engineer_serial])
    team_tester = relationship('TeamDTO', backref='project_as_tester', foreign_keys=[team_tester_serial])

class ProjectStatusDTO(DB):
    __tablename__ = 'project_statuses'
    __table_args__ = {'schema': 'postgres'}

    id = Column(Integer, primary_key=True)
    serial = Column(UUID(as_uuid=True), unique=True, nullable=False, server_default=text('(uuid_generate_v4())'))
    project_serial = Column(UUID(as_uuid=True), ForeignKey('postgres.projects.serial'))
    name = Column(String(255), nullable=False)
    is_initial = Column(Boolean)

    project = relationship('ProjectDTO', back_populates='statuses', foreign_keys=[project_serial])

查询函数实现 #

def _get_project(self, ctx: Context, project_code: str = None, team_engineer_serial: str = None, with_initial_status: bool = False):
    if not project_code and not team_engineer_serial:
        raise BadRequestError("one of 'project_code' or 'team_engineer_serial' must be provided")

    query = ctx.get_db_catalog().query(ProjectDTO)

    if project_code:
        query = query.filter(ProjectDTO.code == project_code)
    else:
        query = query.filter(ProjectDTO.team_engineer_serial == team_engineer_serial)

    if with_initial_status:
        query = query.join(ProjectDTO.statuses).filter(ProjectStatusDTO.is_initial == True)

    query = query.options(
        joinedload(ProjectDTO.statuses),
        joinedload(ProjectDTO.team_engineer)
    )

    result = query.first()

    if not result:
        return None

    return ProjectEntity().from_dto(result)

异常信息 #

执行上述函数时触发以下错误：

sqlalchemy.orm.exc.ObjectDereferencedError: Can't emit change event for attribute 'ProjectDTO.statuses' - parent object of type <ProjectDTO> has been garbage collected.

问题原因及解决方法 #

• 核心问题：当with_initial_status=True时，同时使用join(ProjectDTO.statuses)和joinedload(ProjectDTO.statuses)会导致SQLAlchemy处理关联对象时出现冲突：

  1. join操作将project_statuses表纳入查询，会直接绑定ProjectDTO对象的statuses属性到查询结果的行数据；
  2. joinedload作为预加载机制，会尝试单独加载关联的statuses集合，两种操作对同一属性的加载逻辑冲突，导致对象生命周期管理混乱，最终触发垃圾回收相关异常。

• 针对性解决：

  1. 若需求是筛选出包含初始状态的项目，同时加载该项目所有状态：
     移除join操作，改用exists子查询过滤项目，避免直接关联表导致的属性绑定冲突：

     if with_initial_status:
         subquery = exists().where(
             and_(
                 ProjectStatusDTO.project_serial == ProjectDTO.serial,
                 ProjectStatusDTO.is_initial == True
             )
         )
         query = query.filter(subquery)
     

  2. 若需求是仅加载项目的初始状态：
     保留join但移除joinedload(ProjectDTO.statuses)，或用contains_eager替代joinedload，明确告知SQLAlchemy使用join后的结果填充关联属性：

     if with_initial_status:
         query = query.join(ProjectDTO.statuses).filter(ProjectStatusDTO.is_initial == True)
         query = query.options(
             contains_eager(ProjectDTO.statuses),
             joinedload(ProjectDTO.team_engineer)
         )
     

补充说明 #

• joinedload用于预加载关联对象，不影响主查询筛选逻辑，仅减少后续N+1查询；
• join用于筛选主表数据，同时会改变关联属性的加载方式；
• 两者同时作用于同一关联属性时，会引发SQLAlchemy内部对象状态管理冲突，进而触发异常。

内容的提问来源于stack exchange，提问作者dev-x