Django SignalsとCelery Tasks 2026年版: 適切な使い分けと面接対策ガイド

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

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

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

from django.dispatch import Signal

# カスタムシグナルの定義
order_completed = Signal()
payment_processed = Signal()

# シグナルの送信
from .signals import order_completed

def complete_order(order):
    # 注文完了処理
    order.status = 'completed'
    order.save()
    
    # シグナルを送信
    order_completed.send(
        sender=order.__class__,
        order=order,
        user=order.user
    )

from .signals import order_completed
from .services import NotificationService, InventoryService

@receiver(order_completed)
def send_order_confirmation(sender, order, user, **kwargs):
    """注文完了通知を送信"""
    NotificationService.send_email(
        to=user.email,
        template='order_confirmation',
        context={'order': order}
    )

@receiver(order_completed)
def update_inventory(sender, order, **kwargs):
    """在庫を更新"""
    for item in order.items.all():
        InventoryService.decrease_stock(
            product_id=item.product_id,
            quantity=item.quantity
        )

# celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Tokyo'

# tasks.py
from celery import shared_task
from django.core.mail import send_mail
from .services import ReportGenerator, ImageProcessor

@shared_task(bind=True, max_retries=3)
def send_welcome_email(self, user_id):
    """ウェルカムメールを非同期で送信"""
    try:
        from django.contrib.auth.models import User
        user = User.objects.get(id=user_id)
        send_mail(
            subject='ようこそ！',
            message=f'{user.username}様、ご登録ありがとうございます。',
            from_email='noreply@example.com',
            recipient_list=[user.email],
        )
    except User.DoesNotExist:
        pass
    except Exception as exc:
        self.retry(exc=exc, countdown=60)

@shared_task
def generate_monthly_report(month, year):
    """月次レポートを生成"""
    report = ReportGenerator.create_monthly_report(month, year)
    return report.id

@shared_task(queue='image_processing')
def process_uploaded_image(image_id):
    """アップロード画像を処理"""
    ImageProcessor.resize_and_optimize(image_id)

# views.py
from .tasks import send_welcome_email, generate_monthly_report

def register_user(request):
    user = User.objects.create_user(**form.cleaned_data)
    # タスクをキューに追加（非同期実行）
    send_welcome_email.delay(user.id)
    return redirect('registration_complete')

def request_report(request):
    # タスクをスケジュール（遅延実行）
    result = generate_monthly_report.apply_async(
        args=[request.POST['month'], request.POST['year']],
        countdown=300  # 5分後に実行
    )
    return JsonResponse({'task_id': result.id})

# Signals: 同期実行（リクエスト内で完了）
@receiver(post_save, sender=Order)
def update_order_cache(sender, instance, **kwargs):
    """キャッシュ更新は即座に反映が必要"""
    cache.set(f'order_{instance.id}', instance, timeout=3600)

# Celery: 非同期実行（リクエスト外で処理）
@shared_task
def send_order_notification(order_id):
    """通知送信は遅延しても問題ない"""
    order = Order.objects.get(id=order_id)
    NotificationService.notify_all_stakeholders(order)

# Signalsで避けるべきパターン
@receiver(post_save, sender=Product)
def bad_signal_handler(sender, instance, **kwargs):
    # 重い処理をシグナルで行うとレスポンスが遅延
    generate_product_thumbnail(instance)  # 5秒かかる処理
    sync_to_external_api(instance)  # 外部API呼び出し
    send_notification_to_subscribers(instance)  # 多数のメール送信

# 正しいアプローチ: Celeryに委譲
@receiver(post_save, sender=Product)
def product_saved_handler(sender, instance, **kwargs):
    # 軽量な処理のみシグナルで実行
    cache.delete(f'product_{instance.id}')
    # 重い処理はCeleryに委譲
    process_product_async.delay(instance.id)

@shared_task
def process_product_async(product_id):
    product = Product.objects.get(id=product_id)
    generate_product_thumbnail(product)
    sync_to_external_api(product)
    send_notification_to_subscribers(product)

from django.db import transaction

# 問題: シグナルがトランザクション完了前に実行される
@receiver(post_save, sender=Order)
def notify_on_order_save(sender, instance, **kwargs):
    # トランザクションがロールバックされても通知が送られる可能性
    send_notification.delay(instance.id)

# 解決策: transaction.on_commit を使用
@receiver(post_save, sender=Order)
def notify_on_order_save(sender, instance, created, **kwargs):
    if created:
        transaction.on_commit(
            lambda: send_notification.delay(instance.id)
        )

# 1. モデル間の自動同期
@receiver(post_save, sender=User)
def sync_user_data(sender, instance, **kwargs):
    UserSearchIndex.update_or_create(user=instance)

