如何通过Python自动读取竖线分隔文本文件并导入SQL Server表

现有场景 #

• 待导入的竖线分隔文件示例格式：

name1|9|111|replace|12|sds22|dsd|GDS-sw-E|1|1|0|No|21|43|No|0.1
name1|9|222|replace|33|dfs|SWR|RRR-ddd@-S1|1|1|0|No|33|4|None|0.9

• 原实现的问题：当前代码需要手动逐行把文件内容硬编码到Values变量中，无法自动读取文件内容，批量导入时效率极低且容易出错。

自动化改造后的完整代码 #

import pyodbc

# 可根据实际环境调整的配置
FILE_PATH = "./input_pipe_file.txt"
CONNECTION_STRING = (
    'Driver={SQL Server};'
    'Server=servername\\dbname,1433;'
    'Database=dbname;'
    'UID=uname;'
    'PWD=pass;'
    'Trusted_Connection=no;'
)
INSERT_SQL = "INSERT INTO dbname.schema.table VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"

def import_pipe_file_to_sql():
    conn = pyodbc.connect(CONNECTION_STRING)
    cursor = conn.cursor()
    # 开启快速批量执行模式，提升插入性能
    cursor.fast_executemany = True
    success_count = 0
    try:
        with open(FILE_PATH, 'r', encoding='utf-8') as file_handle:
            for raw_line in file_handle:
                # 去除行首尾的换行符、空白字符
                clean_line = raw_line.strip()
                # 跳过空行
                if not clean_line:
                    continue
                # 按竖线拆分字段
                row_values = clean_line.split('|')
                # 校验字段数量和SQL占位符是否匹配，避免参数数量报错
                if len(row_values) != 16:
                    print(f"跳过格式异常行：{clean_line}，拆分得到{len(row_values)}个字段，要求16个")
                    continue
                cursor.execute(INSERT_SQL, row_values)
                success_count += 1
        # 全部行处理完成后统一提交事务
        conn.commit()
        print(f"导入完成，成功插入{success_count}条数据")
    except Exception as err:
        # 出现异常时回滚所有未提交的变更，保证数据一致性
        conn.rollback()
        print(f"导入中断，已回滚所有操作，错误信息：{str(err)}")
    finally:
        conn.close()

if __name__ == "__main__":
    import_pipe_file_to_sql()

核心逻辑说明 #

• 逐行读取目标文件，不会一次性把全文件加载到内存，哪怕文件体积达到GB级也不会出现内存溢出问题
• 自动跳过空行、校验每行的字段数量，避免因为脏数据导致整个导入流程报错中断
• 保留pyodbc参数化查询的特性，文件内容里的特殊字符不会触发SQL注入风险
• 增加事务回滚逻辑，导入过程中出现任意错误时，不会往库里插入半截的脏数据
• 大文件导入时可以通过开启fast_executemany开关，将插入速度提升数倍

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