如何将API返回的JSON批量插入现有SQL表并实现存在则更新？

核心思路 #

先把API返回的JSON字符串解析为结构化数据集，再利用数据库原生的UPSERT（插入或更新）语法完成批量操作，全程自动化无需手动介入。

步骤1：解析JSON数据 #

以Python为例，用内置json模块解析API响应：

import json

# 模拟API返回的JSON字符串
api_response = '''
[
    {"id": 1, "name": "Alice", "created_at": "2024-01-01 10:00:00"},
    {"id": 2, "name": "Bob", "created_at": "2024-01-02 11:00:00"},
    {"id": 1, "name": "Alice Updated", "created_at": "2024-05-01 09:00:00"}
]
'''

# 解析为Python列表
data = json.loads(api_response)

步骤2：分数据库实现批量UPSERT #

MySQL/MariaDB（用INSERT ... ON DUPLICATE KEY UPDATE） #

假设目标表为users，id是主键或唯一键：

import mysql.connector

# 建立数据库连接
db = mysql.connector.connect(
    host="你的主机地址",
    user="用户名",
    password="密码",
    database="数据库名"
)
cursor = db.cursor()

# 动态生成SQL语句
columns = data[0].keys()
placeholders = ', '.join(['%s'] * len(columns))
# 排除id，其余字段用新值更新
update_clause = ', '.join([f"{col} = VALUES({col})" for col in columns if col != 'id'])

sql = f"""
INSERT INTO users ({', '.join(columns)})
VALUES ({placeholders})
ON DUPLICATE KEY UPDATE {update_clause}
"""

# 提取数据为元组列表
values = [tuple(item.values()) for item in data]

# 批量执行并提交
cursor.executemany(sql, values)
db.commit()

# 关闭连接
cursor.close()
db.close()

PostgreSQL（用INSERT ... ON CONFLICT DO UPDATE） #

import psycopg2

db = psycopg2.connect(
    host="你的主机地址",
    user="用户名",
    password="密码",
    database="数据库名"
)
cursor = db.cursor()

columns = data[0].keys()
placeholders = ', '.join(['%s'] * len(columns))
update_clause = ', '.join([f"{col} = EXCLUDED.{col}" for col in columns if col != 'id'])

sql = f"""
INSERT INTO users ({', '.join(columns)})
VALUES ({placeholders})
ON CONFLICT (id) DO UPDATE SET {update_clause}
"""

values = [tuple(item.values()) for item in data]
cursor.executemany(sql, values)
db.commit()

cursor.close()
db.close()

SQL Server（用MERGE语句） #

import pyodbc

db = pyodbc.connect(
    "DRIVER={ODBC Driver 17 for SQL Server};"
    "SERVER=你的主机地址;"
    "DATABASE=数据库名;"
    "UID=用户名;"
    "PWD=密码"
)
cursor = db.cursor()

# 用临时表存储批量数据，再执行MERGE
sql = """
DECLARE @TempUsers TABLE (id INT, name NVARCHAR(50), created_at DATETIME)
INSERT INTO @TempUsers VALUES (?, ?, ?)

MERGE INTO users AS Target
USING @TempUsers AS Source
ON Target.id = Source.id
WHEN MATCHED THEN
    UPDATE SET name = Source.name, created_at = Source.created_at
WHEN NOT MATCHED THEN
    INSERT (id, name, created_at) VALUES (Source.id, Source.name, Source.created_at);
"""

values = [tuple(item.values()) for item in data]
cursor.executemany(sql, values)
db.commit()

cursor.close()
db.close()

关键注意事项 #

• 必须确保表中id字段是主键或唯一约束，否则UPSERT逻辑无法触发更新。
• 批量操作比逐条执行效率提升显著，适合大数据量场景。
• 注意字段类型匹配：比如created_at若API返回字符串，部分数据库可自动转换，若不行需手动转为日期类型。
• 生产环境建议添加try-except捕获错误，避免单条数据异常导致整个批量操作失败。

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