Notification service patterns: in-app, email, push. Use when: sending user notifications, in-app alerts, email notifications, push notifications, notification preferences.
apps.notifications (EmailTemplate, EmailQueue)transaction.on_commit() to queue notifications after DB changes commit# apps/notifications/services.py
import logging
from .models import Notification
logger = logging.getLogger(__name__)
def notify_user(
*,
user_id: int,
notification_type: str,
title: str,
message: str,
action_url: str = "",
data: dict | None = None,
) -> Notification:
"""Create an in-app notification."""
return Notification.objects.create(
user_id=user_id,
notification_type=notification_type,
title=title,
message=message,
action_url=action_url,
data=data or {},
)
from django.db import transaction
def send_firmware_approved_notification(
*, firmware_id: int, user_id: int
) -> None:
"""Notify via in-app + email when firmware is approved."""
# In-app notification
notify_user(
user_id=user_id,
notification_type="firmware_approved",
title="Firmware Approved",
message="Your firmware submission has been approved.",
action_url=f"/firmwares/{firmware_id}/",
)
# Queue email (async)
transaction.on_commit(
lambda: send_templated_email.delay(
user_id=user_id,
template_name="firmware_approved",
context={"firmware_id": firmware_id},
)
)
def notify_subscribers(
*, topic_id: int, reply_id: int, author_id: int
) -> int:
"""Notify all topic subscribers about a new reply."""
subscriber_ids = list(
ForumTopicSubscription.objects
.filter(topic_id=topic_id, is_active=True)
.exclude(user_id=author_id) # Don't notify the author
.values_list("user_id", flat=True)
)
notifications = [
Notification(
user_id=uid,
notification_type="new_reply",
title="New reply in subscribed topic",
action_url=f"/forum/t/{topic_id}/#reply-{reply_id}",
)
for uid in subscriber_ids
]
Notification.objects.bulk_create(notifications, batch_size=500)
return len(notifications)
def mark_notifications_read(*, user_id: int, notification_ids: list[int]) -> int:
"""Mark specific notifications as read."""
return Notification.objects.filter(
pk__in=notification_ids, user_id=user_id, is_read=False
).update(is_read=True)
def mark_all_read(*, user_id: int) -> int:
return Notification.objects.filter(
user_id=user_id, is_read=False
).update(is_read=True)
@transaction.atomic when paired with model changessend_mail() called directly in a view without Celerybatch_size on bulk notification creation& .\.venv\Scripts\python.exe -m ruff check . --fix
& .\.venv\Scripts\python.exe -m ruff format .
& .\.venv\Scripts\python.exe manage.py check --settings=app.settings_dev