升级至SQLAlchemy 2.0后使用text()无法向MySQL执行INSERT/UPSERT

核心排查方向：事务未提交 #

SQLAlchemy 2.x的Session默认处于事务模式，执行session.execute()后不会自动提交事务，若未手动提交，所有操作会在Session关闭时回滚，这是最常见的原因：

• 执行完语句后添加提交操作：

cursor = session.execute(text(my_query))
session.commit()  # 必须手动提交事务才能写入数据

如果是测试场景，也可以用session.begin()上下文管理器自动处理提交/回滚：

with session.begin():
    session.execute(text(my_query))

其他排查点 #

1. 检查自动提交配置（不推荐ORM场景使用）
   若希望模拟旧版本的自动提交行为，可以在创建引擎时开启自动提交，但可能引发数据一致性问题，需谨慎使用：

   connection = create_engine(constring, connect_args={"autocommit": True})
   

2. 改用参数化写法（可选）
   虽然你的语句直接执行没问题，但参数化写法能避免潜在的语法解析问题，同时提升安全性：

   from sqlalchemy import bindparam
   
   my_query = text("""
       INSERT INTO test (`col_a`, `col_b`)
       VALUES (:a1, :b1), (:a2, :b2), (:a3, :b3) as nd
       ON DUPLICATE KEY 
       UPDATE `col_a` = nd.`col_a`, `col_b` = nd.`col_b`
   """)
   session.execute(my_query, {"a1":4, "b1":7, "a2":5, "b2":8, "a3":6, "b3":9})
   session.commit()
   

3. 使用官方推荐的Session创建方式
   可以改用sessionmaker创建Session，替代create_session，这是SQLAlchemy 2.x的标准用法：

   from sqlalchemy.orm import sessionmaker
   
   Session = sessionmaker(bind=connection)
   session = Session()
   session.execute(text(my_query))
   session.commit()
   

验证方法 #

执行完操作后，若数据未写入，可尝试执行session.rollback()：如果执行后数据依然不存在，说明是未提交的问题；如果数据出现，再排查其他配置问题。

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