如何用Python为含8万行数据的表批量生成并插入不同UUID？

针对8万条数据的场景，逐行循环更新效率极低，推荐以下两种批量处理方案，均基于Python的uuid库生成唯一UUID，通过单次/少量数据库操作完成更新：

方案一：基于SQLAlchemy ORM批量更新 #

适用于使用SQLAlchemy作为ORM框架的项目，核心是通过SQL的CASE语句批量关联ID与UUID，仅需一次数据库请求完成更新：

import uuid
from sqlalchemy import update, case
from your_models import OrderLineItem  # 替换为你的模型类路径
from sqlalchemy.orm import sessionmaker

# 初始化数据库会话
Session = sessionmaker(bind=your_db_engine)  # 替换为你的数据库引擎
session = Session()

# 1. 获取所有未填充UUID的记录ID
target_ids = [item.id for item in session.query(OrderLineItem.id).filter(OrderLineItem.uuid.is_(None)).all()]

# 2. 生成对应数量的UUID，建立ID-UUID映射
uuid_mapping = {item_id: str(uuid.uuid4()) for item_id in target_ids}

# 3. 构建批量更新语句
update_stmt = update(OrderLineItem).values(
    uuid=case(
        [(OrderLineItem.id == idx, uuid_val) for idx, uuid_val in uuid_mapping.items()],
        else_=OrderLineItem.uuid
    )
).where(OrderLineItem.id.in_(target_ids))

# 执行更新并提交
session.execute(update_stmt)
session.commit()
session.close()

优势 #

• 仅需一次数据库交互，避免逐行操作的网络开销
• 内存占用可控：8万条ID的列表仅占用约几百KB内存

方案二：基于原生DBAPI批量更新（以PostgreSQL + psycopg2为例） #

适用于直接使用DBAPI驱动的场景，利用execute_values批量构造更新语句：

import uuid
import psycopg2
from psycopg2.extras import execute_values

# 连接数据库
conn = psycopg2.connect(
    dbname="your_db_name",
    user="your_user",
    password="your_password",
    host="your_host"
)
cur = conn.cursor()

# 1. 获取未填充UUID的记录ID
cur.execute("SELECT id FROM order_line_items WHERE uuid IS NULL;")
target_ids = [row[0] for row in cur.fetchall()]

# 2. 生成ID-UUID配对列表
uuid_pairs = [(item_id, str(uuid.uuid4())) for item_id in target_ids]

# 3. 批量更新：通过VALUES子句关联ID与UUID
execute_values(
    cur,
    """
    UPDATE order_line_items AS oli
    SET uuid = data.uuid_val
    FROM (VALUES %s) AS data(item_id, uuid_val)
    WHERE oli.id = data.item_id;
    """,
    uuid_pairs,
    template="(%s, %s)"
)

# 提交事务并关闭连接
conn.commit()
cur.close()
conn.close()

适配其他数据库 #

如果使用MySQL，可替换为mysql-connector-python的executemany方法，构造批量更新语句：

# MySQL示例片段
import mysql.connector
from mysql.connector import Error

conn = mysql.connector.connect(host='your_host', database='your_db', user='your_user', password='your_pass')
cur = conn.cursor()

# 批量更新语句
update_query = "UPDATE order_line_items SET uuid = %s WHERE id = %s;"
cur.executemany(update_query, uuid_pairs)
conn.commit()

注意事项 #

• 操作前建议备份数据，或在测试环境验证逻辑
• 使用uuid.uuid4()生成随机UUID，确保全局唯一性
• 若数据库本身支持UUID生成（如PostgreSQL的uuid_generate_v4()），也可直接用SQL完成更新，但此方案严格遵循需求使用Python生成UUID

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