System Design ‐ Notification

Flow Diagram

sequenceDiagram
    participant User
    participant Notification Module
    participant Event Publisher
    participant Event Subscriber

    User->>Event Publisher: Trigger event (e.g., create new post, comment, like, save)
    Event Publisher->>Event Subscriber: Publish event
    Event Subscriber->>Notification Module: Send notification
    Notification Module->>User: Send notification

Loading

ERD

erDiagram
  Notice{
      int id PK "auto_increment"
      string user FK "user.id"
      string subject
      string content
      string notice_type "Post|Comment|Reply|Follow|Like|Save|System"
      string read_status "Read|Unread"
      string send_status "Wait|Sent|Error"
      datetime create_at
      datetime update_at
}

Loading

Publish–subscribe in App

Notification Signal

Notification signal is definded in /app/notice/__init__.py:

from blinker import Namespace

signals = Namespace()
notification_signal = signals.signal("notification")

Notification Signal Handler

Notification signal handler is also definded in /app/notice/__init__.py:

def handle_notification(_, **kwargs: dict) -> None:
    """Function to handle notification event."""
 
    (Some Code here)
  
    current_app.logger.info(
        f"Notification: {notice_module}, {notice_action} inserted into database"
    )

notification_signal.connect(handle_notification)

Notification Publisher

See: /app/notice/events.py, use it wherever you need to publish a notification event.

def notice_event(
    user_id: str = "anonymity", notice_type: NoticeTypeEnum = NoticeTypeEnum.SYSTEM
) -> None:
    """Function to send notice event."""

    if current_user.is_authenticated:
        user_id = current_user.id

    current_app.logger.info(
        "Notice Event - user_id: %s, notice_type: %s", user_id, notice_type.value
    )
    notification_signal.send("app", user_id=user_id, notice_type=notice_type.value)

Usage

from app.notice.events import NoticeTypeEnum, notice_event

notice_event(user_id=user.id, notice_type=NoticeTypeEnum.USER_RESET_PASSWORD)