Django自定义命令导入CSV至数据库报错：外键字段无法分配ID

通过自定义Django命令从CSV导入数据时，User、Category、Genre表可成功填充，但处理Title模型时抛出错误：

ValueError: Cannot assign '1': 'Title.category' must be an instance of 'Category'.

现有代码示例：

models = {
    User: 'users.csv',
    Category: 'category.csv',
    Genre: 'genre.csv',
    Title: 'titles.csv',
}


class Command(BaseCommand):
    help = 'Command to automatically populate the database'

    def handle(self, *args, **options):
        path = str(BASE_DIR / 'static/data/')
        for model, csv_file in models.items():
            with open(path + '/' + csv_file, 'r', encoding='utf-8') as file:
                rows = csv.DictReader(file)
                records = [model(**row) for row in rows]                   
                model.objects.bulk_create(records)
            self.stdout.write(self.style.SUCCESS(f'Done {model.__name__})'))

CSV示例文件：

1. users.csv

id,username,email,role,bio,first_name,last_name
100,bingobongo,bingobongo@mail.fake,user

2. category.csv

id,name,slug
1,Movie,movie

3. genre.csv

id,name,slug
1,Drama,drama

4. titles.csv

id,name,year,category
1,The Shawshank Redemption,1994,1

方法1：利用Django外键的_id字段特性 #

Django会为外键字段自动生成{字段名}_id格式的数据库字段，直接给该字段赋值主键ID即可，无需传递模型实例。修改命令的handle方法，针对Title模型单独处理字段映射：

def handle(self, *args, **options):
    path = str(BASE_DIR / 'static/data/')
    for model, csv_file in models.items():
        with open(path + '/' + csv_file, 'r', encoding='utf-8') as file:
            rows = csv.DictReader(file)
            records = []
            for row in rows:
                if model == Title:
                    # 将csv中的category字段转为category_id，直接赋值主键ID
                    row['category_id'] = row.pop('category')
                records.append(model(**row))                   
            model.objects.bulk_create(records)
        self.stdout.write(self.style.SUCCESS(f'Done {model.__name__})'))

方法2：预加载Category实例映射（适合需验证关联实例的场景） #

先将所有Category的ID与实例存入字典，处理Title时直接通过ID获取对应实例，避免重复查询数据库：

def handle(self, *args, **options):
    path = str(BASE_DIR / 'static/data/')
    # 预加载Category的ID-实例映射
    category_map = {}
    
    # 优先处理Category表，确保依赖数据已存在
    if Category in models:
        csv_file = models.pop(Category)
        with open(path + '/' + csv_file, 'r', encoding='utf-8') as file:
            rows = csv.DictReader(file)
            categories = [Category(**row) for row in rows]
            Category.objects.bulk_create(categories)
        # 构建ID到实例的映射字典
        category_map = {cat.id: cat for cat in Category.objects.all()}
        self.stdout.write(self.style.SUCCESS('Done Category'))
    
    # 处理剩余模型
    for model, csv_file in models.items():
        with open(path + '/' + csv_file, 'r', encoding='utf-8') as file:
            rows = csv.DictReader(file)
            records = []
            for row in rows:
                if model == Title:
                    # 通过ID从映射字典中获取Category实例
                    category_id = int(row['category'])
                    row['category'] = category_map[category_id]
                records.append(model(**row))                   
            model.objects.bulk_create(records)
        self.stdout.write(self.style.SUCCESS(f'Done {model.__name__})'))

两种方法对比 #

• 方法1更简洁高效，直接操作数据库字段，无需额外查询，也不依赖表的处理顺序
• 方法2适合需要对关联实例进行验证（如检查ID是否存在）或额外处理的场景，但需确保Category表优先处理

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