Django

pip install unsent

# Unsent Email Configuration
UNSENT_API_KEY = env('UNSENT_API_KEY')
UNSENT_FROM_EMAIL = env('UNSENT_FROM_EMAIL', default='noreply@yourdomain.com')
UNSENT_BASE_URL = env('UNSENT_BASE_URL', default=None)  # Use for local development

# Optional: Configure default email backend
DEFAULT_FROM_EMAIL = UNSENT_FROM_EMAIL

UNSENT_API_KEY=un_...
UNSENT_FROM_EMAIL=noreply@yourdomain.com

from django.conf import settings
from unsent import unsent, types
from typing import Dict, Optional, Tuple
import logging

logger = logging.getLogger(__name__)

class UnsentClient:
    """Singleton Unsent client wrapper."""
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.client = unsent(
                settings.UNSENT_API_KEY,
                base_url=getattr(settings, 'UNSENT_BASE_URL', None),
                raise_on_error=False
            )
        return cls._instance
    
    def send_email(self, payload: types.EmailCreate) -> Tuple[Optional[Dict], Optional[Dict]]:
        """Send an email and log the result."""
        data, error = self.client.emails.send(payload)
        
        if error:
            logger.error(f"Failed to send email to {payload.get('to')}: {error}")
        else:
            logger.info(f"Email sent successfully: {data.get('id')}")
        
        return data, error

# Global instance
unsent_client = UnsentClient()

from django.shortcuts import render, redirect
from django.contrib.auth import login
from django.contrib import messages
from .forms import SignUpForm
from utils.email import unsent_client
from unsent import types

def signup(request):
    if request.method == 'POST':
        form = SignUpForm(request.POST)
        if form.is_valid():
            user = form.save()
            login(request, user)
            
            # Send welcome email
            payload: types.EmailCreate = {
                "to": user.email,
                "from": settings.UNSENT_FROM_EMAIL,
                "subject": "Welcome to Our Platform!",
                "html": f"""
                    <h1>Welcome, {user.get_full_name() or user.username}!</h1>
                    <p>Thanks for joining us. We're excited to have you on board.</p>
                """,
                "text": f"Welcome, {user.get_full_name() or user.username}! Thanks for joining."
            }
            
            data, error = unsent_client.send_email(payload)
            
            if not error:
                messages.success(request, 'Account created! Check your email.')
            
            return redirect('home')
    else:
        form = SignUpForm()
    
    return render(request, 'signup.html', {'form': form})

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth import get_user_model
from services.email_service import EmailService
import logging

logger = logging.getLogger(__name__)
User = get_user_model()

@receiver(post_save, sender=User)
def send_welcome_email_on_user_creation(sender, instance, created, **kwargs):
    """Send welcome email when new user is created."""
    if created and instance.email:
        try:
            EmailService.send_welcome_email(instance)
        except Exception as e:
            logger.error(f"Failed to send welcome email for user {instance.id}: {e}")

from django.apps import AppConfig

class MyAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'myapp'
    
    def ready(self):
        import myapp.signals  # Import signals

from django.core.management.base import BaseCommand
from django.contrib.auth import get_user_model
from unsent import types
from utils.email import unsent_client
from django.conf import settings

User = get_user_model()

class Command(BaseCommand):
    help = 'Send newsletter to all active users'
    
    def add_arguments(self, parser):
        parser.add_argument('--subject', type=str, required=True)
        parser.add_argument('--template', type=str, required=True)
    
    def handle(self, *args, **options):
        subject = options['subject']
        template = options['template']
        
        # Get all active users with email
        users = User.objects.filter(is_active=True, email__isnull=False)
        
        self.stdout.write(f'Sending newsletter to {users.count()} users...')
        
        # Prepare batch
        batch_items = []
        for user in users:
            html = self.render_template(template, user)
            
            batch_items.append({
                "to": user.email,
                "from": settings.UNSENT_FROM_EMAIL,
                "subject": subject,
                "html": html
            })
            
            # Send in batches of 100
            if len(batch_items) >= 100:
                self.send_batch(batch_items)
                batch_items = []
        
        # Send remaining
        if batch_items:
            self.send_batch(batch_items)
        
        self.stdout.write(self.style.SUCCESS('Newsletter sent successfully!'))
    
    def render_template(self, template_name, user):
        from django.template.loader import render_to_string
        return render_to_string(template_name, {'user': user})
    
    def send_batch(self, batch_items):
        data, error = unsent_client.client.emails.batch(batch_items)
        if error:
            self.stdout.write(self.style.ERROR(f'Batch failed: {error}'))

python manage.py send_newsletter --subject "Weekly Newsletter" --template "emails/newsletter.html"

from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from django.http import JsonResponse
import json
import logging

logger = logging.getLogger(__name__)

@csrf_exempt
@require_POST
def unsent_webhook(request):
    """Handle Unsent webhook events."""
    try:
        event = json.loads(request.body)
        event_type = event.get('type')
        data = event.get('data', {})
        
        # Route to appropriate handler
        handlers = {
            'email.delivered': handle_email_delivered,
            'email.bounced': handle_email_bounced,
            'email.opened': handle_email_opened,
            'email.clicked': handle_email_clicked,
            'email.complained': handle_email_complained
        }
        
        handler = handlers.get(event_type)
        if handler:
            handler(data)
        else:
            logger.info(f"Unhandled webhook event: {event_type}")
        
        return JsonResponse({"status": "received"})
    
    except Exception as e:
        logger.error(f"Webhook error: {e}")
        return JsonResponse({"error": str(e)}, status=500)

