Merge pull request #4 from awmath/skip-signals

feat: signals can be skipped by a keyword argument

README.md

@@ -23,11 +23,14 @@ To see them in action use the following snippet:
 from django.dispatch import receiver
 from bulk_signals import signals
 
-@receiver(signals.post_bulk_update, signals.post_bulk_update, signals.post_query_update)
+@receiver(signals.pre_bulk_update, signals.post_bulk_update, signals.post_query_update)
 def debug(*args, **kwargs):
     print(args)
     print(kwargs)
 ```
 
+You can skip the signals on a single execution by using the `skip_signal=True` keyword argument.
+Which keyword should be used for skipping is configurable via the `BULK_SIGNALS_SKIP_KEY="skip_signal"` configuration in the django settings.
+
 # TODO
 -  test against different database backends

bulk_signals/apps.py

@@ -1,8 +1,14 @@
 from django.apps import AppConfig, apps
+from django.conf import settings
 
 from bulk_signals import signals
 
 
+def skip_signal(kwargs):
+    skip_key = getattr(settings, "BULK_SIGNALS_SKIP_KEY", "skip_signal")
+    return kwargs.pop(skip_key, False) is True
+
+
 class BulkSignalsConfig(AppConfig):
     default_auto_field = "django.db.models.BigAutoField"
     name = "bulk_signals"
@@ -14,6 +20,8 @@ def ready(self):
         base_bulk_create = QuerySet.bulk_create
 
         def bulk_create(queryset, objects, **kwargs):
+            if skip_signal(kwargs):
+                return base_bulk_create(queryset, objects, **kwargs)
             # get model label from queryset
             model = apps.get_model(queryset.model._meta.label)
 
@@ -28,7 +36,14 @@ def bulk_create(queryset, objects, **kwargs):
         base_bulk_update = QuerySet.bulk_update
 
         def bulk_update(queryset, objects, fields, **kwargs):
+            # add a queryset hint so update signals won't be
+            # triggerd for bulk_update
             queryset._hints["is_bulk_update"] = True
+
+            # check if the signals should be skipped
+            if skip_signal(kwargs):
+                return base_bulk_update(queryset, objects, fields, **kwargs)
+
             model = apps.get_model(queryset.model._meta.label)
 
             signals.pre_bulk_update.send(
@@ -46,6 +61,10 @@ def bulk_update(queryset, objects, fields, **kwargs):
         base_update = QuerySet.update
 
         def update(queryset, **kwargs):
+            # check if the signals should be skipped
+            if skip_signal(kwargs):
+                return base_update(queryset, **kwargs)
+
             model = apps.get_model(queryset.model._meta.label)
 
             # if this update is part of a bulk_update action skip this part

bulk_signals/tests/conftest.py

@@ -0,0 +1,9 @@
+import pytest
+
+from .models import BulkTestModel
+
+
+@pytest.fixture()
+def objects():
+    BulkTestModel.objects.bulk_create([BulkTestModel() for _ in range(10)])
+    return BulkTestModel.objects.all()

bulk_signals/tests/test_skip.py

@@ -0,0 +1,47 @@
+import pytest
+
+from .models import BulkTestModel
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture(params=[True, False])
+def skip(request):
+    return request.param
+
+
+def test_bulk_create(mocker, skip):
+    create_stub = mocker.patch("bulk_signals.tests.models.create_stub")
+
+    BulkTestModel.objects.bulk_create(
+        [BulkTestModel() for _ in range(10)], skip_signal=skip
+    )
+
+    assert create_stub.call_count == 0 if skip else 1
+
+
+def test_bulk_update(mocker, objects, skip):
+    bulk_update_stub = mocker.patch("bulk_signals.tests.models.update_stub")
+
+    BulkTestModel.objects.bulk_update(objects, ["num"], skip_signal=skip)
+
+    assert bulk_update_stub.call_count == 0 if skip else 1
+
+
+def test_update(mocker, objects, skip):
+    update_stub = mocker.patch("bulk_signals.tests.models.query_update_stub")
+
+    BulkTestModel.objects.update(num=1, skip_signal=skip)
+
+    assert update_stub.call_count == 0 if skip else 2
+
+
+def test_custom_key(mocker, skip, settings):
+    settings.BULK_SIGNALS_SKIP_KEY = "different"
+    create_stub = mocker.patch("bulk_signals.tests.models.create_stub")
+
+    BulkTestModel.objects.bulk_create(
+        [BulkTestModel() for _ in range(10)], different=skip
+    )
+
+    assert create_stub.call_count == 0 if skip else 1