Added support for skip_deletes, skip_creates, and skip_updates in bulk_sync method.

Author: tempoxylophone

bulk_sync/__init__.py

@@ -1,12 +1,11 @@
 from collections import OrderedDict
 import logging
-
 from django.db import transaction
 
 logger = logging.getLogger(__name__)
 
 
-def bulk_sync(new_models, key_fields, filters, batch_size=None, fields=None):
+def bulk_sync(new_models, key_fields, filters, batch_size=None, fields=None, skip_creates=False, skip_updates=False, skip_deletes=False):
     """ Combine bulk create, update, and delete.  Make the DB match a set of in-memory objects.
 
     `new_models`: Django ORM objects that are the desired state.  They may or may not have `id` set.
@@ -50,21 +49,25 @@ def get_key(obj):
                 new_obj.id = old_obj.id
                 existing_objs.append(new_obj)
 
+    if skip_creates is False:
         db_class.objects.bulk_create(new_objs, batch_size=batch_size)
+        
+    if skip_updates is False:
         db_class.objects.bulk_update(existing_objs, fields=fields, batch_size=batch_size)
 
-        # delete stale ones...
+    if skip_deletes is False:
+        # delete stale objects
         objs.filter(pk__in=[_.pk for _ in list(obj_dict.values())]).delete()
 
-        assert len(existing_objs) == len(new_models) - len(new_objs)
+    assert len(existing_objs) == len(new_models) - len(new_objs)
 
-        stats = {"created": len(new_objs), "updated": len(new_models) - len(new_objs), "deleted": len(obj_dict)}
+    stats = {"created": len(new_objs), "updated": len(new_models) - len(new_objs), "deleted": len(obj_dict)}
 
-        logger.debug(
-            "{}: {} created, {} updated, {} deleted.".format(
-                db_class.__name__, stats["created"], stats["updated"], stats["deleted"]
-            )
+    logger.debug(
+        "{}: {} created, {} updated, {} deleted.".format(
+            db_class.__name__, stats["created"], stats["updated"], stats["deleted"]
         )
+    )
 
     return {"stats": stats}
 

tests/tests.py

@@ -104,6 +104,58 @@ def test_fields_parameter(self):
         self.assertEqual(0, ret["stats"]["created"])
         self.assertEqual(0, ret["stats"]["deleted"])
 
+    def test_skip_deletes(self):
+        c1 = Company.objects.create(name="My Company LLC")
+
+        e1 = Employee.objects.create(name="Scott", age=40, company=c1)
+        e2 = Employee.objects.create(name="Isaac", age=9, company=c1)
+
+        # update Scott - this makes Isaac is the "stale object" that would be deleted if skip_deletes were False
+        new_objs = [
+            Employee(name="Scott", age=41, company=c1),
+        ]
+
+        # but Isaac should remain when the skip_deletes flag is True
+        ret = bulk_sync(new_models=new_objs, filters=None, key_fields=("name",), skip_deletes=True)
+
+        self.assertEqual(2, Employee.objects.count())
+        self.assertEqual(["Scott", "Isaac"], [x.name for x in Employee.objects.all().order_by('id')])
+
+    def test_skip_creates(self):
+        c1 = Company.objects.create(name="My Company LLC")
+
+        e1 = Employee.objects.create(name="Scott", age=40, company=c1)
+        e2 = Employee.objects.create(name="Isaac", age=9, company=c1)
+
+        # create a new employee that will be ignored
+        new_objs = [
+            Employee(name="John", age=52, company=c1)
+        ]
+
+        ret = bulk_sync(new_models=new_objs, filters=None, key_fields=("name",), skip_creates=True, skip_deletes=True)
+
+        self.assertEqual(2, Employee.objects.count())
+        self.assertEqual(["Scott", "Isaac"], [x.name for x in Employee.objects.all().order_by('id')])
+
+    def test_skip_updates(self):
+        c1 = Company.objects.create(name="My Company LLC")
+
+        e1 = Employee.objects.create(name="Scott", age=40, company=c1)
+        e2 = Employee.objects.create(name="Isaac", age=9, company=c1)
+
+        # create a new employee that will be ignored
+        new_objs = [
+            Employee(name="Scott", age=100, company=c1)
+        ]
+
+        ret = bulk_sync(new_models=new_objs, filters=None, key_fields=("name",), skip_updates=True)
+
+        # the age should not have been updated
+        new_e1 = Employee.objects.get(id=e1.id)
+        self.assertEqual(40, new_e1.age)
+
+
+
 class BulkCompareTests(TestCase):
     """ Test `bulk_compare` method """