django.dispatch.receiver()の使い方

特定のテーブルのデータが変更されたらログ出したい時とかありますね。
簡単な例としては、商品情報を変更した場合のみログを出したいとか。
そんな時に使えるのがsignal！

Django includes a “signal dispatcher” which helps decoupled applications get notified when actions occur elsewhere in the framework. In a nutshell, signals allow certain senders to notify a set of receivers that some action has taken place. They’re especially useful when many pieces of code may be interested in the same events.

意味わかりませんね。
例を見ていきましょう。

例1 レコードに新規追加された時のログとか通知に

printで出力しますけど、logger使えば当然ログ出力できますので試してみてください。

django/testproject/webtestapp/signals.py

from django.db.models.signals import post_save
from django.dispatch import receiver

from webtestapp.models import Product


@receiver(post_save, sender=Product)
def create_product(sender, instance, created, **kwargs):
    print('作成時のテスト')
    if created:
        print('[Product]: 商品情報を作成しました')

django/testproject/webtestapp/apps.py

from django.apps import AppConfig


class WebtestappConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'webtestapp'

    def ready(self):
        from webtestapp import signals

apps.pyにreadyを追記。importするだけで暗黙的に(よしなに)使えます。

試しにDjango shellで実行してみると

>>> from webtestapp.models import Product
>>> Product.objects.create(name='sample', price=100, category='daily', description='サンプル')
作成時のテスト
[Product]: 商品情報を作成しました
<Product: sample>

明示的に書く場合

先ほどの例ではimportだけreadyに追加しました。
明示的に書く場合は以下のように書きます。こちらの方が安全かと思います。

django/testproject/webtestapp/apps.py

from django.apps import AppConfig
from django.core.signals import request_finished

class WebtestappConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'webtestapp'

    def ready(self):
        from webtestapp import signals
        request_finished.connect(signals.create_product)

例２　レコードに追加した時に他のデータも作成。更新ならログ出す

例えば、結構重要なデータを新規で作ったときに履歴やログデータとして別テーブルに保存したい。変更された場合はログ出したいって想定で。
apps.pyは例1と同じです。

django/testproject/webtestapp/signals.py

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.forms import model_to_dict

from webtestapp.models import Product, ProductHistory

@receiver(post_save, sender=Product)
def create_product(sender, instance, created, **kwargs):
    data = model_to_dict(instance, fields=[field.name for field in instance._meta.fields])
    # idをproduct_idに変更
    data['product_id'] = data.pop('id')
    if created:
        print('[Product]: 商品情報を作成しました')
        ProductHistory.objects.create(**data)
        print('履歴を作成')
    else:
        print('[Product]: 商品情報を更新しました')

>>> from webtestapp.models import Product, ProductHistory
>>> new_p = Product.objects.create(name='sample3', price=300, category='daily', description='サンプル3')
[Product]: 商品情報を作成しました
>>> ph = ProductHistory.objects.get(product_id=new_p.id)
>>> ph
<ProductHistory: sample3>
>>> new_p.price = 500
>>> new_p.save()
[Product]: 商品情報を更新しました

新規作成と更新で処理を分けることができましたね。

補足

これだと保存されたら毎回処理が呼ばれます。
priceが変更になった時だけログ出したいとか処理させたいなどの場合は、post_saveを使うのではなくpre_saveというsignalを使ってDBからデータを取得後、更新されたinstanceの値を比較して違う場合にログを出せば良いと思います。(保存処理でエラー出たらログだけ出ちゃいそうですが)
※ pre_saveはデータをDBに保存前に処理を行なってくれるsignal

注意

querysetのupdate()使うとsave()メソッドを使っていないのでsignal処理が動きません。
というか、その他にもupdate()使うとauto_now効かないとかあるので使用するときには注意必要ですね、、
model_to_dictは面倒な書き方してますが
many_to_manyなければmodel_to_dict(instance)で良いはずです。

今回使用したモデル定義

django/testproject/webtestapp/models.py

class Product(models.Model):
    CATEGORY = (
            ('daily', '日用品'),
            ('child', '子供向け'),
            ('sports', 'スポーツ'),
            )

    name = models.CharField(max_length=100)
    price = models.IntegerField()
    category = models.CharField(max_length=50, choices=CATEGORY)
    description = models.CharField(max_length=2000, blank=True)
    updated_at = models.DateTimeField(auto_now=True)
    created_at = models.DateTimeField(auto_now_add=True, null=True)
    tags = models.ManyToManyField(Tag)

    def __str__(self):
        return self.name


class ProductHistory(models.Model):
    CATEGORY = (
            ('daily', '日用品'),
            ('child', '子供向け'),
            ('sports', 'スポーツ'),
            )

    product_id = models.IntegerField(null=True)
    name = models.CharField(max_length=100)
    price = models.IntegerField()
    category = models.CharField(max_length=50, choices=CATEGORY)
    description = models.CharField(max_length=2000, blank=True)
    updated_at = models.DateTimeField(auto_now=True)
    created_at = models.DateTimeField(auto_now_add=True, null=True)

    def __str__(self):
        return self.name

注意点単体テストの際など

unit testなど実行するときに何度も同じものが実行される可能性があります。
そういう場合の対策に被らないように文字列を与えてあげるといいらしいです。

https://docs.djangoproject.com/en/4.1/topics/signals/#preventing-duplicate-signals

apps.py

from django.core.signals import request_finished

request_finished.connect(my_callback, dispatch_uid="my_unique_identifier")

これは試してないですけど、自分でき決めた文字列をdispatch_uid引数に渡せばいいだけ。

使用したsignalの解説

今回使用したsignalを少し解説。

post_save

django.db.models.signals.post_save

save()メソッドが呼ばれた後に実行される

引数

receiver_function(sender, instance, created, raw, using, update_fields)

sender:
- モデルクラス渡す
instance:
- saveしたモデルインスタンス。保存したデータのID使ってログ出すとか色々使えそう。
created: boolean
- レコード追加されたときにTrueなる。更新処理の時はFalseなのでよく使うはず
raw: boolean
- モデルが直で保存される場合 (フィクスチャをロードするときとか？)、True。Trueの時、DBの他のレコードいじると問題出るかも。
using:
- 使用されるデータベースエイリアス
update_fields:
- 更新されるフィールド名。save()に渡した引数つまり更新するフィールドが入る。特定のフィールドを更新されたらログ出したいとかで。
- 更新の時にフィールド何も渡さなかった場合はNoneになる。instace.save()みたいに全部保存するとNoneになるので使いにくいかも。

update_fieldsについて補足

update_fieldsのデータをprintしてみます

>>> new_p.price = 1000
>>> new_p.save(update_fields=['price'])
frozenset({'price'})
[Product]: 商品情報を更新しました

frozenset({'price'})がupdate_fields引数で受け取った値。

これをnew_p.save(update_fields=['price'])でなく、new_p.save()するとNoneになる

その他のSignal

今回使っていないsignalも多数ありますので公式ページのリンク載せておきます。

pre_save: 保存する前にやらせたい処理がある場合に
post_delete: 削除されたときにやらせたい処理書く(Django4.1からの機能)
m2m_change: モデルでManyToManyFieldが変更されたときに行う
pre_migrate、post_migrate: 本番環境でmigration不安なときに正常に終了したらlogとか通知処理させたいとか(Django 4.0からの機能)

色々あるので使えそうなものは試してみてください。

[Django]データ作成・保存時に処理させるSignalを使う[django.dispatch.receiver]