Python向SQL插入API数据时忽略主键重复项的实现方案

现有Python代码用于将API数据插入Azure SQL表，表中已存储09/10/23的历史数据，主键为id列。当调整API参数拉取06/10/23-09/10/23的多日数据时，因09/10/23的id已存在，触发主键冲突导致整批数据无法插入。需要实现逐行插入逻辑：遇到主键冲突时忽略该行，继续插入其他不存在的条目。

### 导入依赖库 ###
import json
import requests
import pandas as pd
from sqlalchemy import create_engine, exc
from datetime import datetime, timedelta
import pyodbc

### API 请求逻辑 ###
# 计算前一日日期（可根据需求修改为自定义日期范围）
previous_day = datetime.now() - timedelta(days=1)
date_from = previous_day.strftime('%Y-%m-%dT00:00:00')
date_to = previous_day.strftime('%Y-%m-%dT23:59:59')

# API 配置
url = "XXXX"
payload = {}
headers = {'Authorization': 'XXXX'}

# 发起请求
response = requests.request("GET", url, headers=headers, data=payload)

if response.status_code == 200:
    # 解析JSON为DataFrame
    data = response.json()
    df = pd.json_normalize(data['data'])
    
    # 提取charging_periods中的tariff_id
    def extract_tariff_id(charging_periods):
        if isinstance(charging_periods, list) and len(charging_periods) > 0:
            first_entry = charging_periods[0]
            tariff_id = first_entry.get('tariff_id', 'Unknown Tariff ID')
            return tariff_id
        else:
            return 'Unknown Tariff ID'

    df['tariff_id'] = df['charging_periods'].apply(extract_tariff_id)
    df.drop(columns=['charging_periods'], inplace=True)

    ### Azure SQL 插入逻辑 ###
    # 数据库连接参数
    server = 'XXXX'
    database = 'XXXX'
    username = 'XXXX'
    password = 'XXXX'
    driver = 'XXXX'

    # 创建SQLAlchemy引擎
    connection_str = f"DRIVER={{{driver}}};SERVER={server};DATABASE={database};UID={username};PWD={password}"
    engine = create_engine(f"mssql+pyodbc:///?odbc_connect={connection_str}")

    # 逐行插入，处理主键冲突
    for idx, row in df.iterrows():
        try:
            # 将单行转为DataFrame后插入
            row_df = pd.DataFrame([row])
            row_df.to_sql('TABLE_NAME', engine, if_exists='append', index=False, schema='ev')
            print(f"成功插入第{idx+1}行数据")
        except exc.IntegrityError:
            # 捕获主键冲突错误，跳过该行
            print(f"第{idx+1}行数据主键已存在，跳过插入")
    
    print("\n数据插入完成！")
    print(df)
    print(df.info())

else:
    print(f"API请求失败，状态码：{response.status_code}")

• 移除冗余的pyodbc直接连接逻辑，统一使用SQLAlchemy引擎处理连接，避免连接资源泄漏
• 将原整批df.to_sql插入替换为逐行遍历插入，通过iterrows()循环处理每一行数据
• 捕获sqlalchemy.exc.IntegrityError异常（主键冲突属于此类），遇到冲突时跳过当前行，继续执行后续插入
• 修正原代码中变量名错误（Engine改为engine），并将逐行插入逻辑移至API请求成功的分支内，避免未获取数据时执行插入报错
• 添加简单日志输出，便于跟踪每一行的插入状态

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