シグナルのインポート

from django.db.models.signals import (
    pre_save,
    post_save,
    pre_delete,
    post_delete,
    m2m_changed,
)
from django.core.signals import request_started, request_finished

myapp/signals.py

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth.models import User

@receiver(post_save, sender=User)
def user_saved(sender, instance, created, **kwargs):
    if created:
        print(f"新しいユーザーが作成されました: {instance.username}")
    else:
        print(f"ユーザーが更新されました: {instance.username}")

myapp/signals.py

from django.db.models.signals import post_save
from django.contrib.auth.models import User

def user_saved(sender, instance, created, **kwargs):
    if created:
        print(f"新しいユーザーが作成されました: {instance.username}")

# connect() で手動接続
post_save.connect(user_saved, sender=User)

myapp/apps.py

from django.apps import AppConfig

class MyappConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'myapp'

    def ready(self):
        import myapp.signals  # ここでシグナルを読み込む

ディレクトリ構成

myapp/
├── __init__.py
├── apps.py          # ready() でシグナルを読み込む
├── models.py        # モデル定義
├── signals.py       # シグナルハンドラをまとめる
├── views.py
└── admin.py

myapp/signals.py

import django.dispatch

# カスタムシグナルの定義
order_completed = django.dispatch.Signal()  # 注文完了シグナル
payment_received = django.dispatch.Signal()  # 支払い受領シグナル

myapp/views.py

from myapp.signals import order_completed

def complete_order(request, order_id):
    order = Order.objects.get(id=order_id)
    order.status = 'completed'
    order.save()

    # カスタムシグナルを送信
    order_completed.send(
        sender=Order,
        order=order,
        user=request.user,
    )
    return redirect('order_detail', order_id=order.id)

notifications/signals.py

from django.dispatch import receiver
from myapp.signals import order_completed

@receiver(order_completed)
def send_order_confirmation(sender, order, user, **kwargs):
    """注文完了メールを送信"""
    send_mail(
        subject=f'注文 #{order.id} が完了しました',
        message=f'{user.username} 様、ご注文ありがとうございます。',
        from_email='shop@example.com',
        recipient_list=[user.email],
    )

@receiver(order_completed)
def update_inventory(sender, order, **kwargs):
    """在庫を更新"""
    for item in order.items.all():
        item.product.stock -= item.quantity
        item.product.save()

send_robust() の使用例

# send_robust() は例外が起きてもすべての受信者を実行する
results = order_completed.send_robust(
    sender=Order,
    order=order,
    user=request.user,
)

# 各受信者の結果を確認
for receiver_func, response in results:
    if isinstance(response, Exception):
        print(f"エラー: {receiver_func.__name__} - {response}")
    else:
        print(f"成功: {receiver_func.__name__}")

トランザクション完了後に実行する

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

@receiver(post_save, sender=Order)
def notify_order_saved(sender, instance, created, **kwargs):
    # post_save はトランザクションのコミット前に発火する可能性がある
    # 外部APIの呼び出しなどはトランザクション完了後に行う
    transaction.on_commit(
        lambda: send_notification(instance)
    )

無限ループを防ぐパターン

@receiver(post_save, sender=Article)
def update_word_count(sender, instance, **kwargs):
    # NG: save() を呼ぶと post_save が再発火 → 無限ループ
    # instance.word_count = len(instance.content.split())
    # instance.save()

    # OK: update() はシグナルを発火しない
    Article.objects.filter(pk=instance.pk).update(
        word_count=len(instance.content.split())
    )

dispatch_uid の指定

# dispatch_uid を指定すると、同じIDで2回以上接続されない
@receiver(post_save, sender=User, dispatch_uid="create_user_profile")
def create_profile(sender, instance, created, **kwargs):
    if created:
        Profile.objects.create(user=instance)

accounts/models.py

from django.db import models
from django.contrib.auth.models import User

class Profile(models.Model):
    user = models.OneToOneField(User, on_delete=models.CASCADE)
    bio = models.TextField(blank=True, default='')
    avatar = models.ImageField(upload_to='avatars/', blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return f'{self.user.username} のプロフィール'

accounts/signals.py

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth.models import User
from .models import Profile

@receiver(post_save, sender=User, dispatch_uid="create_user_profile")
def create_profile(sender, instance, created, **kwargs):
    """ユーザー作成時にプロフィールを自動作成"""
    if created:
        Profile.objects.create(user=instance)

@receiver(post_save, sender=User, dispatch_uid="save_user_profile")
def save_profile(sender, instance, **kwargs):
    """ユーザー保存時にプロフィールも保存"""
    if hasattr(instance, 'profile'):
        instance.profile.save()

accounts/apps.py

from django.apps import AppConfig

class AccountsConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'accounts'

    def ready(self):
        import accounts.signals

audit/models.py

from django.db import models

class AuditLog(models.Model):
    ACTION_CHOICES = [
        ('create', '作成'),
        ('update', '更新'),
        ('delete', '削除'),
    ]
    model_name = models.CharField(max_length=100)
    object_id = models.IntegerField()
    action = models.CharField(max_length=10, choices=ACTION_CHOICES)
    changes = models.JSONField(default=dict, blank=True)
    timestamp = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ['-timestamp']

    def __str__(self):
        return f'{self.model_name} #{self.object_id} {self.action}'

audit/signals.py

from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from .models import AuditLog

# 監査対象のモデル一覧
AUDITED_MODELS = []

def audit_model(model_class):
    """デコレータ: モデルを監査対象に登録"""
    AUDITED_MODELS.append(model_class)
    return model_class

@receiver(post_save)
def log_save(sender, instance, created, **kwargs):
    if sender not in AUDITED_MODELS:
        return

    AuditLog.objects.create(
        model_name=sender.__name__,
        object_id=instance.pk,
        action='create' if created else 'update',
    )

@receiver(post_delete)
def log_delete(sender, instance, **kwargs):
    if sender not in AUDITED_MODELS:
        return

    AuditLog.objects.create(
        model_name=sender.__name__,
        object_id=instance.pk,
        action='delete',
    )

myapp/models.py（監査対象のモデル）

from django.db import models
from audit.signals import audit_model

@audit_model
class Article(models.Model):
    title = models.CharField(max_length=200)
    content = models.TextField()
    # ... このモデルの作成・更新・削除が自動でログに記録される

tests/test_user.py

from django.test import TestCase
from django.contrib.auth.models import User
from django.db.models.signals import post_save
from accounts.signals import create_profile

class UserTestCase(TestCase):
    def setUp(self):
        # テスト中はプロフィール自動作成を無効化
        post_save.disconnect(create_profile, sender=User)

    def tearDown(self):
        # テスト後に再接続
        post_save.connect(create_profile, sender=User)

    def test_user_creation_without_profile(self):
        user = User.objects.create_user('testuser', 'test@example.com', 'pass')
        self.assertFalse(hasattr(user, 'profile'))