SQLAlchemy defer.raiseload=True参数未生效，求解决方案

问题现象 #

设置defer(User.address, raiseload=True)后，访问未加载的address字段未触发预期的InvalidRequestError，反而能直接打印字段值。

原因分析 #

在同一个会话中，插入并提交User对象后，这些实例会被保存在会话的**identity map（身份映射）**中。后续查询相同主键的对象时，SQLAlchemy默认会直接返回缓存中的完整对象，不会执行新的SELECT语句，也不会应用defer/raiseload的字段加载规则。

解决方案 #

有两种方式可以绕过会话缓存，让raiseload=True生效：

方案1：重新开启新会话

提交数据后关闭当前会话，重新创建会话执行查询，此时缓存中没有目标对象，会从数据库加载指定字段：

import random
import string
from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, defer

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'user'

    user_id: Mapped[int] = mapped_column(primary_key=True)
    firstname: Mapped[str]
    lastname: Mapped[str]
    address: Mapped[str]

engine = create_engine('sqlite:///:memory:', echo=True)
Base.metadata.create_all(engine)

# 插入数据的会话
with Session(engine) as session:
    data = []
    for i in range(1, 10):
        user = User(
            user_id=i,
            firstname=''.join(random.choices(string.ascii_lowercase, k=16)),
            lastname=''.join(random.choices(string.ascii_lowercase, k=16)),
            address=''.join(random.choices(string.ascii_lowercase, k=32)),
        )
        data.append(user)
    session.add_all(data)
    session.commit()

# 新会话执行查询
with Session(engine) as session:
    user = session.scalar(select(User).options(defer(User.address, raiseload=True)))
    print(user.address)  # 触发InvalidRequestError

方案2：使用populate_existing()选项

在查询中添加populate_existing()，强制SQLAlchemy忽略identity map中的缓存，从数据库重新加载对象并应用字段加载规则：

import random
import string
from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, defer, populate_existing

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'user'

    user_id: Mapped[int] = mapped_column(primary_key=True)
    firstname: Mapped[str]
    lastname: Mapped[str]
    address: Mapped[str]

engine = create_engine('sqlite:///:memory:', echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    data = []
    for i in range(1, 10):
        user = User(
            user_id=i,
            firstname=''.join(random.choices(string.ascii_lowercase, k=16)),
            lastname=''.join(random.choices(string.ascii_lowercase, k=16)),
            address=''.join(random.choices(string.ascii_lowercase, k=32)),
        )
        data.append(user)
    session.add_all(data)
    session.commit()

    # 添加populate_existing()强制刷新
    user = session.scalar(select(User).options(
        defer(User.address, raiseload=True),
        populate_existing()
    ))
    print(user.address)  # 触发InvalidRequestError

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