如何使用Sqlalchemy运行含SELECT INTO或CREATE TABLE AS的查询？

1. 先修正T-SQL语法与执行问题 #

（1）CREATE TABLE AS的T-SQL语法错误 #

T-SQL不支持CREATE TABLE ... FROM这种写法，它没有类似PostgreSQL的CREATE TABLE AS SELECT语法，等价实现要么用SELECT INTO，要么先建表再插入数据。

（2）SELECT INTO无报错但不生成表的核心原因 #

你执行代码时缺少事务提交步骤。pyodbc默认连接为非自动提交模式，执行SELECT INTO后如果不手动提交，事务会回滚，表自然不会被创建。

修复后的原生查询执行代码：

query = 'select top 10 * into dbo.test_table from dbo.main_table'

engine = create_engine(....)
conn = engine.raw_connection()
cursor = conn.cursor()
try:
    cursor.execute(query)
    conn.commit()  # 必须手动提交事务
    # 可选：检查执行反馈
    print(cursor.messages)
finally:
    cursor.close()
    conn.close()

2. 替代的T-SQL写法（先建表再插入） #

如果需要显式定义表结构，可采用「建表+插入」的组合：

-- 先创建与原表结构匹配的空表
CREATE TABLE dbo.test_table (
    id INT,
    username VARCHAR(50),
    create_time DATETIME,
    -- 此处需与dbo.main_table的列一一对应
);

-- 插入前10条数据
INSERT INTO dbo.test_table
SELECT TOP 10 * FROM dbo.main_table;

执行时同样需要调用conn.commit()提交事务。

3. 用SQLAlchemy（pyodbc）实现的两种方式 #

方式一：Core执行原生查询并提交 #

from sqlalchemy import create_engine

engine = create_engine("mssql+pyodbc://your_connection_string")

with engine.connect() as conn:
    conn.execute("select top 10 * into dbo.test_table from dbo.main_table")
    conn.commit()  # 提交事务

方式二：ORM风格复制表结构并插入数据 #

这种方式无需写原生SQL，靠SQLAlchemy反射原表结构：

from sqlalchemy import create_engine, MetaData, Table, select

engine = create_engine("mssql+pyodbc://your_connection_string")
metadata = MetaData()

# 反射原表结构
main_table = Table("main_table", metadata, autoload_with=engine, schema="dbo")

# 创建新表（复制原表列结构）
test_table = Table(
    "test_table",
    metadata,
    *[col.copy() for col in main_table.columns],
    schema="dbo"
)
metadata.create_all(engine)  # 执行建表

# 插入前10条数据
with engine.connect() as conn:
    stmt = select(main_table).limit(10)
    data = conn.execute(stmt).fetchall()
    conn.execute(test_table.insert(), data)
    conn.commit()

额外排查点 #

• 确认数据库账号拥有CREATE TABLE权限，权限不足会导致隐性失败，可通过cursor.messages查看执行细节。
• 若连接字符串中设置了autocommit=True，可省略手动提交步骤。

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