使用SQLAlchemy执行含自定义变量的多段MySQL语句报错求助

问题原因 #

你遇到的报错本质是pymysql默认不允许在单个execute()调用中执行多条SQL语句，和SQLAlchemy本身无关，是底层pymysql的限制。另外开头的字符集警告是因为连接时未指定正确字符集，导致读取系统变量时出现编码问题。

解决方案 #

方案1：启用多语句执行支持 #

创建cursor时指定multi=True，即可执行包含多个语句的SQL：

from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://root:root@localhost:3306/test?charset=utf8mb4", echo=True)
c = engine.raw_connection()
# 开启多语句模式
cursor = c.cursor(multi=True)
# 执行多语句SQL并遍历结果
for result in cursor.execute("set @a=0;set @b=null;select @a:=if(@b=num,@a+1,1) as rk,@b:=num from logs;", multi=True):
    if result.with_rows:
        rows = result.fetchall()
        print(rows)
c.commit()
cursor.close()
c.close()

注意：开启多语句模式需警惕SQL注入风险，确保执行的SQL是安全可信的。

方案2：拆分SQL语句，分多次执行 #

将多语句拆分为单独的execute()调用，更安全直观：

from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://root:root@localhost:3306/test?charset=utf8mb4", echo=True)
c = engine.raw_connection()
cursor = c.cursor()

# 逐条执行SQL
cursor.execute("set @a=0;")
cursor.execute("set @b=null;")
cursor.execute("select @a:=if(@b=num,@a+1,1) as rk,@b:=num from logs;")
# 获取查询结果
rows = cursor.fetchall()
print(rows)

c.commit()
cursor.close()
c.close()

方案3：指定字符集解决编码警告 #

在数据库连接URL末尾添加?charset=utf8mb4，解决开头出现的字符集警告，建议配合上述方案一起使用。

补充说明 #

• MySQL会话级变量@a、@b在连接未关闭时会保持状态，因此分多次执行也能正常工作。
• 若使用SQLAlchemy的ORM/Core层，可通过text()配合execute()分步执行：

from sqlalchemy import create_engine, text

engine = create_engine("mysql+pymysql://root:root@localhost:3306/test?charset=utf8mb4", echo=True)
with engine.connect() as conn:
    conn.execute(text("set @a=0;"))
    conn.execute(text("set @b=null;"))
    result = conn.execute(text("select @a:=if(@b=num,@a+1,1) as rk,@b:=num from logs;"))
    rows = result.fetchall()
    print(rows)
    conn.commit()

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