# myproject/adapters.py
from allauth.account.adapter import DefaultAccountAdapter
from allauth.socialaccount.adapter import DefaultSocialAccountAdapter
from django.core.exceptions import ValidationError
from django.shortcuts import redirect
from allauth.exceptions import ImmediateHttpResponse
import logging
logger = logging.getLogger(__name__)
class MyAccountAdapter(DefaultAccountAdapter):
def is_open_for_signup(self, request):
"""Only allow signup with valid invitation code."""
invitation_code = request.session.get('invitation_code')
if not invitation_code:
response = redirect('/invitation-required/')
raise ImmediateHttpResponse(response)
return True
def clean_email(self, email):
"""Only allow company email addresses."""
if not email.endswith('@mycompany.com'):
raise ValidationError(
"Please use your company email address."
)
return super().clean_email(email)
def save_user(self, request, user, form, commit=True):
"""Add custom fields from signup form."""
user = super().save_user(request, user, form, commit=False)
user.department = form.cleaned_data.get('department')
user.employee_id = form.cleaned_data.get('employee_id')
if commit:
user.save()
return user
def get_login_redirect_url(self, request):
"""Redirect based on user role."""
user = request.user
if user.is_staff:
return "/admin/dashboard/"
elif user.department == 'sales':
return "/sales/dashboard/"
return "/dashboard/"
def post_login(self, request, user, **kwargs):
"""Log login and update last login IP."""
logger.info(f"User {user.username} logged in from {self.get_client_ip(request)}")
user.last_login_ip = self.get_client_ip(request)
user.save(update_fields=['last_login_ip'])
return super().post_login(request, user, **kwargs)
def send_mail(self, template_prefix, email, context):
"""Log all emails sent."""
logger.info(f"Sending {template_prefix} email to {email}")
super().send_mail(template_prefix, email, context)
class MySocialAccountAdapter(DefaultSocialAccountAdapter):
def pre_social_login(self, request, sociallogin):
"""Link social accounts to existing users by email."""
if sociallogin.is_existing:
return
if sociallogin.email_addresses:
email = sociallogin.email_addresses[0].email
try:
from django.contrib.auth import get_user_model
User = get_user_model()
user = User.objects.get(email=email)
sociallogin.connect(request, user)
logger.info(
f"Linked {sociallogin.account.provider} account to "
f"existing user {user.username}"
)
except User.DoesNotExist:
pass
def is_open_for_signup(self, request, sociallogin):
"""Only allow specific social providers."""
allowed_providers = ['google', 'github', 'microsoft']
if sociallogin.account.provider not in allowed_providers:
return False
return super().is_open_for_signup(request, sociallogin)
def populate_user(self, request, sociallogin, data):
"""Extract additional data from social provider."""
user = super().populate_user(request, sociallogin, data)
# Mark social users as pre-verified
user.is_verified = True
return user
