如何在Python中自动将API返回的JSON插入SQL Server并建表？

当然有不少Python工具能帮你搞定这种JSON到SQL Server的自动化建表+插入操作，再也不用手动写重复的建表语句和插入脚本啦！下面给你推荐几个实用的方案，适配不同的需求场景：

1. Pandas + SQLAlchemy（最省心的快速实现方案） #

Pandas可以一键把JSON转成DataFrame，再结合SQLAlchemy的to_sql方法，能自动根据DataFrame的列名和数据类型创建SQL Server表，还支持批量插入，效率比你现在的逐行插入高太多。

示例代码 #

import requests
import json
import pandas as pd
from sqlalchemy import create_engine
import constant

# 1. 拉取API数据
response = requests.get(url, headers=headers, auth=auth)
parse_response = json.loads(response.text)

# 2. 把JSON数组转成DataFrame
df = pd.DataFrame(parse_response)

# 3. 用SQLAlchemy创建数据库连接
# 连接格式：mssql+pyodbc:///?odbc_connect=你的连接字符串
conn_str = f"mssql+pyodbc:///?odbc_connect={constant.DW_CONNECTION}"
engine = create_engine(conn_str)

# 4. 自动建表+插入数据
# if_exists参数可选：'fail'（表存在就报错，默认）、'replace'（覆盖原表）、'append'（追加数据）
df.to_sql(
    name='LeadSource',  # 目标表名
    con=engine,
    if_exists='append',
    index=False,  # 不要把DataFrame的索引当成字段插入
    chunksize=1000  # 批量插入的批次大小，大数据量下能提升效率
)

# 关闭连接
engine.dispose()

优势 #

• 几乎零额外代码就能完成建表插入，Pandas会自动推断SQL数据类型（比如Python的str映射成SQL的NVARCHAR，bool映射成BIT）
• 批量插入的效率远高于逐行execute
• 支持灵活的表存在策略，适配不同的数据更新需求

注意事项 #

• 如果JSON有嵌套结构，Pandas会把嵌套部分转成字符串，你可以用pd.json_normalize()先扁平化数据
• 自动推断的类型可能不符合预期（比如长文本被设为NVARCHAR(255)），这时候可以手动指定dtype参数覆盖：

  from sqlalchemy.types import NVARCHAR, INTEGER, BIT
  
  df.to_sql(
      name='LeadSource',
      con=engine,
      if_exists='append',
      index=False,
      dtype={
          'RequestGuid': NVARCHAR(36),
          'Key': NVARCHAR(100),
          'Description': NVARCHAR(MAX),
          'LocationNumber': INTEGER,
          'Inactive': BIT,
          'ShowOnline': BIT
      }
  )
  

2. 手动生成建表语句（适合完全自定义表结构的场景） #

如果Pandas的自动推断满足不了你的精细需求（比如要加主键、索引、特定字段长度），可以自己写逻辑遍历JSON字段，推断对应SQL类型，生成CREATE TABLE语句，再用executemany批量插入。

示例代码片段 #

import requests
import json
import pyodbc
import constant

def get_sql_type(value):
    """根据Python值推断SQL Server数据类型"""
    if isinstance(value, str):
        return "NVARCHAR(MAX)" if len(value) > 255 else "NVARCHAR(255)"
    elif isinstance(value, int):
        return "INT"
    elif isinstance(value, bool):
        return "BIT"
    elif isinstance(value, float):
        return "FLOAT"
    elif value is None:
        return "NVARCHAR(255)"  # 默认类型，可按需调整
    else:
        return "NVARCHAR(MAX)"

# 拉取数据
response = requests.get(url, headers=headers, auth=auth)
parse_response = json.loads(response.text)

if not parse_response:
    print("没有数据可处理")
    exit()

# 用第一条记录的字段作为表列
sample_record = parse_response[0]
columns = list(sample_record.keys())
sql_columns = [f"[{col}] {get_sql_type(sample_record[col])}" for col in columns]

# 生成建表语句（仅当表不存在时创建）
create_table_sql = f"""
IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'LeadSource')
CREATE TABLE LeadSource (
    {', '.join(sql_columns)}
)
"""

# 连接数据库执行操作
conn = pyodbc.connect(constant.DW_CONNECTION)
cursor = conn.cursor()

# 执行建表语句
cursor.execute(create_table_sql)
conn.commit()

# 生成参数化插入语句，批量执行
placeholders = ', '.join(['?' for _ in columns])
insert_sql = f"INSERT INTO LeadSource ({', '.join([f'[{col}]' for col in columns])}) VALUES ({placeholders})"

# 把所有记录转成元组列表
records = [tuple(record[col] for col in columns) for record in parse_response]
cursor.executemany(insert_sql, records)
conn.commit()

conn.close()

优势 #

• 完全掌控表结构，能添加主键、索引、约束等自定义设置
• 适合处理复杂JSON结构或特殊数据类型的场景

注意事项 #

• 要处理JSON中字段缺失的情况（比如部分记录没有某个字段），可以添加默认值或者跳过处理
• 数据类型推断逻辑需要根据你的实际数据调整，避免出现类型不匹配的问题

3. 专门的JSON转SQL工具（小众但针对性强） #

还有一些第三方库专门做JSON到SQL的转换，比如json2sql（不过它对SQL Server的支持需要额外调整），或者sqlalchemy-utils里的辅助工具，但总体来说前面两种方案已经能覆盖绝大多数日常需求了。

总结一下：追求快速实现选Pandas + SQLAlchemy，需要精细控制表结构就自己写建表逻辑结合pyodbc的批量插入。

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