Signal System

from django.dispatch import Signal

notify = Signal()

def notify_handler(verb, **kwargs):
    """
    Create Notification instances from signal calls.
    
    Args:
        verb (str): Action verb describing what happened
        **kwargs: Additional parameters for notification creation
        
    Required kwargs:
        recipient: User, Group, QuerySet, or list of Users to notify
        sender: Actor object that performed the action
        
    Optional kwargs:
        target: Target object the action was performed on
        action_object: Object linked to the action itself
        public (bool): Whether notification is public (default: True)
        description (str): Text description of the notification
        timestamp (datetime): Custom timestamp (default: now)
        level (str): Notification level (default: 'info')
        actor_for_concrete_model (bool): ContentType behavior (default: True)
        target_for_concrete_model (bool): ContentType behavior (default: True)
        action_object_for_concrete_model (bool): ContentType behavior (default: True)
        Additional kwargs stored in data JSONField if USE_JSONFIELD=True
        
    Returns:
        list: List of created Notification instances
        
    Raises:
        Various exceptions based on model validation and database constraints
    """

from django.contrib.auth.models import User
from notifications.signals import notify

# Get users
actor = User.objects.get(username='john')
recipient = User.objects.get(username='jane')

# Send simple notification
notify.send(
    sender=actor,          # Required: who performed the action
    recipient=recipient,   # Required: who should be notified
    verb='followed',       # Required: what action was performed
    description='John started following you'
)

from myapp.models import Post

post = Post.objects.get(id=1)

# Notify about action on target object
notify.send(
    sender=actor,
    recipient=post.author,  # Notify the post author
    verb='liked',
    target=post,           # The post that was liked
    description=f'{actor.username} liked your post "{post.title}"'
)

from myapp.models import Comment

comment = Comment.objects.create(
    post=post,
    author=actor,
    content='Great post!'
)

# Notify about new comment (action object) on post (target)
notify.send(
    sender=actor,
    recipient=post.author,
    verb='commented on',
    action_object=comment,  # The comment that was created
    target=post,           # The post that was commented on
    description=f'{actor.username} commented on your post'
)

# Notify multiple users individually
recipients = [user1, user2, user3]
notify.send(
    sender=actor,
    recipient=recipients,   # List of users
    verb='posted',
    target=post,
    description='New post available'
)

# Notify all users in a group
from django.contrib.auth.models import Group
editors = Group.objects.get(name='editors')
notify.send(
    sender=actor,
    recipient=editors,      # All users in group will be notified
    verb='submitted',
    target=post,
    description='New post submitted for review'
)

# Notify users from queryset
active_users = User.objects.filter(is_active=True)
notify.send(
    sender=actor,
    recipient=active_users,  # All active users
    verb='announced',
    description='System maintenance announcement'
)

# Set notification level
notify.send(
    sender=actor,
    recipient=recipient,
    verb='failed to process',
    level='error',          # 'success', 'info', 'warning', 'error'
    description='Payment processing failed'
)

# Include additional data (requires USE_JSONFIELD=True)
notify.send(
    sender=actor,
    recipient=recipient,
    verb='earned',
    description='Achievement unlocked!',
    # Additional data stored in JSONField
    achievement_name='First Post',
    points_earned=100,
    badge_url='/static/badges/first-post.png'
)

from django.utils import timezone
from datetime import timedelta

# Custom timestamp
past_time = timezone.now() - timedelta(hours=2)
notify.send(
    sender=actor,
    recipient=recipient,
    verb='completed',
    timestamp=past_time,
    description='Task completed 2 hours ago'
)

# Private notification (not public)
notify.send(
    sender=actor,
    recipient=recipient,
    verb='sent private message',
    public=False,           # Only visible to recipient
    description='You have a private message'
)

# Automatic connection in notifications/base/models.py
from notifications.signals import notify
from notifications.base.models import notify_handler

notify.connect(notify_handler, dispatch_uid='notifications.models.notification')

# Available after app is ready
import notifications
notifications.notify.send(
    sender=actor,
    recipient=recipient,
    verb='updated',
    description='Profile updated'
)

try:
    notify.send(
        sender=actor,
        recipient=recipient,
        verb='performed action',
        target=some_object
    )
except Exception as e:
    # Handle notification creation errors
    logger.error(f"Failed to create notification: {e}")
    # Application logic continues normally

from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import BlogPost

@receiver(post_save, sender=BlogPost)
def notify_blog_post_created(sender, instance, created, **kwargs):
    if created:
        # Notify followers when new blog post is created
        followers = instance.author.followers.all()
        if followers.exists():
            notify.send(
                sender=instance.author,
                recipient=followers,
                verb='published',
                target=instance,
                description=f'New blog post: {instance.title}'
            )