Django Email Verification with Celery and Redis
Static Assets Configuration
Create a directory named assets in the project root to hold static files like CSS, JavaScript, and images. Update settings.py to reference this directory:
import os
STATIC_URL = '/static/'
STATICFILES_DIRS = [
os.path.join(BASE_DIR, 'assets'),
]
User Registration View
Implement a class-based view to handle registration. The get method displays the form, while the post method processes input, validates data, creates an inactive user, and triggers an email verification task.
from django.shortcuts import render, redirect
from django.views.generic import View
from django.urls import reverse
from django.contrib.auth import get_user_model
import re
User = get_user_model()
class SignUpView(View):
def get(self, request):
return render(request, 'register.html')
def post(self, request):
username = request.POST.get('username')
password = request.POST.get('password')
email = request.POST.get('email')
agree_terms = request.POST.get('agree_terms')
if not all([username, password, email]):
return render(request, 'register.html', {'error': 'All fields are required.'})
if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', email):
return render(request, 'register.html', {'error': 'Invalid email format.'})
if agree_terms != 'on':
return render(request, 'register.html', {'error': 'You must agree to the terms.'})
try:
user = User.objects.create_user(username=username, email=email, password=password)
user.is_active = False
user.save()
# Generate token and send email
token = user.generate_verification_token()
from core.tasks import send_activation_email
send_activation_email.delay(username, email, token)
return redirect(reverse('auth:login'))
except Exception:
return render(request, 'register.html', {'error': 'Username already exists.'})
Token Generation
Add a method to the User model to generate a cryptographically signed token containing the user's ID. This ensures only the legitimate owner can activate the account.
from itsdangerous import URLSafeTimedSerializer as Serializer
from django.conf import settings
class User(AbstractUser):
def generate_verification_token(self):
serializer = Serializer(settings.SECRET_KEY, 3600)
data = {'user_id': self.id}
return serializer.dumps(data)
Asynchronous Email with Celery
Create a Celery task to send emails asynchronously. Configure the broker to point to a Redis instance. In a file named core/tasks.py:
from celery import Celery
import os
import django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
django.setup()
from django.core.mail import send_mail
from django.conf import settings
app = Celery('core', broker='redis://localhost:6379/1')
@app.task
def send_activation_email(username, recipient, token):
subject = 'Activate Your Account'
message = ''
sender = settings.EMAIL_HOST_USER
html_message = f'''
<h1>Hello, {username}</h1>
<p>Please click the link below to activate your account:</p>
<a href="http://127.0.0.1:8000/auth/activate/{token}">Activate Now</a>
'''
send_mail(subject, message, sender, [recipient], html_message=html_message)
Configure SMTP settings in settings.py:
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.example.com'
EMAIL_PORT = 25
EMAIL_HOST_USER = 'admin@example.com'
EMAIL_HOST_PASSWORD = 'password123'
DEFAULT_FROM_EMAIL = 'MyApp '
Account Activation
Create a view to decode the token and activate the user if the token is valid and unexpired.
from itsdangerous import SignatureExpired, BadSignature
from django.http import HttpResponse
class ActivationView(View):
def get(self, request, token):
serializer = Serializer(settings.SECRET_KEY)
try:
data = serializer.loads(token)
except (SignatureExpired, BadSignature):
return HttpResponse('Activation link is invalid or has expired.')
user_id = data.get('user_id')
try:
user = User.objects.get(id=user_id)
except User.DoesNotExist:
return HttpResponse('User not found.')
user.is_active = True
user.save()
return HttpResponse('Account activated successfully.')
User Authentication
Implement the login flow using Django's built-in authentication system, ensuring the user is active before establishing a session.
from django.contrib.auth import authenticate, login
class SignInView(View):
def get(self, request):
return render(request, 'login.html')
def post(self, request):
username = request.POST.get('username')
password = request.POST.get('password')
user = authenticate(request, username=username, password=password)
if user is not None:
if user.is_active:
login(request, user)
return redirect(reverse('store:index'))
else:
return render(request, 'login.html', {'error': 'Account is not active.'})
else:
return render(request, 'login.html', {'error': 'Invalid username or password.'})
Redis Session Configuration
Configure the project to use Redis as the session backend for better performance. Install django-redis and update settings.
CACHES = {
"default": {
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": "redis://127.0.0.1:6379/2",
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
}
}
}
SESSION_ENGINE = "django.contrib.sessions.backends.cache"
SESSION_CACHE_ALIAS = "default"