当前位置：首页 > article >正文

每天40分玩转Django：Django Email

article 2024/12/29 0:16:51

Django Email

一、知识要点总览表

类别	知识点
基础配置	SMTP设置、Email配置项
发送方式	同步发送、异步发送
邮件类型	纯文本、HTML邮件、带附件邮件
异步任务	Celery集成、任务队列
高级特性	邮件模板、批量发送

二、Email基础配置

1. 配置settings.py

# settings.py
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.gmail.com'  # 邮件服务器
EMAIL_PORT = 587  # 端口号
EMAIL_USE_TLS = True  # 使用TLS加密
EMAIL_HOST_USER = 'your-email@gmail.com'  # 发件人邮箱
EMAIL_HOST_PASSWORD = 'your-password'  # 邮箱密码或应用专用密码
DEFAULT_FROM_EMAIL = 'your-email@gmail.com'  # 默认发件人

2. 测试邮件配置

# test_email.py
from django.core.mail import send_mail
from django.conf import settings

def test_email_config():
    subject = '测试邮件'
    message = '这是一封测试邮件'
    from_email = settings.DEFAULT_FROM_EMAIL
    recipient_list = ['recipient@example.com']
    
    try:
        send_mail(
            subject,
            message,
            from_email,
            recipient_list,
            fail_silently=False,
        )
        return "邮件发送成功！"
    except Exception as e:
        return f"邮件发送失败：{str(e)}"

三、发送各类型邮件

1. 发送HTML邮件

# views.py
from django.core.mail import EmailMultiAlternatives
from django.template.loader import render_to_string
from django.utils.html import strip_tags

def send_html_email(request):
    subject = '欢迎订阅我们的服务'
    from_email = settings.DEFAULT_FROM_EMAIL
    to_email = 'subscriber@example.com'

    # 渲染HTML内容
    html_content = render_to_string('email_template.html', {
        'username': 'John Doe',
        'activation_link': 'http://example.com/activate'
    })
    text_content = strip_tags(html_content)  # 生成纯文本内容

    email = EmailMultiAlternatives(
        subject,
        text_content,
        from_email,
        [to_email]
    )
    email.attach_alternative(html_content, "text/html")
    email.send()

<!-- templates/email_template.html -->
<!DOCTYPE html>
<html>
<head>
    <title>欢迎订阅</title>
</head>
<body>
    <h1>您好，{{ username }}!</h1>
    <p>感谢您订阅我们的服务。请点击以下链接激活您的账户：</p>
    <a href="{{ activation_link }}">激活账户</a>
</body>
</html>

2. 发送带附件的邮件

# views.py
from django.core.mail import EmailMessage
import os

def send_attachment_email(request):
    subject = '月度报告'
    body = '请查收附件中的月度报告。'
    from_email = settings.DEFAULT_FROM_EMAIL
    to_email = ['manager@example.com']
    
    email = EmailMessage(
        subject,
        body,
        from_email,
        to_email
    )
    
    # 添加附件
    file_path = os.path.join(settings.MEDIA_ROOT, 'reports/monthly_report.pdf')
    with open(file_path, 'rb') as f:
        email.attach('monthly_report.pdf', f.read(), 'application/pdf')
    
    email.send()

四、异步任务处理

1. Celery配置

# celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

# 设置默认Django settings模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')

app = Celery('your_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

2. 创建异步邮件任务

# tasks.py
from celery import shared_task
from django.core.mail import send_mail
from django.conf import settings

@shared_task
def send_async_email(subject, message, recipient_list):
    """
    异步发送邮件的任务
    """
    return send_mail(
        subject,
        message,
        settings.DEFAULT_FROM_EMAIL,
        recipient_list,
        fail_silently=False,
    )

@shared_task
def send_bulk_emails(email_data_list):
    """
    批量发送邮件的任务
    """
    results = []
    for email_data in email_data_list:
        try:
            send_mail(
                email_data['subject'],
                email_data['message'],
                settings.DEFAULT_FROM_EMAIL,
                email_data['recipients'],
                fail_silently=False,
            )
            results.append({
                'status': 'success',
                'recipients': email_data['recipients']
            })
        except Exception as e:
            results.append({
                'status': 'failed',
                'recipients': email_data['recipients'],
                'error': str(e)
            })
    return results

3. 在视图中使用异步任务

# views.py
from .tasks import send_async_email, send_bulk_emails
from django.http import JsonResponse

def send_newsletter(request):
    # 获取订阅用户列表
    subscribers = ['user1@example.com', 'user2@example.com', 'user3@example.com']
    
    # 创建异步任务
    task = send_async_email.delay(
        '最新新闻订阅',
        '这是本周的新闻摘要...',
        subscribers
    )
    
    return JsonResponse({
        'status': 'success',
        'task_id': task.id,
        'message': '邮件发送任务已创建'
    })

def send_bulk_newsletter(request):
    # 批量邮件数据
    email_data_list = [
        {
            'subject': '个性化新闻推送',
            'message': '基于您的兴趣推荐...',
            'recipients': ['user1@example.com']
        },
        {
            'subject': '个性化新闻推送',
            'message': '您可能感兴趣的内容...',
            'recipients': ['user2@example.com']
        }
    ]
    
    # 创建批量发送任务
    task = send_bulk_emails.delay(email_data_list)
    
    return JsonResponse({
        'status': 'success',
        'task_id': task.id,
        'message': '批量邮件发送任务已创建'
    })

五、邮件发送流程图

在这里插入图片描述

六、监控和错误处理

1. 邮件发送状态监控

# utils.py
from django.core.mail import get_connection
from django.core.mail.message import EmailMessage
from django.conf import settings
import logging

logger = logging.getLogger(__name__)

class EmailSender:
    def __init__(self):
        self.connection = get_connection()
    
    def send_email_with_monitoring(self, subject, message, recipient_list):
        """
        带监控的邮件发送函数
        """
        try:
            email = EmailMessage(
                subject,
                message,
                settings.DEFAULT_FROM_EMAIL,
                recipient_list,
                connection=self.connection
            )
            
            # 记录发送开始
            logger.info(f"开始发送邮件到 {recipient_list}")
            
            # 发送邮件并记录结果
            result = email.send()
            
            if result:
                logger.info(f"邮件成功发送到 {recipient_list}")
                return True
            else:
                logger.error(f"邮件发送失败: {recipient_list}")
                return False
                
        except Exception as e:
            logger.error(f"邮件发送异常: {str(e)}")
            raise

2. 重试机制

# tasks.py
from celery import shared_task
from celery.exceptions import MaxRetriesExceededError

@shared_task(bind=True, max_retries=3)
def send_email_with_retry(self, subject, message, recipient_list):
    """
    带重试机制的异步邮件发送任务
    """
    try:
        send_mail(
            subject,
            message,
            settings.DEFAULT_FROM_EMAIL,
            recipient_list,
            fail_silently=False,
        )
    except Exception as exc:
        try:
            # 重试任务，延迟5分钟
            self.retry(exc=exc, countdown=300)
        except MaxRetriesExceededError:
            # 超过最大重试次数，记录错误
            logger.error(f"邮件发送失败，超过最大重试次数: {recipient_list}")
            raise