Demo the deadlock

django_bulk_load/bulk_load.py

@@ -301,6 +301,8 @@ def bulk_update_models(
     else:
         queries = [update_query]
 
+    helper = f"select id from {table_name} order by id asc for update"
+    queries = [helper] + queries
     return bulk_load_models_with_queries(
         models=models,
         loading_table_name=loading_table_name,

tests/test_bulk_update_models.py

@@ -1,21 +1,45 @@
 from datetime import datetime, timezone
 
-from django.test import TestCase
+from django.db import close_old_connections
+from django.test import TestCase, TransactionTestCase
 from django_bulk_load import bulk_update_models, generate_greater_than_condition
+import queue
 from .test_project.models import (
     TestComplexModel,
     TestForeignKeyModel,
 )
-
-
-class E2ETestBulkUpdateModels(TestCase):
+import random
+import threading
+import sys
+
+def do_update(bucket):
+    objects = [x for x in TestComplexModel.objects.all()]
+    for current in objects:
+        current.integer_field = random.randint(10,100000)
+    try:
+        bulk_update_models(objects)
+    except Exception:
+        bucket.put(sys.exc_info())
+    close_old_connections()
+
+class E2ETestBulkUpdateModels(TransactionTestCase):
     def test_integer_field_change(self):
-        model1 = TestComplexModel(integer_field=1)
-        model1.save()
-        model1.integer_field = 2
-        bulk_update_models([model1])
-        saved_model = TestComplexModel.objects.get()
-        self.assertEqual(saved_model.integer_field, 2)
+        models = [TestComplexModel(integer_field=1) for x in range(10000)]
+        error_bucket = queue.Queue()
+
+        for model in models:
+            model.save()
+        threads = [
+                threading.Thread(target=do_update, args=[error_bucket])
+                for x in range(40)
+                ]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        while not error_bucket.empty():
+            raise error_bucket.get()[1]
+
 
     def test_string_field_change(self):
         model1 = TestComplexModel(string_field="hello")