PostgreSQL批量导入百万级CSV并获取ID，兼容唯一约束与Django模型

针对你这个百万级姓名批量导入Django模型并返回对应ID的需求，用PostgreSQL的COPY+Upsert绝对是最高效的方案——毕竟ORM循环百万条数据慢到离谱，下面给你详细的实现步骤：

核心就是用PostgreSQL的COPY命令做最快的批量导入，再结合9.5+版本支持的Upsert（INSERT ... ON CONFLICT）避免重复插入，最后通过关联查询一次性拿到所有姓名对应的ID，不管是新创建的还是数据库已有的。

首先你的Django模型已经定义好了，先确认表名（避免Django自动生成的表名混乱）：

from django.db import models

class Persons(models.Model):
    person_name = models.CharField(max_length=10, unique=True)

    class Meta:
        db_table = 'persons'  # 明确指定表名，方便后续SQL操作

然后写Python代码执行批量导入和ID查询：

from django.db import connection

def batch_import_persons(csv_file_path):
    with connection.cursor() as cursor:
        # 1. 创建临时表，用来暂存CSV数据
        cursor.execute("""
            CREATE TEMP TABLE temp_persons (
                person_name VARCHAR(10) NOT NULL
            );
        """)
        
        # 2. 用COPY命令把CSV数据导入临时表
        # 如果你的CSV有表头，先打开文件后执行next(f)跳过表头
        with open(csv_file_path, 'r', encoding='utf-8') as f:
            # 跳过表头（如果CSV第一行是"Name"这类表头的话）
            # next(f)
            # 因为你的CSV每行是一个姓名，字段分隔符设为不冲突的字符即可，这里用逗号
            cursor.copy_from(f, 'temp_persons', columns=('person_name',), sep=',')
        
        # 3. Upsert+查询所有姓名对应的ID
        # 用WITH子句先执行插入（已存在的跳过），再关联原表拿到所有记录的ID
        cursor.execute("""
            WITH inserted AS (
                INSERT INTO persons (person_name)
                SELECT person_name FROM temp_persons
                ON CONFLICT (person_name) DO NOTHING
                RETURNING id, person_name
            )
            SELECT p.id, p.person_name
            FROM persons p
            INNER JOIN temp_persons tp ON p.person_name = tp.person_name;
        """)
        
        # 把结果转换成{姓名: ID}的字典，方便后续使用
        name_id_map = {name: person_id for person_id, name in cursor.fetchall()}
        
        # 临时表会在会话结束后自动删除，手动清理也可以
        cursor.execute("DROP TABLE temp_persons;")
    
    return name_id_map

这个方案的优势 #

• 速度最快：COPY是PostgreSQL专门为批量导入设计的命令，比ORM的bulk_create快N倍，百万级数据导入几分钟就能搞定
• 原子性：整个操作在一个数据库会话里完成，不会出现部分导入的情况
• 一次性拿全ID：不管是新插入的还是数据库已有的姓名，都能一次性获取对应的ID，不用再单独查询

如果你不想写原生SQL的COPY部分，可以用django-postgres-copy这个第三方工具来处理导入，核心逻辑和方案一差不多，只是COPY部分用工具封装：

首先安装工具：

pip install django-postgres-copy

然后修改模型，添加CopyManager：

from django.db import models
from postgres_copy import CopyManager

class Persons(models.Model):
    person_name = models.CharField(max_length=10, unique=True)
    
    objects = CopyManager()  # 添加上这个管理器，用来执行COPY操作

    class Meta:
        db_table = 'persons'

然后写导入代码：

from django.db import connection

def import_with_postgres_copy(csv_file_path):
    with connection.cursor() as cursor:
        # 1. 创建临时表
        cursor.execute("""
            CREATE TEMP TABLE temp_persons (
                person_name VARCHAR(10) NOT NULL
            );
        """)
        
        # 2. 用django-postgres-copy的copy_from导入临时表
        # 如果CSV有表头，设置header=True
        Persons.objects.copy_from(
            csv_file_path,
            'temp_persons',
            columns=('person_name',),
            # header=True
        )
        
        # 3. 同样的Upsert+查询逻辑
        cursor.execute("""
            WITH inserted AS (
                INSERT INTO persons (person_name)
                SELECT person_name FROM temp_persons
                ON CONFLICT (person_name) DO NOTHING
                RETURNING id, person_name
            )
            SELECT p.id, p.person_name
            FROM persons p
            INNER JOIN temp_persons tp ON p.person_name = tp.person_name;
        """)
        
        name_id_map = {name: person_id for person_id, name in cursor.fetchall()}
        
        cursor.execute("DROP TABLE temp_persons;")
    
    return name_id_map

• CSV路径问题：如果用cursor.copy_from，CSV文件路径是PostgreSQL服务器能访问的路径（比如服务器上的本地路径）；本地开发时服务器和客户端在同一台机器，直接用本地路径就行。生产环境可以把CSV上传到服务器，或者用psql的\copy命令（客户端路径），这时候可以用subprocess执行psql命令。
• 字符编码：确保CSV文件的编码和数据库一致（一般是UTF-8），避免乱码。
• 测试先行：先拿10条数据测试逻辑是否正确，再处理百万级数据，避免出错。

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