使用SQLAlchemy的text()构造在SQL Server中创建临时表

能否使用SQLAlchemy的text()构造（将SQL原样传递至数据库）在SQL Server中创建临时表？执行以下Python代码时返回错误：This result object does not return rows. It has been closed automatically.

用户提供的代码：

import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy.sql import text

# DWH as User DSN (ODBC)
engine = create_engine('mssql+pyodbc://DWH')

con = engine.connect()

sql_query_text = text(
"""
CREATE TABLE #employee
(
emp_id INT IDENTITY PRIMARY KEY,
last_name VARCHAR(30) NOT NULL,
first_name VARCHAR(30) NOT NULL,
hire_date DATETIME NOT NULL,
job_title VARCHAR(50) NOT NULL
);


INSERT INTO #employee
VALUES ('Smith', 'James', '3/1/2016', 'Staff Accountant'),
('Williams', 'Roberta', '2/7/2004', 'Sr. Software Engineer'),
('Washington','Mark','8/19/2014', 'HelpDesk Technician');


SELECT * FROM #employee;
"""
)

rs = con.execute(sql_query_text)

# convert result to DataFrame
df = pd.DataFrame(rs.fetchall())

# Error
sqlalchemy.exc.ResourceClosedError:
This result object does not return rows. It has been closed automatically.

报错的核心原因是：同一个text()块里包含多个SQL语句时，SQLAlchemy默认只返回第一个语句的执行结果——而CREATE TABLE和INSERT都是不返回行的操作，等到你调用fetchall()获取SELECT的结果时，结果集已经被自动关闭了。

下面提供两种可行的解决办法：

办法1：开启多结果集支持 #

在text()对象上添加multiple_results=True执行选项，让SQLAlchemy能处理多个结果集，之后跳过前两个无返回的操作结果，获取最后SELECT的结果：

import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy.sql import text

engine = create_engine('mssql+pyodbc://DWH')
con = engine.connect()

# 添加execution_options开启多结果集支持
sql_query_text = text("""
CREATE TABLE #employee
(
emp_id INT IDENTITY PRIMARY KEY,
last_name VARCHAR(30) NOT NULL,
first_name VARCHAR(30) NOT NULL,
hire_date DATETIME NOT NULL,
job_title VARCHAR(50) NOT NULL
);


INSERT INTO #employee
VALUES ('Smith', 'James', '3/1/2016', 'Staff Accountant'),
('Williams', 'Roberta', '2/7/2004', 'Sr. Software Engineer'),
('Washington','Mark','8/19/2014', 'HelpDesk Technician');


SELECT * FROM #employee;
""").execution_options(multiple_results=True)

rs = con.execute(sql_query_text)
# 跳过前两个无返回的结果集
rs.nextset()
rs.nextset()
# 获取SELECT的结果并转为DataFrame
df = pd.DataFrame(rs.fetchall())

办法2：拆分SQL语句独立执行 #

把创建表、插入数据、查询这三个操作拆分成单独的execute()调用，确保每一步执行完成后再处理下一个操作，这样能精准获取查询的结果集：

import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy.sql import text

engine = create_engine('mssql+pyodbc://DWH')
con = engine.connect()

# 1. 创建临时表
con.execute(text("""
CREATE TABLE #employee
(
emp_id INT IDENTITY PRIMARY KEY,
last_name VARCHAR(30) NOT NULL,
first_name VARCHAR(30) NOT NULL,
hire_date DATETIME NOT NULL,
job_title VARCHAR(50) NOT NULL
);
"""))

# 2. 插入数据
con.execute(text("""
INSERT INTO #employee
VALUES ('Smith', 'James', '3/1/2016', 'Staff Accountant'),
('Williams', 'Roberta', '2/7/2004', 'Sr. Software Engineer'),
('Washington','Mark','8/19/2014', 'HelpDesk Technician');
"""))

# 3. 查询数据并转为DataFrame
rs = con.execute(text("SELECT * FROM #employee;"))
df = pd.DataFrame(rs.fetchall())

额外注意 #

SQL Server的本地临时表（以#开头）仅存在于当前连接会话中，所以所有相关操作必须在同一个连接实例中完成，不要中途关闭连接再重新打开。

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