如何将Python字典转换的DataFrame插入SQL Server数据库表？

阿华AIGC实验室

2026-5-22

我来帮你搞定把DataFrame插入SQL Server的问题！先梳理下你的需求：从Excel提取对应数据构造3个字段的DataFrame，然后插入到字段匹配的SQL Server表中。先把你的代码补全并优化，分两种方案给你参考：

方案一：用Pandas + SQLAlchemy（最简便推荐）

这个方法利用Pandas内置的to_sql方法，配合SQLAlchemy的连接引擎，代码简洁且不容易出错：

import pandas as pd
from sqlalchemy import create_engine

# 1. 读取Excel数据（注意新版本Pandas用sheet_name而非sheetname）
fp = "你的Excel文件路径.xlsx"
data = pd.read_excel(fp, sheet_name="CRM View")

# 2. 提取对应字段的值（这里要根据你Excel的实际列名调整）
# 假设date对应Excel第4行（索引3）的"日期"列，event对应第13行（索引12）的"事件"列
date_val = data.loc[3, "日期"]  # 替换成你Excel里的实际列名
sita_val = "ABZPD"
event_val = data.iloc[12, "事件"]  # 替换成实际列名

# 3. 构造符合要求的DataFrame（单行数据要加列表包裹，避免列名混乱）
df = pd.DataFrame({
    'date': [date_val],
    'sita': [sita_val],
    'event': [event_val]
})

# 4. 配置SQL Server连接（支持SQL身份验证或Windows身份验证）
# SQL身份验证格式：mssql+pymssql://用户名:密码@服务器地址/数据库名
# Windows身份验证格式：mssql+pymssql://@服务器地址/数据库名?trusted_connection=yes
engine = create_engine("mssql+pymssql://your_username:your_password@your_server/your_db")

# 5. 插入数据到SQL表
try:
    df.to_sql(
        name="你的SQL表名",  # 替换成目标表名
        con=engine,
        if_exists="append",  # 可选：append=追加数据，replace=覆盖原表，fail=表存在则报错
        index=False  # 不要把DataFrame的索引作为字段插入
    )
    print("数据插入成功！")
except Exception as e:
    print(f"插入出错：{str(e)}")

方案二：用Pymssql手动插入（适合需要精细控制的场景）

如果需要逐行处理或自定义插入逻辑，可以用pymssql手动执行插入语句：

import pandas as pd
import pymssql

# 1. 读取Excel并构造DataFrame（步骤同上）
fp = "你的Excel文件路径.xlsx"
data = pd.read_excel(fp, sheet_name="CRM View")
date_val = data.loc[3, "日期"]
sita_val = "ABZPD"
event_val = data.iloc[12, "事件"]
df = pd.DataFrame({
    'date': [date_val],
    'sita': [sita_val],
    'event': [event_val]
})

# 2. 建立SQL Server连接
conn = pymssql.connect(
    server="your_server",
    user="your_username",
    password="your_password",
    database="your_db"
)
cursor = conn.cursor()

# 3. 构造插入SQL语句（字段名要和你的SQL表完全匹配）
insert_sql = """
INSERT INTO 你的SQL表名 (date, sita, event)
VALUES (%s, %s, %s)
"""

# 4. 逐行插入数据
try:
    for _, row in df.iterrows():
        cursor.execute(insert_sql, (row['date'], row['sita'], row['event']))
    conn.commit()
    print("所有数据插入成功！")
except Exception as e:
    conn.rollback()
    print(f"插入失败：{str(e)}")
finally:
    cursor.close()
    conn.close()