pip install django-celery-ses boto3
python
INSTALLED_APPS = [
...
'django_celery_ses',
]
EMAIL_BACKEND = 'django_celery_ses.backends.CeleryEmailBackend'
AWS_ACCESS_KEY_ID = 'Your-Access-Key-ID'
AWS_SECRET_ACCESS_KEY = 'Your-Secret-Access-Key'
EMAIL_FROM_ADDRESS = 'your-email@example.com'
python
BROKER_URL = 'redis://localhost:6379/0'
CELERY_TASK_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers.DatabaseScheduler'
python
from django.core.mail import send_mail
from celery import shared_task
@shared_task
def send_email_task(subject, message, recipient_list):
send_mail(subject, message, 'your-email@example.com', recipient_list)
python
from .tasks import send_email_task
def send_email_view(request):
send_email_task.delay('Hello', 'This is a test email', ['recipient@example.com'])
return HttpResponse('Email sent successfully!')
