Django signals w akcji

Aktualnie pracuję nad projektem Django w ramach AIQuest, który posiada wiele aplikacji, które klient chce rozdzielić na micro-services w późniejszym terminie. Musieliśmy ustalić jak chcemy podejść do problemu, tak aby nie tworzyć zbyt wielu zależności między wspomnianymi aplikacjami. Po rozmowach między sobą oraz z klientem stwierdziliśmy, że wykorzystamy Signals - strategię która pozwala na kontakt między rozdzielonymi aplikacjami, gdy specjalne wydarzenie będzie miało miejsce.

W tym wpisie pokażę jak stworzyć testy oraz zaimplementować prosty signal oraz receiver.

Use case naszego przykładu

Użytkownik robi zapytanie HTTP POST do endpointu z aplikacji api w naszym projekcie Django. Zapytanie te powinno stworzyć signal, który powiadomi drugą aplikację nazwaną users. Aplikacja users powinna posiadać receiver, który pobierze te dane i wrzuci je do bazy danych.

Zacznijmy od testów

Dla tego przykładu będziemy używać pytest oraz unittest.mock.

Zaczniemy od stworzenia testu w api/tests.py aby sprawdzić czy odpalenie endpointu w API wyśle signal, który będzie zawierał pole name. Zmienna client przekazana do testu jest to fixture w pliku conftest.py i reprezentuje instancje Django test client.

# api/tests.py
import pytest
from unittest import mock

def test__add_user_signal(client):
    # Request, który jest wysłany przez użytkownika
    request = {
        'name': 'Lukasz'
    }

    # Używamy mock aby przechwycić wysyłany sygnał
    with mock.patch('api.signals.user_signal.send') as signal_send:

        # Pobieramy odpowiedź po wywołaniu endpointa
        response = client.post(
            '/api/v1/api/user/',
            data=request,
            content_type='application/json'
        )
        
        # Sprawdzamy czy sygnał został wysłany
        signal_send.assert_called_once_with(
            name='Lukasz'
        )

Teraz, gdy mamy już testy dla naszego sygnału, powinniśmy napisać testy, które sprawdzą czy receiver w aplikacji users działa tak jak powinien. Następujący kod powinien znaleźć się w users/tests.py.

# users/tests.py
import pytest

from api.signals import user_signal  # Importujemy sygnał
from users.models import User  # Importujemy model User


def test__can_receive_signal():
    # Sprawdzamy ile przedmiotów było w bazie
    old_count = len(User.objects.all())

    # Wysyłamy sygnał
    user_signal.send(
        None,
        name='Lukasz'
    )

    # Pobieramy nową ilość
    new_count = len(User.objects.all())

    # Sprawdzamy czy User został dodany
    assert new_count > old_count

Tworzymy Django Signal

Skoro już skończyliśmy pisać testy, powinniśmy zaimplementować nasz faktycnzy kod. Zaczniemy od stworzenia pliku api/signals.py, który będzie przechowywał wszystkie sygnały, które są wysyłane z aplikacji api (dla naszego przykładu będzie to tylko jeden).

# api/signals.py
from django.dispatch import Signal

# Tworzymy sygnał
user_signal = Signal(
    providing_args=['name']
)

Skoro to mamy z głowy, to powinniśmy pójść do naszego widoku odpowiedzialnego za tworzenie Usera i nadpisać perform_create() dla api/v1/api/user, tak aby wysyłało następujący kod:

from .signals import user_signal  # Dodaj na górze pliku

...

def perform_create(self, serializer):
    # Wcześniejszy kod idzie tutaj
    user_signal.send(
        name=serializer.validated_data['name']
    )

I to wszystko co jest potrzebne by wysłać sygnał.

Tworzymy receiver

Aby nasz receiver zaczął działać musimy ogarnąć pare rzeczy. Przede wszystkim musimy dodać następującą linijkę w users/__init__.py:

# users/__init__.py
default_app_config = 'users.apps.UserConfig'

Teraz musimy dodać funkcję ready() aby zaimportować receivers w pliku users/apps.py:

# users/apps.py
from django.apps import AppConfig


class UsersConfig(AppConfig):
    name = 'users'

    def ready(self):
        # Importujemy receivers
        from users import receivers

To pozwoli naszej aplikacji na zapewnienie działania receivers, co prowadzi nas do następnego kroku, czyli napisania kodu tworzenia nowego usera, gdy przyleci do nas sygnał zawierający pole name. Stwórz nowy plik nazwany users/receivers.py z następującym kodem:

# users/receivers.py
from django.dispatch import receiver

from users import service
from api.signals import user_signal
from users.models import User


@receiver(user_signal)
def new_user_handler(sender, **kwargs):
    # Zapisujemy Usera do bazy danych
    user = User(name=kwargs['name'])
    user.save()

W faktycznym projekcie, warto przenieść operacje bazodanowe do oddzielnej warstwy serwisowej, ale dla tego przykładu tyle wystarczy. Jeżeli wszystko zrobiłeś tak jak trzeba, to oba testy powinny przejść pozytywnie. Dobra robota!