### Django 中 `bulk_update_or_create` 的使用方法 尽管当前官方版本的 Django 并未提供内置的 `bulk_update_or_create` 方法，但可以通过第三方库或自定义实现来完成此功能。以下是基于引用中的相关内容以及扩展的知识点所提供的解决方案。 #### 第三方库支持 可以借助 `django-bulk-update` 库[^1] 来简化批量更新和创建的操作。该库提供了高效的批量处理能力，适用于大规模数据集场景下的性能优化需求。 安装方式如下： ```bash pip install django-bulk-update ``` 假设有一个模型类 `Text_one` 定义如下： ```python from django.db import models class Text_one(models.Model): name = models.CharField(max_length=100, unique=True) title = models.CharField(max_length=100) ``` 下面是一个完整的示例代码展示如何使用 `bulk_update_or_create` 功能： ```python from django_bulk_update.helper import bulk_update_or_create from .models import Text_one def perform_bulk_update_or_create(): # 准备要更新或创建的数据列表 data_to_process = [ {"name": "天衣无缝5", "title": "新标题"}, {"name": "新增记录1", "title": "初始标题"}, {"name": "已存在名称2", "title": "修改后的标题"} ] # 将字典转换为对象实例 instances = [] for item in data_to_process: obj, created = Text_one.objects.get_or_create( name=item["name"], defaults={"title": item["title"]} ) if not created and "title" in item: setattr(obj, "title", item["title"]) instances.append(obj) # 执行批量更新或创建操作 updated_count, created_count = bulk_update_or_create(Text_one, instances, ["title"]) print(f"Updated {updated_count} records.") print(f"Created {created_count} new records.") perform_bulk_update_or_create() ``` 以上代码实现了以下逻辑： - 遍历输入数据并尝试获取现有记录。 - 如果记录不存在，则调用 `get_or_create` 创建新的实例；如果存在则仅设置需更新的字段。 - 调用 `bulk_update_or_create` 进行高效的一次性写入操作。 注意：这里使用的 `bulk_update_or_create` 是由外部工具提供的增强版函数，能够同时返回被更新和新建的数量统计信息[^1]。 --- #### 自定义实现方案 如果没有引入额外依赖项的需求，也可以手动编写类似的批量化逻辑。核心思路在于利用 Django 提供的基础 API (`filter`, `update`) 和事务管理机制(`transaction.atomic`)组合而成。 下面是不依赖任何插件的手动实现版本： ```python from django.db import transaction from .models import Text_one @transaction.atomic def custom_bulk_update_or_create(data_list, model_class, key_field="name"): existing_records = { record[key_field]: record for record in model_class.objects.filter(**{ f"{key_field}__in": [item[key_field] for item in data_list] }).values(key_field, *model_class._meta.fields_map.keys()) } to_create = [] to_update = [] for entry in data_list: primary_key_value = entry[key_field] if primary_key_value in existing_records: instance = existing_records[primary_key_value] for field_name, value in entry.items(): if hasattr(instance, field_name) and getattr(instance, field_name) != value: setattr(instance, field_name, value) to_update.append(instance) else: to_create.append(model_class(**entry)) if to_create: model_class.objects.bulk_create(to_create) if to_update: model_class.objects.bulk_update(to_update, fields=list(entry.keys())) custom_bulk_update_or_create([ {"name": "天衣无缝5", "title": "新标题"}, {"name": "新增记录1", "title": "初始标题"}, ], Text_one, key_field="name") ``` 上述脚本执行流程包括以下几个部分： 1. 查询数据库中已经存在的条目集合； 2. 对于每一条传入的数据判断其是否存在； 3. 存在即加入待更新队列，反之进入待插入数组； 4. 最终分别调用 `bulk_create` 及 `bulk_update` 实现原子化提交更改至持久层[^2]. --- #### 注意事项 - **唯一约束**：为了确保正确区分哪些是需要更新的对象而哪些应该作为全新实体对待，请务必指定具有唯一性的键属性（如主键或者设置了 UNIQUE 约束的其他字段）用于匹配查找条件。 - **性能考量**：当涉及大量数据时建议启用分页加载策略减少内存占用压力，并合理配置批次大小参数提升吞吐效率。 - **异常捕获**：实际部署环境中应当增加必要的错误检测环节以便及时发现潜在隐患问题所在位置。 ---