PostgreSQL下SQLAlchemy更新后返回旧数据的技术问题

问题背景 #

需要在PostgreSQL环境中，基于SQLAlchemy编写查询，实现更新操作后返回数据的旧值。现有User模型定义如下：

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, server_default=text("nextval('\"public\".users_id_seq'::regclass)"))
    last_name = Column(String(255), nullable=False)
    first_name = Column(String(255), nullable=False)
    email = Column(String(255), nullable=False, unique=True)

目标原生SQL #

对应的正确原生SQL语句如下：

update
    users x
        set
            email = 'example@example.com'
        from
            (
            select
                *
            from
                users
            where
                id = 1 for update) y
        where
            x.id = y.id
        returning 
            y.*

错误尝试及问题 #

尝试过以下SQLAlchemy代码，但执行失败：

old_user: User = aliased(User, name='old_user')

update_stmt = update(User) \
    .where(user_old.id == User.id) \
    .where(User.id == 1) \
    .values(email = "example@example.com") \
    .returning(old_user)

return session.execute(update_stmt)

触发错误：
sqlalchemy.exc.NoSuchColumnError: Could not locate column in row for column 'old_user.id'

另外，若直接返回非别名对象，得到的是更新后的新值；尝试return(text("public.old_user.*"))则会触发NotImplementedError。

正确实现方案 #

以下是对应原生SQL逻辑的正确SQLAlchemy实现：

from sqlalchemy import update, aliased
from sqlalchemy.orm import Session
# 导入你的User模型和Base等依赖

def update_user_and_get_old_data(session: Session, user_id: int, new_email: str):
    # 创建User模型的别名，对应原生SQL中的子查询y
    old_user = aliased(User)
    
    # 构建更新语句，完整映射原生SQL逻辑
    update_stmt = (
        update(User)
        # 构建FROM子句，关联锁定的旧数据行
        .from_select(
            ["id"],
            old_user.select().where(old_user.id == user_id).with_for_update()
        )
        .where(User.id == old_user.id)
        .values(email=new_email)
        # 返回别名对象的所有列，即旧数据
        .returning(*old_user.columns)
    )
    
    # 执行语句并获取旧数据
    result = session.execute(update_stmt)
    return result.one()

# 使用示例
old_user_data = update_user_and_get_old_data(session, 1, "example@example.com")

关键说明 #

• 使用aliased(User)创建别名对象，对应原生SQL中锁定旧数据的子查询
• 通过.from_select()方法实现原生SQL的FROM子句逻辑，同时用.with_for_update()添加行锁，避免并发更新问题
• 在.returning()中明确传入*old_user.columns，让SQLAlchemy能正确解析返回的旧数据字段，而非直接返回别名对象
• 执行后通过result.one()获取单行旧数据结果（若需处理多行可调整为result.all()）

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