通知機能のDB設計で意識したポイントと実装

みなさんこんにちは！Yuyaです。
今回はXのクローンアプリで「通知機能」を実装した際に
DB設計について意識したこと、学んだことをアウトプットしていきます。

目的

• Xの通知機能を実装する

詳細

• ユーザーアクション（リポスト、いいね、コメント）があった際に通知ページで誰からのアクションがあったかわかるようにする
• リポスト、いいねの場合は専用のアイコンを出力
• コメントの場合は、アイコンを出力せずコメント内容を出力
• ユーザーのアクションがあった場合は、通知一覧に表示するとともにメールで通知

前提

機能実装の前に以下のモデルがあることを前提とします。

今回は設計思想にフォーカスするため、最低限のコードしか書いていません。

各モデル

• ユーザーモデル

apps/accounts/models.py

class CustomUser(AbstractUser):
    # 基本的なユーザー情報

• ポストモデル

apps/posts/models.py

class Post(models.Model):
    user = models.ForeignKey(accounts.CustomUser, ...)
    # 投稿の基本情報

• コメントモデル

apps/comments/models.py

class Comment(models.Model):
    user = models.ForeignKey(accounts.CustomUser, ...)
    post = models.ForeignKey(posts.Post, ...)
    message = models.CharField(...)

通知モデルの構成

通知モデルにおいて以下のことが必要であると考えました。

どんな情報が必要か

• どのポストへの通知か
• どのアクションの通知か
• アクションを行ったユーザーは誰か

受信者の特定方法

• 受信者は投稿作成者として設計
• 通知モデルには受信者フィールドを含めず、投稿から参照で取得
• リレーションを使用することでデータの重複を避ける

アクションタイプの設計

• choices を使用して'like', 'repost', 'comment'を管理
• 将来的な拡張性を考慮した設計

自己通知の除外

• 自分の投稿への自分のアクションは通知しない設計
• パフォーマンスとユーザー体験の両方を考慮

重複防止

• 一意制約により同じユーザーの同じポストへの同じアクションの重複を防止
• いいね取り消し時の通知削除に対応

通知モデル

apps/notifications/models.py

from django.db import models
from django.core.exceptions import ValidationError
from apps.comments.services import CommentService

# Create your models here.

class Notification(models.Model):
    ACTION_TYPE = [
        ('repost', 'リポスト'),
        ('like', 'いいね'),
        ('comment', 'コメント'),
    ]

    post = models.ForeignKey(
        "posts.Post",
        verbose_name="ポストID",
        on_delete=models.CASCADE,
        related_name="notifications"
    )
    actor = models.ForeignKey(
        "accounts.CustomUser",
        verbose_name="ユーザーID",
        on_delete=models.CASCADE,
        related_name="notifications"
    )
    action_type = models.CharField(
        choices=ACTION_TYPE,
        verbose_name='アクションタイプ',
        max_length=50,
    )
    created_at = models.DateTimeField(
        verbose_name="作成日時",
        auto_now_add=True
    )
    updated_at = models.DateTimeField(
        verbose_name="更新日時",
        auto_now=True
    )

    class Meta:
        ordering = ["-id"]
        db_table = "notifications"
        constraints = [
            # DBレベルでpost, actor, action_type の重複を防ぐ
            models.UniqueConstraint(
                fields=['post', 'actor', 'action_type'],
                name=('unique_post_actor_action_type')
            )
        ]

    # アプリレベルで二重通知を防ぐ
    def clean(self):
        if Notification.objects.filter(post=self.post, actor=self.actor, action_type=self.action_type).exists():
            raise ValidationError("この通知は既に実行済みです")

    def __str__(self):
        return f"UserID:{self.actor.id} -> {self.post.id} is {self.action_type}"

通知がいいね、リポストの場合

最終的な画面は以下のイメージです。

• 左側
  • どの通知かわかるアイコン
  • リポスト → 矢印
  • いいね → ハート
• 右側
  • アクションしたユーザーの情報
  • 投稿元の内容

これらの情報の取得は割と容易で
actor, postの各リレーションから取得することができます。

実装コード

モデルにアイコン取得メソッドを追加

apps/notifications/models.py

...
def get_notification_icon(self):
    """
    通知アイコンを管理する関数

    Returns:
        str: SVGアイコンのHTML文字列。該当するアイコンがない場合は空文字を返す
    """
    icons = {
        'repost': '''<svg xmlns="http://www.w3.org/2000/svg" width="28" height="28" fill="#49BF00" class="bi bi-arrow-down-up" viewBox="0 0 16 16">
                    <path fill-rule="evenodd" d="M11.5 15a.5.5 0 0 0 .5-.5V2.707l3.146 3.147a.5.5 0 0 0 .708-.708l-4-4a.5.5 0 0 0-.708 0l-4 4a.5.5 0 1 0 .708.708L11 2.707V14.5a.5.5 0 0 0 .5.5m-7-14a.5.5 0 0 1 .5.5v11.793l3.146-3.147a.5.5 0 0 1 .708.708l-4 4a.5.5 0 0 1-.708 0l-4-4a.5.5 0 0 1 .708-.708L4 13.293V1.5a.5.5 0 0 1 .5-.5"/>
                    </svg>''',
        'like': '''<svg xmlns="http://www.w3.org/2000/svg" width="28" height="28" fill="red" class="bi bi-heart-fill" viewBox="0 0 16 16">
                    <path fill-rule="evenodd" d="M8 1.314C12.438-3.248 23.534 4.735 8 15-7.534 4.736 3.562-3.248 8 1.314"/>
                    </svg>''',
    }
    return icons.get(self.action_type, '')
...

シグナルで通知の自動生成

シグナル設定

apps/notifications/signals.py

import logging

