如何将含嵌套字典与列表的Python数据导入SQL并拆分关联表

针对嵌套结构数据无法通过JSON直接导入MySQL的问题，我们可以通过Python手动拆解数据，分三步完成三张关联表的批量插入，全程使用自增主键建立关联，避免依赖外部person_id。

先确认目标表结构 #

以MySQL为例，确保三张表的结构符合需求：

-- People表：自增主键，扁平存储extras中的字段
CREATE TABLE People (
    person_inner_id INT AUTO_INCREMENT PRIMARY KEY,
    person_id VARCHAR(50) NOT NULL,
    name VARCHAR(50) NOT NULL,
    likes_swimming BOOLEAN NOT NULL,
    likes_cooking BOOLEAN NOT NULL
);

-- Skills表：存储唯一技能，自增主键
CREATE TABLE Skills (
    skill_id INT AUTO_INCREMENT PRIMARY KEY,
    skill VARCHAR(50) UNIQUE NOT NULL
);

-- Skills_People关联表：实现People与Skills的多对多关联
CREATE TABLE Skills_People (
    person_inner_id INT NOT NULL,
    skill_id INT NOT NULL,
    PRIMARY KEY (person_inner_id, skill_id),
    FOREIGN KEY (person_inner_id) REFERENCES People(person_inner_id),
    FOREIGN KEY (skill_id) REFERENCES Skills(skill_id)
);

Python批量插入实现步骤 #

我们使用mysql-connector-python库完成数据库操作，先安装依赖：

pip install mysql-connector-python

1. 数据库连接与核心插入逻辑 #

import mysql.connector
from mysql.connector import Error

# 替换为你的数据库配置
DB_CONFIG = {
    'host': 'localhost',
    'database': 'your_database_name',
    'user': 'your_username',
    'password': 'your_password'
}

def get_db_connection():
    """创建并返回数据库连接"""
    connection = None
    try:
        connection = mysql.connector.connect(**DB_CONFIG)
    except Error as e:
        print(f"连接失败: {e}")
    return connection

def bulk_insert_data(raw_data):
    connection = get_db_connection()
    if not connection:
        return

    cursor = connection.cursor()
    try:
        # --------------------------
        # 第一步：插入People表，扁平化extras
        # --------------------------
        people_records = []
        person_skill_mapping = {}  # 临时存储外部person_id对应的技能列表

        for person in raw_data:
            # 拆解嵌套的extras字典为扁平字段
            people_records.append((
                person['person_id'],
                person['name'],
                person['extras']['likes_swimming'],
                person['extras']['likes_cooking']
            ))
            # 记录当前用户的技能，后续关联用
            person_skill_mapping[person['person_id']] = person['skills']

        # 批量插入People表
        insert_people_sql = """
            INSERT INTO People (person_id, name, likes_swimming, likes_cooking)
            VALUES (%s, %s, %s, %s)
        """
        cursor.executemany(insert_people_sql, people_records)
        connection.commit()

        # 获取外部person_id与自增person_inner_id的映射
        cursor.execute(
            "SELECT person_inner_id, person_id FROM People WHERE person_id IN (%s)"
            % ','.join(['%s'] * len(person_skill_mapping.keys())),
            tuple(person_skill_mapping.keys())
        )
        inner_id_map = {row[1]: row[0] for row in cursor.fetchall()}

        # --------------------------
        # 第二步：去重插入Skills表
        # --------------------------
        # 收集所有技能并去重
        all_skills = set()
        for skills in person_skill_mapping.values():
            all_skills.update(skills)

        # 使用INSERT IGNORE避免重复插入已存在的技能
        insert_skill_sql = "INSERT IGNORE INTO Skills (skill) VALUES (%s)"
        cursor.executemany(insert_skill_sql, [(skill,) for skill in all_skills])
        connection.commit()

        # 获取技能与skill_id的映射
        cursor.execute(
            "SELECT skill_id, skill FROM Skills WHERE skill IN (%s)"
            % ','.join(['%s'] * len(all_skills)),
            tuple(all_skills)
        )
        skill_id_map = {row[1]: row[0] for row in cursor.fetchall()}

        # --------------------------
        # 第三步：插入关联表Skills_People
        # --------------------------
        association_records = []
        for ext_person_id, skills in person_skill_mapping.items():
            inner_person_id = inner_id_map[ext_person_id]
            for skill in skills:
                association_records.append((inner_person_id, skill_id_map[skill]))

        insert_assoc_sql = """
            INSERT INTO Skills_People (person_inner_id, skill_id)
            VALUES (%s, %s)
        """
        cursor.executemany(insert_assoc_sql, association_records)
        connection.commit()

        print("所有数据插入完成！")

    except Error as e:
        print(f"插入出错: {e}")
        connection.rollback()  # 出错回滚，保证数据一致性
    finally:
        # 关闭连接
        if connection.is_connected():
            cursor.close()
            connection.close()

# 你的原始数据（注意修正person_id的字符串引号，原代码中未加引号会报错）
raw_lst = [ 
        {
            "person_id": "HZT998", "name": 'john', "skills": ['python', 'sql', 'r'], 
            "extras": {"likes_swimming": False, "likes_cooking": True}},
        {
            "person_id": "HTY954", "name": 'peter', "skills": ['python', 'r', 'c#'], 
            "extras": {"likes_swimming": True, "likes_cooking": False}}
      ]

# 执行批量插入
bulk_insert_data(raw_lst)

注意事项 #

• 原数据中的person_id值（如HZT998）需要加字符串引号，否则Python会将其视为未定义变量导致报错。
• 使用INSERT IGNORE处理Skills表的去重，确保相同技能只存储一次。
• 所有操作包裹在事务中，出错时自动回滚，避免数据不一致。
• 通过查询获取主键映射，确保关联关系准确无误。

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