# 2. キャッシュ無効化
@receiver(post_save, sender=Article)
@receiver(post_delete, sender=Article)
def invalidate_article_cache(sender, instance, **kwargs):
    cache.delete_pattern(f'article_{instance.id}*')
    cache.delete('article_list')

# 3. 監査ログの記録
@receiver(pre_save, sender=SensitiveData)
def log_data_change(sender, instance, **kwargs):
    if instance.pk:
        old = SensitiveData.objects.get(pk=instance.pk)
        AuditLog.objects.create(
            model='SensitiveData',
            object_id=instance.pk,
            changes=get_changes(old, instance)
        )

# 1. 外部API連携
@shared_task(bind=True, autoretry_for=(RequestException,), retry_backoff=True)
def sync_to_crm(self, customer_id):
    customer = Customer.objects.get(id=customer_id)
    CRMClient().update_customer(customer.to_crm_format())

# 2. バッチ処理
@shared_task
def process_daily_analytics():
    yesterday = date.today() - timedelta(days=1)
    events = Event.objects.filter(date=yesterday)
    analytics = AnalyticsProcessor.process(events)
    DailyReport.objects.create(date=yesterday, data=analytics)

# 3. 定期実行タスク
from celery.schedules import crontab

app.conf.beat_schedule = {
    'cleanup-expired-sessions': {
        'task': 'myapp.tasks.cleanup_sessions',
        'schedule': crontab(hour=3, minute=0),
    },
    'generate-weekly-digest': {
        'task': 'myapp.tasks.send_weekly_digest',
        'schedule': crontab(day_of_week=1, hour=9, minute=0),
    },
}

# 回答例に含めるべきポイント:

# 1. デバッグの困難さ
# シグナルは暗黙的な呼び出しのため、コードの追跡が難しい

# 2. 循環参照のリスク
@receiver(post_save, sender=ModelA)
def update_model_b(sender, instance, **kwargs):
    instance.model_b.save()  # これがModelAの更新を引き起こす可能性

# 3. テストの複雑化
from django.db.models.signals import post_save
from django.test import TestCase
from unittest.mock import patch

class TestWithoutSignals(TestCase):
    def test_without_signal(self):
        # シグナルを無効化してテスト
        post_save.disconnect(create_user_profile, sender=User)
        try:
            user = User.objects.create(username='test')
            # プロフィールは作成されない
        finally:
            post_save.connect(create_user_profile, sender=User)

# 冪等性を確保したタスク実装
from celery import shared_task
from django.db import transaction

@shared_task(bind=True)
def process_payment(self, payment_id):
    with transaction.atomic():
        payment = Payment.objects.select_for_update().get(id=payment_id)
        
        # 既に処理済みならスキップ
        if payment.status == 'processed':
            return {'status': 'already_processed'}
        
        # 冪等キーで重複実行を防止
        idempotency_key = f'payment_{payment_id}'
        if cache.get(idempotency_key):
            return {'status': 'duplicate_request'}
        
        cache.set(idempotency_key, True, timeout=3600)
        
        # 実際の処理
        result = PaymentGateway.charge(payment)
        payment.status = 'processed'
        payment.save()
        
        return {'status': 'success', 'result': result}

# 推奨パターン: シグナルからCeleryタスクを起動
from django.db import transaction

@receiver(post_save, sender=Order)
def handle_order_created(sender, instance, created, **kwargs):
    if created:
        # トランザクション完了後にタスクを実行
        transaction.on_commit(lambda: [
            send_order_confirmation.delay(instance.id),
            update_inventory.delay(instance.id),
            notify_warehouse.delay(instance.id),
        ])

# アンチパターン: Celeryタスク内でシグナルを多発させる
@shared_task
def bad_batch_update(ids):
    for id in ids:
        # 各saveでシグナルが発火し、パフォーマンス劣化
        obj = MyModel.objects.get(id=id)
        obj.status = 'updated'
        obj.save()  # シグナル発火

# 改善版: bulk_updateでシグナルを回避
@shared_task
def good_batch_update(ids):
    MyModel.objects.filter(id__in=ids).update(status='updated')
    # 必要なら明示的にシグナルを1回だけ送信
    batch_updated.send(sender=MyModel, ids=ids)

from celery import shared_task
from celery.exceptions import MaxRetriesExceededError

@shared_task(
    bind=True,
    max_retries=5,
    default_retry_delay=60,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True
)
def resilient_external_call(self, data):
    try:
        response = ExternalService.call(data)
        return response
    except RateLimitError as exc:
        # カスタムリトライ間隔
        raise self.retry(exc=exc, countdown=exc.retry_after)
    except FatalError as exc:
        # リトライせずに失敗
        raise exc
    except MaxRetriesExceededError:
        # 最大リトライ回数超過時の処理
        FailedTask.objects.create(
            task_name='resilient_external_call',
            data=data,
            error='Max retries exceeded'
        )
        raise