from django.db.models.signals import post_save, post_delete
from apps.accounts.models import CustomUser
from django.dispatch import receiver
from apps.profiles.models import Profile
from apps.reposts.models import Repost
from apps.likes.models import Like
from apps.comments.models import Comment
from apps.notifications.models import Notification
from django.core.mail import send_mail
from config.settings import DEFAULT_FROM_EMAIL

def create_notification(post, actor, action_type):
    """
    通知を作成する（自己通知は除外）

    Args:
        post: 対象の投稿
        actor: アクションを実行したユーザー
        action_type: ユーザーアクションの種類（repost, like, comment)
    """
    try:
        # 自己通知を除外
        if post.user.id == actor.id:
            return
        
        # 通知作成
        Notification.objects.create(
            post=post,
            actor=actor,
            action_type=action_type
        )

        subject = f"{action_type}通知"
        message = f"{actor.profile.nickname}さんが、あなたの投稿に{action_type}しました"
        from_email = DEFAULT_FROM_EMAIL

        if subject and message and from_email:
            try:
                send_mail(subject, message, from_email, [post.user.email])
                print(f"Email sent for {action_type} notification")
            except Exception as e:
                print(f"Failed to send email for {action_type} notification: {e}")
        print(f"Successfully created {action_type} notification")

    except Exception as e:
        logging.error(f"Failed to create {action_type} notification: {e}")

def delete_notification(post, actor, action_type):
    """
    通知を削除する

    Args:
        post: 対象の投稿
        actor: アクションを実行したユーザー
        action_type: ユーザーアクションの種類（repost, like, comment)
    """
    try:
        # 通知削除
        Notification.objects.filter(
            post=post,
            actor=actor,
            action_type=action_type
        ).delete()
        print(f"Successfully deleted {action_type} notification")
    except Exception as e:
        print(f"Failed to delete {action_type} notification: {e}")


# プロフィール作成
@receiver(post_save, sender=CustomUser)
def create_profile(sender, instance, created, **kwargs):
    if created:
        Profile.objects.create(
            user=instance,
            nickname=instance.username
        )

# リポスト通知
@receiver(post_save, sender=Repost)
def create_repost_notification(sender, instance, created, **kwargs):
    if created:
        create_notification(
            post=instance.post,
            actor=instance.user,
            action_type='repost',
        )

# リポスト削除時、通知レコードも削除
@receiver(post_delete, sender=Repost)
def delete_repost_notification(sender, instance, **kwargs):
    delete_notification(
        post=instance.post,
        actor=instance.user,
        action_type='repost',
    )

# いいね通知
@receiver(post_save, sender=Like)
def create_like_notification(sender, instance, created, **kwargs):
    if created:
        create_notification(
            post=instance.post,
            actor=instance.user,
            action_type='like',
        )

# いいね削除時、通知レコードも削除
@receiver(post_delete, sender=Like)
def delete_like_notification(sender, instance, **kwargs):
    delete_notification(
        post=instance.post,
        actor=instance.user,
        action_type='like',
    )

# コメント通知
@receiver(post_save, sender=Comment)
def create_comment_notification(sender, instance, created, **kwargs):
    if created:
        create_notification(
            post=instance.post,
            actor=instance.user,
            action_type='comment'
        )

アプリの設定

apps/notifications/apps.py

from django.apps import AppConfig


class NotificationsConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'apps.notifications'
    verbose_name='通知'

    def ready(self):
        import apps.notifications.signals # noqa: F401

URL設定

メインURL設定

config/urls.py

from django.contrib import admin
from django.urls import include, path
from apps.accounts.views import CustomLoginView, HomePageView

urlpatterns = [
    ...
    path('notifications/', include('apps.notifications.urls')),
    ...
]

通知アプリのURL設定

apps/notifications/urls.py

from django.urls import path
from .views import NotificationListView

app_name = 'notifications'

urlpatterns = [
    path('', NotificationListView.as_view(), name="index"),
]

ビューの実装

apps/notifications/views.py

from django.contrib.auth.mixins import LoginRequiredMixin
from django.views.generic.list import ListView
from .models import Notification

# Create your views here.

class NotificationListView(LoginRequiredMixin, ListView):
    model = Notification
    context_object_name = "notifications"
    template_name = "notifications.html"

    def get_queryset(self):
        return Notification.objects.filter(post__user=self.request.user).select_related(
            'actor',  # アクションしたユーザーの情報
            'actor__profile',  # アクションしたユーザーのプロフィール情報
            'post'  # アクションされた投稿の情報
        ).order_by('-created_at')
    
    def get_context_data(self, **kwargs):
        return super().get_context_data(**kwargs)

テンプレートからのアクセス

{% for notification in notifications %}
    {{ notification.get_notification_icon|safe }}
    <p>{{ notification.actor.username }} {{ notification.action_type }} your post</p>
    <img src="{{ notification.actor.profile.icon.url }}" alt="プロフィール画像">
    <p>{{ notification.post.message }}</p>
{% endfor %}

まとめ

今回はXクローンアプリの通知機能実装において、DB設計で意識したポイントを中心に解説しました。

設計のポイント

• リレーションの活用: 受信者情報を投稿から参照することでデータ重複を回避
• 拡張性を考慮: choicesでアクションタイプを管理し、将来的な機能追加に対応
• 重複防止: DB制約とアプリレベル両方で二重通知を防止
• パフォーマンス重視: select_relatedでN+1問題を回避
• 自動化の実現: シグナルによる通知の自動生成・削除

学んだこと

通知機能は単純に見えて、実際は様々な考慮事項があることを学びました。特にDB設計の段階で「どんな情報が必要か」「どう効率的に取得するか」を整理することの重要性を実感しています。

次回はコメント通知の実装について、今回とは別のデータ取得パターンを紹介予定です！