Changes for Recoco

django_webhook/apps.py

@@ -4,6 +4,7 @@
 
 class WebhooksConfig(AppConfig):
     name = "django_webhook"
+    default_auto_field = "django.db.models.AutoField"
 
     def ready(self):
         from django.conf import settings

django_webhook/signals.py

@@ -6,8 +6,10 @@
 from django.db import models
 from django.db.models.signals import ModelSignal, post_delete, post_save
 from django.forms import model_to_dict
+from django.utils.module_loading import import_string
 
 from django_webhook.models import Webhook
+from typing import assert_never
 
 from .tasks import fire_webhook
 from .util import cache
@@ -37,55 +39,69 @@ def run(self, sender, created=False, instance=None, **kwargs):
                 action_type = UPDATE
             case "post_delete":
                 action_type = DELETE
+            case _:
+                assert_never(self.signal_name)
 
         topic = f"{self.model_label}/{action_type}"
-        webhook_ids = _find_webhooks(topic)
+        webhook_ids = self.find_webhooks(topic=topic, instance=instance)
 
         for id, uuid in webhook_ids:
             payload = dict(
                 topic=topic,
-                object=model_dict(instance),
+                object=self.model_dict(instance),
                 object_type=self.model_label,
                 webhook_uuid=str(uuid),
             )
             fire_webhook.delay(id, payload)
 
+    def find_webhooks(self, topic: str, instance=None):
+        return _find_webhooks(topic)
+
     def connect(self):
         self.signal.connect(
             self.run, sender=self.model_cls, weak=False, dispatch_uid=self.uid  # type: ignore
         )
 
     @property
     def uid(self):
-        return f"django_webhook_{self.model_label}_{self.signal}"
+        return f"django_webhook_{self.model_label}_{self.signal_name}"
 
     @property
     def model_label(self):
         return self.model_cls._meta.label
 
+    def model_dict(self, model):
+        """
+        Returns the model instance as a dict, nested values for related models.
+        """
+        fields = {
+            field.name: field.value_from_object(model) for field in model._meta.fields
+        }
+        return model_to_dict(model, fields=fields)  # type: ignore
+
 
 def connect_signals():
+    SignalListenerClass = import_string(
+        dotted_path=settings.DJANGO_WEBHOOK.get(
+            "SIGNAL_LISTENER", "django_webhook.signals.SignalListener"
+        )
+    )
+    if not issubclass(SignalListenerClass, SignalListener):
+        raise ValueError(
+            f"{SignalListenerClass} must be a subclass of {SignalListener}"
+        )
+
     for cls in _active_models():
-        post_save_listener = SignalListener(
+        post_save_listener = SignalListenerClass(
             signal=post_save, signal_name="post_save", model_cls=cls
         )
-        post_delete_listener = SignalListener(
+        post_delete_listener = SignalListenerClass(
             signal=post_delete, signal_name="post_delete", model_cls=cls
         )
         post_save_listener.connect()
         post_delete_listener.connect()
 
 
-def model_dict(model):
-    """
-    Returns the model instance as a dict, nested values for related models.
-    """
-    fields = {
-        field.name: field.value_from_object(model) for field in model._meta.fields
-    }
-    return model_to_dict(model, fields=fields)  # type: ignore
-
-
 def _active_models():
     model_names = settings.DJANGO_WEBHOOK.get("MODELS", [])
     model_classes = []

tests/test_override.py

@@ -0,0 +1,90 @@
+from django.db.models.base import Model as Model
+from django.db.models.signals import ModelSignal
+from django_webhook.signals import SignalListener, connect_signals
+from django_webhook.settings import defaults
+from django.db.models.signals import post_save, post_delete
+from tests.models import Country
+from django.test import override_settings
+import pytest
+
+from django_webhook.test_factories import WebhookFactory, WebhookTopicFactory
+import json
+
+from unittest.mock import patch
+
+
+class DummySignalListener:
+    pass
+
+
+def test_invalid_signal_listener():
+    with override_settings(
+        DJANGO_WEBHOOK=defaults
+        | {"SIGNAL_LISTENER": "tests.test_override.UnknownSignalListener"}
+    ):
+        with pytest.raises(ImportError):
+            connect_signals()
+
+    with override_settings(
+        DJANGO_WEBHOOK=defaults
+        | {"SIGNAL_LISTENER": "tests.test_override.DummySignalListener"}
+    ):
+        with pytest.raises(ValueError):
+            connect_signals()
+
+
+class CustomSignalListener(SignalListener):
+    def run(self, sender, created=False, instance=None, **kwargs):
+        if isinstance(instance, Country) and instance.name in [
+            "France",
+            "Spain",
+            "Italy",
+            "Germany",
+        ]:
+            return super().run(sender, created, instance, **kwargs)
+
+    def model_dict(self, model):
+        return {"id": model.id, "country_name": model.name}
+
+
+def _disconnect_signals():
+    assert post_save.disconnect(
+        sender=Country,
+        dispatch_uid="django_webhook_tests.Country_post_save",
+    )
+    assert post_delete.disconnect(
+        sender=Country,
+        dispatch_uid="django_webhook_tests.Country_post_delete",
+    )
+
+
+@pytest.mark.django_db
+def test_override_signal_listener(responses):
+    country = Country.objects.create(name="France")
+    webhook = WebhookFactory(
+        topics=[
+            WebhookTopicFactory(name="tests.Country/update")
+        ]
+    )
+    responses.post(webhook.url)
+
+    _disconnect_signals()
+
+    with override_settings(
+        DJANGO_WEBHOOK={
+            "SIGNAL_LISTENER": "tests.test_override.CustomSignalListener",
+            "MODELS": ["tests.Country"],
+        }
+        | defaults
+    ):
+        connect_signals()
+
+    country.save()
+
+    assert len(responses.calls) == 1
+    assert json.loads(responses.calls[0].request.body) == {
+        "topic": "tests.Country/update",
+        "object": {"id": country.id, "country_name": "France"},
+        "object_type": "tests.Country",
+        "webhook_uuid": str(webhook.uuid),
+    }