SQLAlchemy Upsert问题：插入正常但更新未生效求助

问题重现 #

会话管理器实现 #

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.exc import SQLAlchemyError
import contextlib

class DatabaseSessionManager:
    def __init__(self, connection_string):
        self.engine = create_engine(connection_string)
        self.session_factory = sessionmaker(bind=self.engine)
        self.Session = scoped_session(self.session_factory)

    @contextlib.contextmanager
    def session_scope(self):
        """Provide a transactional scope around a series of operations."""
        session = self.Session()
        try:
            yield session
            session.commit()
        except SQLAlchemyError as e:
            session.rollback()
            raise
        finally:
            session.close()

Upsert函数实现 #

def upsert(session, customer):
    try:
        # Attempt to find the customer by id
        existing_customer = session.query(Customer).filter_by(id=customer.id).first()

        if existing_customer:
            # Update existing customer
            for key, value in vars(customer).items():
                if hasattr(existing_customer, key) and value is not None:
                    setattr(existing_customer, key, value)
        else:
            # Create new customer
            session.add(customer)

    except SQLAlchemyError as e:
        print(f"Error occurred: {e}")
        session.rollback()
    finally:
        session.close()

测试现象 #

• 插入新ID的Customer对象时，数据可正常写入数据库：

  from datetime import datetime
  test_customer = Customer(
      id=6962763399483,  # Manually assign a unique ID
      first_name='Melanie',
      last_name='Dopp',
      email='john.doe@example.com',
      orders_count='9000',
      total_spent='19000',
      created_at=datetime.utcnow(),
      last_order_created_at=datetime.utcnow()  # Assign a relevant datetime or None
  )
  

• 编辑同一ID的对象时，变更无法提交；重新查询返回旧值：

  with dbm.session_scope() as session:
      print(session.query(Customer).filter_by(id=6962763399483).first().last_name)
  # returns Ruby
  

问题原因 #

1. 会话被手动提前关闭：Upsert函数的finally块中调用了session.close()，但会话的生命周期由session_scope上下文管理器负责，手动关闭会导致上下文后续的session.commit()失效——会话已关闭，提交操作无法执行。
2. Scoped Session误用：scoped_session是线程局部的会话实例，手动关闭后，后续操作可能复用旧的会话缓存，导致查询返回未更新的旧数据。
3. 属性遍历的隐患：用vars(customer)遍历属性时，会包含SQLAlchemy实例的私有属性（如_sa_instance_state），虽然有hasattr判断，但可能误操作无关属性，也容易遗漏或错误更新字段。

修复方案 #

1. 修正Upsert函数，移除手动关闭会话 #

删除finally块中的session.close()，让上下文管理器管理会话生命周期；同时优化字段更新逻辑，避免遍历私有属性：

def upsert(session, customer):
    try:
        existing_customer = session.query(Customer).filter_by(id=customer.id).first()
        if existing_customer:
            # 明确指定需要更新的字段，避免操作私有属性
            update_fields = ['first_name', 'last_name', 'email', 'orders_count', 'total_spent', 'last_order_created_at']
            for field in update_fields:
                value = getattr(customer, field)
                if value is not None:
                    setattr(existing_customer, field, value)
        else:
            session.add(customer)
    except SQLAlchemyError as e:
        print(f"Error occurred: {e}")
        session.rollback()
        # 重新抛出异常，让上下文管理器统一处理
        raise

2. 规范会话使用流程 #

所有数据库操作必须在session_scope的上下文块内完成，确保会话的提交、回滚、关闭由管理器统一处理：

# 正确的更新操作
with dbm.session_scope() as session:
    update_customer = Customer(
        id=6962763399483,
        last_name='Dopp'  # 仅传入需要更新的字段
    )
    upsert(session, update_customer)

# 查询时使用新的会话上下文，避免读取旧缓存
with dbm.session_scope() as session:
    target_customer = session.query(Customer).filter_by(id=6962763399483).first()
    print(target_customer.last_name)  # 应返回Dopp

3. 可选：简化会话管理器（非多线程场景） #

如果不是多线程环境，无需使用scoped_session，可简化管理器实现：

class DatabaseSessionManager:
    def __init__(self, connection_string):
        self.engine = create_engine(connection_string)
        self.session_factory = sessionmaker(bind=self.engine)

    @contextlib.contextmanager
    def session_scope(self):
        session = self.session_factory()
        try:
            yield session
            session.commit()
        except SQLAlchemyError as e:
            session.rollback()
            raise
        finally:
            session.close()

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