def handle_email_delivered(data):
    """Handle email delivered event."""
    email_id = data.get('emailId')
    recipient = data.get('recipient')
    logger.info(f"Email {email_id} delivered to {recipient}")
    # Update EmailLog model, track analytics

def handle_email_bounced(data):
    """Handle email bounced event."""
    from django.contrib.auth import get_user_model
    User = get_user_model()
    
    recipient = data.get('recipient')
    bounce_type = data.get('bounceType')
    
    logger.warning(f"Email bounced: {recipient} ({bounce_type})")
    
    # Mark user email as bounced
    try:
        user = User.objects.get(email=recipient)
        user.email_bounced = True
        user.save()
    except User.DoesNotExist:
        pass

def handle_email_opened(data):
    """Handle email opened event."""
    email_id = data.get('emailId')
    logger.info(f"Email {email_id} opened")
    # Track open rate in analytics

def handle_email_clicked(data):
    """Handle link clicked event."""
    email_id = data.get('emailId')
    link = data.get('link')
    logger.info(f"Link clicked in {email_id}: {link}")
    # Track click rate

def handle_email_complained(data):
    """Handle spam complaint."""
    from django.contrib.auth import get_user_model
    User = get_user_model()
    
    recipient = data.get('recipient')
    logger.warning(f"Spam complaint from {recipient}")
    
    # Automatically unsubscribe
    try:
        user = User.objects.get(email=recipient)
        user.email_subscribed = False
        user.save()
    except User.DoesNotExist:
        pass

from django.urls import path
from . import views

urlpatterns = [
    path('webhooks/unsent/', views.unsent_webhook, name='unsent_webhook'),
]

from django.test import TestCase
from django.contrib.auth import get_user_model
from unittest.mock import patch, Mock
from services.email_service import EmailService

User = get_user_model()

class EmailServiceTestCase(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username='testuser',
            email='test@example.com',
            first_name='Test',
            last_name='User'
        )
    
    @patch('utils.email.unsent_client.client')
    def test_send_welcome_email_success(self, mock_client):
        """Test successful welcome email send."""
        mock_client.emails.send.return_value = ({"id": "email_123"}, None)
        
        data, error = EmailService.send_welcome_email(self.user)
        
        self.assertIsNone(error)
        self.assertEqual(data['id'], 'email_123')
        mock_client.emails.send.assert_called_once()
    
    @patch('utils.email.unsent_client.client')
    def test_send_welcome_email_failure(self, mock_client):
        """Test failed welcome email send."""
        mock_client.emails.send.return_value = (None, {"message": "API Error"})
        
        data, error = EmailService.send_welcome_email(self.user)
        
        self.assertIsNone(data)
        self.assertIsNotNone(error)
        self.assertEqual(error['message'], 'API Error')

class WebhookTestCase(TestCase):
    def test_email_delivered_webhook(self):
        """Test email delivered webhook processing."""
        payload = {
            "type": "email.delivered",
            "data": {
                "emailId": "email_123",
                "recipient": "test@example.com"
            }
        }
        
        response = self.client.post(
            '/webhooks/unsent/',
            data=payload,
            content_type='application/json'
        )
        
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()['status'], 'received')

python manage.py test

from django.db import models
from django.contrib.auth import get_user_model

User = get_user_model()

class EmailLog(models.Model):
    STATUS_CHOICES = [
        ('pending', 'Pending'),
        ('sent', 'Sent'),
        ('delivered', 'Delivered'),
        ('bounced', 'Bounced'),
        ('complained', 'Complained'),
    ]
    
    user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True)
    unsent_id = models.CharField(max_length=100, unique=True)
    recipient = models.EmailField()
    subject = models.CharField(max_length=200)
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')
    sent_at = models.DateTimeField(auto_now_add=True)
    delivered_at = models.DateTimeField(null=True, blank=True)
    opened_at = models.DateTimeField(null=True, blank=True)
    opened_count = models.IntegerField(default=0)
    clicked_count = models.IntegerField(default=0)
    
    class Meta:
        ordering = ['-sent_at']
        indexes = [
            models.Index(fields=['unsent_id']),
            models.Index(fields=['recipient']),
            models.Index(fields=['status']),
        ]
    
    def __str__(self):
        return f"{self.subject} to {self.recipient}"

import threading

_thread_locals = threading.local()

def get_current_request():
    return getattr(_thread_locals, 'request', None)

class RequestMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        _thread_locals.request = request
        response = self.get_response(request)
        _thread_locals.request = None
        return response

MIDDLEWARE = [
    'myapp.middleware.RequestMiddleware',
    # ...other middleware
]

from django.apps import AppConfig
from django.core.exceptions import ImproperlyConfigured

class MyAppConfig(AppConfig):
    name = 'myapp'
    
    def ready(self):
        from django.conf import settings
        
        if not hasattr(settings, 'UNSENT_API_KEY'):
            raise ImproperlyConfigured('UNSENT_API_KEY is required')
        
        if not settings.UNSENT_API_KEY:
            raise ImproperlyConfigured('UNSENT_API_KEY cannot be empty')