Complete Examples
Here are complete examples showing typical integration patterns for the django-notifiedby package.
User Lifecycle Management
This example shows how to integrate NotifiedBy into a typical SaaS application with user signup, upgrades, and cancellations.
# models.py
from django.db import models
from django.contrib.auth.models import User
class UserProfile(models.Model):
user = models.OneToOneField(User, on_delete=models.CASCADE)
plan = models.CharField(max_length=20, default='free') # 'free', 'premium'
signup_date = models.DateTimeField(auto_now_add=True)
# signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from notifiedby import (
subscribe_to_flow,
unsubscribe_from_flow,
set_recipient_flags,
clear_recipient_flags,
)
@receiver(post_save, sender=User)
def on_user_created(sender, instance, created, **kwargs):
if created:
# Subscribe to welcome flow
subscribe_to_flow(
flow_trigger='WELCOME_SIGNUP',
email=instance.email,
first_name=instance.first_name,
last_name=instance.last_name,
flags=['free_plan', 'new_user'],
)
# views.py
from django.shortcuts import render, redirect
from django.contrib.auth.decorators import login_required
from notifiedby import subscribe_to_flow, unsubscribe_from_flow
@login_required
def upgrade_to_premium(request):
"""Handle premium upgrade"""
user = request.user
profile = user.userprofile
# Update database
profile.plan = 'premium'
profile.save()
# Update recipient flags
set_recipient_flags(email=user.email, flags=['premium'])
clear_recipient_flags(email=user.email, flags=['free_plan'])
# Subscribe to premium onboarding
subscribe_to_flow(
flow_trigger='PREMIUM_ONBOARDING',
email=user.email,
first_name=user.first_name,
)
return redirect('dashboard')
@login_required
def cancel_subscription(request):
"""Handle subscription cancellation"""
user = request.user
profile = user.userprofile
# Update database
profile.plan = 'cancelled'
profile.save()
# Update flags
set_recipient_flags(email=user.email, flags=['cancelled'])
clear_recipient_flags(email=user.email, flags=['premium', 'free_plan'])
# Unsubscribe from premium flow and subscribe to re-engagement
unsubscribe_from_flow(
flow_trigger='PREMIUM_ONBOARDING',
email=user.email,
)
subscribe_to_flow(
flow_trigger='REENGAGE_INACTIVE',
email=user.email,
)
return redirect('dashboard')
E-commerce Order Notifications
Send order confirmation emails and manage post-purchase flows.
# models.py
class Order(models.Model):
customer_email = models.EmailField()
customer_name = models.CharField(max_length=100)
order_number = models.CharField(max_length=20, unique=True)
total_amount = models.DecimalField(max_digits=10, decimal_places=2)
status = models.CharField(max_length=20, default='pending')
created_at = models.DateTimeField(auto_now_add=True)
# signals.py
from django.db.models.signals import post_save
from notifiedby import subscribe_to_flow
@receiver(post_save, sender=Order)
def on_order_created(sender, instance, created, **kwargs):
if created and instance.status == 'confirmed':
# Subscribe to post-purchase flow
subscribe_to_flow(
flow_trigger='ORDER_CONFIRMATION',
email=instance.customer_email,
first_name=instance.customer_name,
flags=['customer', 'recent_purchase'],
)
Transactional Email Integration
Send transactional emails and track delivery status.
# utils.py
from django.core.mail import EmailMessage
from notifiedby import send_email_via_api, get_email_delivery_status
import logging
logger = logging.getLogger(__name__)
def send_password_reset_email(user):
"""Send password reset email and log delivery status"""
reset_url = generate_password_reset_url(user)
message = EmailMessage(
subject='Reset Your Password',
body=f'Click here to reset: {reset_url}',
to=[user.email],
)
email_id = send_email_via_api(message)
if email_id:
logger.info(f"Password reset email sent to {user.email}, ID: {email_id}")
# Could store email_id in database for later status checking
return email_id
else:
logger.error(f"Failed to send password reset email to {user.email}")
return None
def check_email_delivery(email_id):
"""Check delivery status of a sent email"""
try:
status = get_email_delivery_status(email_id)
return status
except Exception as e:
logger.error(f"Failed to check delivery status for {email_id}: {e}")
return None
# Usage in a management command or periodic task
def check_pending_emails():
"""Check status of emails that haven't been delivered yet"""
# This would query your database for pending emails
pending_emails = EmailLog.objects.filter(status='pending')
for email_log in pending_emails:
status = check_email_delivery(email_log.email_id)
if status:
email_log.status = status['status']
email_log.save()
if status['status'] == 'bounced':
# Handle bounced email - maybe mark user as inactive
handle_bounced_email(email_log.user)
Admin Integration
Add NotifiedBy management to Django admin.
# admin.py
from django.contrib import admin
from django.contrib.admin import ModelAdmin
from notifiedby import (
list_flow_subscriptions,
list_recipients,
set_recipient_flags,
clear_recipient_flags,
)
class NotifiedByAdmin(admin.ModelAdmin):
"""Custom admin for NotifiedBy integration"""
def get_urls(self):
from django.urls import path
urls = super().get_urls()
custom_urls = [
path('flows/', self.admin_site.admin_view(self.flow_subscriptions_view)),
path('recipients/', self.admin_site.admin_view(self.recipients_view)),
]
return custom_urls + urls
def flow_subscriptions_view(self, request):
"""View to manage flow subscriptions"""
subscriptions = list_flow_subscriptions()
context = {
'subscriptions': subscriptions['results'],
'total_count': subscriptions['count'],
}
return self.render_to_response(context, 'admin/notifiedby_flows.html')
def recipients_view(self, request):
"""View to manage recipients"""
recipients = list_recipients()
context = {
'recipients': recipients,
}
return self.render_to_response(context, 'admin/notifiedby_recipients.html')
# Register with admin
admin.site.register(SomeModel, NotifiedByAdmin)