Django Signals extension for unifying data passed to receivers
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys


def main():
    """Run administrative tasks."""
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "tests.testapp.settings")

    from django.core.management import execute_from_command_line

    execute_from_command_line(sys.argv)


if __name__ == "__main__":
    main()

manage.py
ivellios/django-unified-signals
from .signals import UnifiedSignal

src/unified_signals/__init__.py
ivellios/django-unified-signals
class UnifiedSignalMessageTypeError(ValueError):
    pass

src/unified_signals/exceptions.py
ivellios/django-unified-signals
import typing

from django.dispatch import Signal

from .exceptions import UnifiedSignalMessageTypeError


class UnifiedSignal(Signal):
    """

    Example of defining the signal

        @dataclasses.dataclass
        SomeSignalMessage:
            some_field: int

        some_signal = UnifiedSignal(SomeSignalMessage)

    Example of sending the signal:

        some_signal.send(
            sender=SomeSenderClass,
            message=SomeSignalMessage(
                some_field=5
            )
        )

    Example of receiving the signal data:

        @receiver(some_signal)
        def handle_some_signal(sender, message: SomeSignalMessage, **kwargs):
            print(message.some_field)

    """

    def __init__(self, message_class: typing.Type, use_caching=False):
        # NOTE: In pre 4.0 Django versions you would need
        #       to add providing_args param also.
        super().__init__(use_caching=use_caching)
        self.message_class = message_class

    def _check_message_class(self, message):
        if not isinstance(message, self.message_class):
            raise UnifiedSignalMessageTypeError(
                f"Wrong message dataclass passed to the signal send: {message.__class__}. Expected {self.message_class}"
            )

    def send(self, sender: typing.Any, message: typing.Any = None, **named):
        self._check_message_class(message)
        return super().send(sender, message=message, **named)

    def send_robust(
        self,
        sender: typing.Any,
        message: typing.Any = None,
        **named: typing.Any,
    ):
        self._check_message_class(message)
        return super().send_robust(sender, **named)

src/unified_signals/signals.py
ivellios/django-unified-signals
import dataclasses
from unittest import mock

import pytest
from django.dispatch import receiver

from unified_signals.signals import UnifiedSignal
from unified_signals.exceptions import UnifiedSignalMessageTypeError


@dataclasses.dataclass
class DataMock:
    required_field: int
    optional_field: int = 5


class SenderMock:
    pass


def test_event_signal():
    signal = UnifiedSignal(DataMock)
    assert True


def test_send_signal_without_data():
    # given
    signal = UnifiedSignal(DataMock)

    # when, then
    with pytest.raises(UnifiedSignalMessageTypeError):
        signal.send(SenderMock)


def test_send_signal_with_wrong_data_type():
    # given
    signal = UnifiedSignal(DataMock)

    @dataclasses.dataclass
    class OtherDataMock:
        pass

    # when, then
    with pytest.raises(UnifiedSignalMessageTypeError):
        signal.send(SenderMock, OtherDataMock())


def test_send_signal_with_proper_data_type():
    # given
    signal = UnifiedSignal(DataMock)

    @receiver(signal)
    def handle_signal(sender, message: DataMock, **kwargs):
        assert message.required_field == 10
        assert message.__class__ == DataMock

    # when
    signal.send(
        mock.Mock(),
        DataMock(
            required_field=10,
        ),
    )


def test_send_robust_signal():
    signal = UnifiedSignal(DataMock)

    @receiver(signal)
    def handle_signal(sender, message: DataMock, **kwargs):
        assert message.required_field == 10
        assert message.__class__ == DataMock

    signal.send_robust(
        mock.Mock(),
        DataMock(
            required_field=10,
        ),
    )


def test_send_robus_signal_checks_for_wrong_type():
    signal = UnifiedSignal(DataMock)

    with pytest.raises(UnifiedSignalMessageTypeError):
        signal.send_robust(mock.Mock(), 10)

tests/test_signals.py
ivellios/django-unified-signals
# minimal settings for running tests

SECRET_KEY = "Some secret key"

tests/testapp/settings.py
ivellios/django-unified-signals