这是我的用例:
我有多个并行运行的celery任务。
每个任务可以批量创建或更新许多对象。为此,我使用django-bulk。
因此,我基本上使用一个非常方便的函数insert_or_update_many:
1.它首先执行选择操作。 2.如果找到对象,则更新它们。 3.否则,它会创建它们。
但这引入了并发问题。例如:如果对象在步骤1期间不存在,则将其添加到稍后要插入的对象列表中。但在此期间可能会发生另一个Celery任务已经创建了该对象,并且当尝试执行批量插入(步骤3)时,我会收到重复输入错误的错误消息。
我想我需要将这三个步骤包装在一个“阻塞”块中。 我已经阅读了有关事务的文章,并尝试在
但这对我不起作用。我不确定自己是否完全理解了事务。基本上,我需要一个块,在其中可以放置多个操作,并确保没有其他进程或线程正在(写)访问我的数据库资源。
我有多个并行运行的celery任务。
每个任务可以批量创建或更新许多对象。为此,我使用django-bulk。
因此,我基本上使用一个非常方便的函数insert_or_update_many:
1.它首先执行选择操作。 2.如果找到对象,则更新它们。 3.否则,它会创建它们。
但这引入了并发问题。例如:如果对象在步骤1期间不存在,则将其添加到稍后要插入的对象列表中。但在此期间可能会发生另一个Celery任务已经创建了该对象,并且当尝试执行批量插入(步骤3)时,我会收到重复输入错误的错误消息。
我想我需要将这三个步骤包装在一个“阻塞”块中。 我已经阅读了有关事务的文章,并尝试在
with transaction.commit_on_success:
块内包装步骤1、2、3。with transaction.commit_on_success():
cursor.execute(sql, parameters)
existing = set(cursor.fetchall())
if not skip_update:
# Find the objects that need to be updated
update_objects = [o for (o, k) in object_keys if k in existing]
_update_many(model, update_objects, keys=keys, using=using)
# Find the objects that need to be inserted.
insert_objects = [o for (o, k) in object_keys if k not in existing]
# Filter out any duplicates in the insertion
filtered_objects = _filter_objects(con, insert_objects, key_fields)
_insert_many(model, filtered_objects, using=using)
但这对我不起作用。我不确定自己是否完全理解了事务。基本上,我需要一个块,在其中可以放置多个操作,并确保没有其他进程或线程正在(写)访问我的数据库资源。