Flask-SQLAlchemy 3.1.1执行原生MySQL查询报错求助

问题本质 #

升级到Flask-SQLAlchemy 3.x后，get_engine()的参数逻辑发生变化：不再接受current_app作为参数，传入后会导致绑定键变为未绑定的LocalProxy，触发KeyError；同时db.engines返回的是EngineWrapper包装对象，无法直接调用execute，必须先获取原生SQLAlchemy引擎实例。

修复代码示例 #

默认数据库执行原生查询 #

替换原有代码：

db.get_engine(current_app).execute(qry_string)

改为：

# 先获取原生引擎，用text包裹查询字符串（SQLAlchemy 2.x强制要求）
result = db.engines[None].engine.execute(db.text(qry_string))

指定数据库绑定执行原生查询 #

替换原有代码：

db.get_engine(current_app, my_database_target).execute(qry_string)

改为：

result = db.engines[my_database_target].engine.execute(db.text(qry_string))

重要注意事项 #

SQLAlchemy 2.x版本开始，执行原生字符串查询必须用text()方法包裹，否则会抛出ArgumentError。直接使用Flask-SQLAlchemy实例的text()方法即可，无需额外导入SQLAlchemy的text类。

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