如何用SQLAlchemy+pyodbc执行含GO语句的T-SQL文件？

问题背景 #

使用SQLAlchemy + pyodbc执行T-SQL文件时，遇到文件中包含大量GO批处理分隔符的情况，例如：

-- query.sql
USE db_name
GO

SET ANSI_NULLS ON
GO

SET QUOTED_IDENTIFIER ON
GO

CREATE OR ALTER VIEW ...

原执行代码直接读取整个文件作为单个SQL语句执行，会因为CREATE OR ALTER VIEW需要作为批处理第一条语句而报错：

from sqlalchemy import create_engine, text

with open("query.sql") as fp:
    query = text(fp.read())

engine = create_engine("mssql+pyodbc://...")
with engine.begin() as conn:
    conn.execute(query)

解决方案 #

GO是SQL Server的批处理分隔符，并非标准SQL语法，SQLAlchemy的text()无法识别。需要将SQL内容按GO拆分成独立批处理，逐批执行：

实现步骤 #

1. 读取SQL文件内容，按GO（忽略大小写、前后空白）拆分成多个批处理片段
2. 过滤掉空的或仅含空白的片段
3. 逐个执行每个有效批处理片段

修改后的代码 #

import re
from sqlalchemy import create_engine, text

def execute_sql_file(file_path, engine):
    with open(file_path, 'r') as fp:
        sql_content = fp.read()
    
    # 按GO拆分，仅匹配单独一行的GO（前后可含空白，忽略大小写）
    batches = re.split(r'^\s*GO\s*$', sql_content, flags=re.MULTILINE | re.IGNORECASE)
    
    with engine.begin() as conn:
        for batch in batches:
            cleaned_batch = batch.strip()
            if cleaned_batch:
                conn.execute(text(cleaned_batch))

# 初始化引擎并执行脚本
engine = create_engine("mssql+pyodbc://...")
execute_sql_file("query.sql", engine)

关键说明 #

• 正则表达式r'^\s*GO\s*$'确保只拆分单独成行的GO，避免误拆SQL语句中包含"GO"的字符串（比如字段名、注释）
• 每个批处理执行前清理空白并跳过空片段，避免执行无效语句
• 通过engine.begin()维持事务上下文，确保所有批处理要么全部成功，要么全部回滚

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