自从Django添加了bulk_update支持,现在这是有点可能的,尽管您需要为每个批次执行3个数据库调用(获取、批量创建和批量更新)。这里挑战在于如何为通用目的的函数创建一个良好的接口,因为您希望该函数既支持高效查询又支持更新。下面是我实现的一种方法,它专为具有许多公共标识键(可以为空)和一个在批次中变化的标识键的bulk update_or_create而设计。
这是在基本模型上实现的方法,但可以独立使用。这还假定基本模型在模型名为updated_on的auto_now
时间戳;如果不是这种情况,则已对假定此处进行代码行的注释以便于修改。
为了将其分批使用,请在调用之前将更新分块。这也是绕过数据的一种方式,其中辅助标识符可以具有少量值之一,而无需更改接口。
class BaseModel(models.Model):
updated_on = models.DateTimeField(auto_now=True)
@classmethod
def bulk_update_or_create(cls, common_keys, unique_key_name, unique_key_to_defaults):
"""
common_keys: {field_name: field_value}
unique_key_name: field_name
unique_key_to_defaults: {field_value: {field_name: field_value}}
ex. Event.bulk_update_or_create(
{"organization": organization}, "external_id", {1234: {"started": True}}
)
"""
with transaction.atomic():
filter_kwargs = dict(common_keys)
filter_kwargs[f"{unique_key_name}__in"] = unique_key_to_defaults.keys()
existing_objs = {
getattr(obj, unique_key_name): obj
for obj in cls.objects.filter(**filter_kwargs).select_for_update()
}
create_data = {
k: v for k, v in unique_key_to_defaults.items() if k not in existing_objs
}
for unique_key_value, obj in create_data.items():
obj[unique_key_name] = unique_key_value
obj.update(common_keys)
creates = [cls(**obj_data) for obj_data in create_data.values()]
if creates:
cls.objects.bulk_create(creates)
update_fields = {"updated_on"}
updates = []
for key, obj in existing_objs.items():
obj.update(unique_key_to_defaults[key], save=False)
update_fields.update(unique_key_to_defaults[key].keys())
updates.append(obj)
if existing_objs:
cls.objects.bulk_update(updates, update_fields)
return len(creates), len(updates)
def update(self, update_dict=None, save=True, **kwargs):
""" Helper method to update objects """
if not update_dict:
update_dict = kwargs
update_fields = {"updated_on"}
for k, v in update_dict.items():
setattr(self, k, v)
update_fields.add(k)
if save:
self.save(update_fields=update_fields)
使用示例:
class Event(BaseModel):
organization = models.ForeignKey(Organization)
external_id = models.IntegerField(unique=True)
started = models.BooleanField()
organization = Organization.objects.get(...)
updates_by_external_id = {
1234: {"started": True},
2345: {"started": True},
3456: {"started": False},
}
Event.bulk_update_or_create(
{"organization": organization}, "external_id", updates_by_external_id
)
可能出现的竞争条件
上述代码利用了事务和select-for-update来防止更新时的竞争条件。但是,如果两个线程或进程试图创建具有相同标识符的对象,则可能存在插入时的竞争条件。
简单的解决方法是确保您的common_keys和unique_key的组合是一个由数据库强制执行的唯一性约束(这是此函数的预期使用方式)。这可以通过unique_key引用具有unique=True
字段的字段,或者将unique_key与子集common_keys结合起来,并通过UniqueConstraint强制实施唯一性来实现。通过数据库强制实施唯一性保护,如果多个线程尝试执行冲突创建,则除了一个以外的所有线程都将失败,并显示IntegrityError
。由于封闭事务,失败的线程不会执行任何更改,可以安全地重试或忽略它们(一个失败的冲突创建可能只被视为第一个创建,然后立即被覆盖)。
如果无法利用唯一性约束,则需要实现自己的并发控件或锁定整个表。