I am creating a register method using djangorestframework-simplejwt. I created a custom user in auth using AbstractBaseUser. I get this error when I make a POST request of a jwt oauth access token from the client to my API at /google.
This is the stack trace:
Internal Server Error: /google/
Traceback (most recent call last):
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/django/core/handlers/exception.py", line 55, in inner
response = get_response(request)
^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/django/core/handlers/base.py", line 197, in _get_response
response = wrapped_callback(request, *callback_args, **callback_kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/django/views/decorators/csrf.py", line 56, in wrapper_view
return view_func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/django/views/generic/base.py", line 104, in view
return self.dispatch(request, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/rest_framework/views.py", line 509, in dispatch
response = self.handle_exception(exc)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/rest_framework/views.py", line 469, in handle_exception
self.raise_uncaught_exception(exc)
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/rest_framework/views.py", line 480, in raise_uncaught_exception
raise exc
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/rest_framework/views.py", line 506, in dispatch
response = handler(request, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/auth/views.py", line 186, in post
serializer.is_valid(raise_exception=True)
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/rest_framework/serializers.py", line 227, in is_valid
self._validated_data = self.run_validation(self.initial_data)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/rest_framework/serializers.py", line 426, in run_validation
value = self.to_internal_value(data)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/rest_framework/serializers.py", line 485, in to_internal_value
validated_value = validate_method(validated_value)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/auth/serializers.py", line 178, in validate_auth_token
return register_social_user(
^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/auth/register.py", line 69, in register_social_user
'tokens': user.tokens()
^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/auth/models.py", line 78, in tokens
refresh = RefreshToken.for_user(self)
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/rest_framework_simplejwt/tokens.py", line 288, in for_user
OutstandingToken.objects.create(
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/django/db/models/manager.py", line 87, in manager_method
return getattr(self.get_queryset(), name)(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/django/db/models/query.py", line 656, in create
obj = self.model(**kwargs)
^^^^^^^^^^^^^^^^^^^^
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/django/db/models/base.py", line 543, in __init__
_setattr(self, field.name, rel_obj)
File "/home/pcname/dev-projects/selfey/venv/lib/python3.11/site-packages/django/db/models/fields/related_descriptors.py", line 266, in __set__
raise ValueError(
ValueError: Cannot assign "<User: [email protected]>": "OutstandingToken.user" must be a "User" instance.
This is my views.py:
from rest_framework import status
from rest_framework.response import Response
from rest_framework.generics import GenericAPIView
from .serializers import GoogleSocialAuthSerializer
from django.shortcuts import render
from rest_framework import generics, status, views, permissions
from .serializers import RegisterSerializer, SetNewPasswordSerializer, ResetPasswordEmailRequestSerializer, EmailVerificationSerializer, LoginSerializer, LogoutSerializer
from rest_framework.response import Response
from rest_framework_simplejwt.tokens import RefreshToken
from .models import User
from .utils import Util
from django.contrib.sites.shortcuts import get_current_site
from django.urls import reverse
import jwt
from django.conf import settings
from drf_yasg.utils import swagger_auto_schema
from drf_yasg import openapi
from .renderers import UserRenderer
from django.contrib.auth.tokens import PasswordResetTokenGenerator
from django.utils.encoding import smart_str, force_str, smart_bytes, DjangoUnicodeDecodeError
from django.utils.http import urlsafe_base64_decode, urlsafe_base64_encode
from django.contrib.sites.shortcuts import get_current_site
from django.urls import reverse
from .utils import Util
from django.shortcuts import redirect
from django.http import HttpResponsePermanentRedirect
import os
class CustomRedirect(HttpResponsePermanentRedirect):
allowed_schemes = [os.environ.get('APP_SCHEME'), 'http', 'https']
class RegisterView(generics.GenericAPIView):
serializer_class = RegisterSerializer
renderer_classes = (UserRenderer,)
def post(self, request):
user = request.data
serializer = self.serializer_class(data=user)
serializer.is_valid(raise_exception=True)
serializer.save()
user_data = serializer.data
user = User.objects.get(email=user_data['email'])
current_site = get_current_site(request).domain
relativeLink = reverse('email-verify')
absurl = 'http://'+current_site+relativeLink+"?token="+str(token)
email_body = 'Hi '+user.username + \
' Use the link below to verify your email \n' + absurl
data = {'email_body': email_body, 'to_email': user.email,
'email_subject': 'Verify your email'}
Util.send_email(data)
return Response(user_data, status=status.HTTP_201_CREATED)
class VerifyEmail(views.APIView):
serializer_class = EmailVerificationSerializer
token_param_config = openapi.Parameter(
'token', in_=openapi.IN_QUERY, description='Description', type=openapi.TYPE_STRING)
@swagger_auto_schema(manual_parameters=[token_param_config])
def get(self, request):
token = request.GET.get('token')
try:
payload = jwt.decode(token, settings.SECRET_KEY)
user = User.objects.get(id=payload['user_id'])
if not user.is_verified:
user.is_verified = True
user.save()
return Response({'email': 'Successfully activated'}, status=status.HTTP_200_OK)
except jwt.ExpiredSignatureError as identifier:
return Response({'error': 'Activation Expired'}, status=status.HTTP_400_BAD_REQUEST)
except jwt.exceptions.DecodeError as identifier:
return Response({'error': 'Invalid token'}, status=status.HTTP_400_BAD_REQUEST)
class LoginAPIView(generics.GenericAPIView):
serializer_class = LoginSerializer
def post(self, request):
serializer = self.serializer_class(data=request.data)
serializer.is_valid(raise_exception=True)
return Response(serializer.data, status=status.HTTP_200_OK)
class RequestPasswordResetEmail(generics.GenericAPIView):
serializer_class = ResetPasswordEmailRequestSerializer
def post(self, request):
serializer = self.serializer_class(data=request.data)
email = request.data.get('email', '')
if User.objects.filter(email=email).exists():
user = User.objects.get(email=email)
uidb64 = urlsafe_base64_encode(smart_bytes(user.id))
token = PasswordResetTokenGenerator().make_token(user)
current_site = get_current_site(
request=request).domain
relativeLink = reverse(
'password-reset-confirm', kwargs={'uidb64': uidb64, 'token': token})
redirect_url = request.data.get('redirect_url', '')
absurl = 'http://'+current_site + relativeLink
email_body = 'Hello, \n Use link below to reset your password \n' + \
absurl+"?redirect_url="+redirect_url
data = {'email_body': email_body, 'to_email': user.email,
'email_subject': 'Reset your passsword'}
Util.send_email(data)
return Response({'success': 'We have sent you a link to reset your password'}, status=status.HTTP_200_OK)
class PasswordTokenCheckAPI(generics.GenericAPIView):
serializer_class = SetNewPasswordSerializer
def get(self, request, uidb64, token):
redirect_url = request.GET.get('redirect_url')
try:
id = smart_str(urlsafe_base64_decode(uidb64))
user = User.objects.get(id=id)
if not PasswordResetTokenGenerator().check_token(user, token):
if len(redirect_url) > 3:
return CustomRedirect(redirect_url+'?token_valid=False')
else:
return CustomRedirect(os.environ.get('FRONTEND_URL', '')+'?token_valid=False')
if redirect_url and len(redirect_url) > 3:
return CustomRedirect(redirect_url+'?token_valid=True&message=Credentials Valid&uidb64='+uidb64+'&token='+token)
else:
return CustomRedirect(os.environ.get('FRONTEND_URL', '')+'?token_valid=False')
except DjangoUnicodeDecodeError as identifier:
try:
if not PasswordResetTokenGenerator().check_token(user):
return CustomRedirect(redirect_url+'?token_valid=False')
except UnboundLocalError as e:
return Response({'error': 'Token is not valid, please request a new one'}, status=status.HTTP_400_BAD_REQUEST)
class SetNewPasswordAPIView(generics.GenericAPIView):
serializer_class = SetNewPasswordSerializer
def patch(self, request):
serializer = self.serializer_class(data=request.data)
serializer.is_valid(raise_exception=True)
return Response({'success': True, 'message': 'Password reset success'}, status=status.HTTP_200_OK)
class LogoutAPIView(generics.GenericAPIView):
serializer_class = LogoutSerializer
permission_classes = (permissions.IsAuthenticated,)
def post(self, request):
serializer = self.serializer_class(data=request.data)
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(status=status.HTTP_204_NO_CONTENT)
class GoogleSocialAuthView(GenericAPIView):
serializer_class = GoogleSocialAuthSerializer
def post(self, request):
"""
POST with "auth_token"
Send an idtoken as from google to get user information
"""
serializer = self.serializer_class(data=request.data)
serializer.is_valid(raise_exception=True)
data = ((serializer.validated_data)['auth_token'])
return Response(data, status=status.HTTP_200_OK)
serializers.py
from django.db import models
# Create your models here.
from django.contrib.auth.models import (
User, AbstractBaseUser, BaseUserManager, PermissionsMixin, Group, Permission)
from django.db import models
from rest_framework_simplejwt.tokens import RefreshToken
class UserManager(BaseUserManager):
def create_user(self, username, email, password=None):
if username is None:
raise TypeError('Users should have a username')
if email is None:
raise TypeError('Users should have a Email')
user = self.model(username=username, email=self.normalize_email(email))
user.set_password(password)
user.save()
return user
def create_superuser(self, username, email, password=None):
if password is None:
raise TypeError('Password should not be none')
user = self.create_user(username, email, password)
user.is_superuser = True
user.is_staff = True
user.save()
return user
AUTH_PROVIDERS = {'google': 'google', 'email': 'email'}
class User(AbstractBaseUser, PermissionsMixin):
username = models.CharField(max_length=255, unique=True, db_index=True)
email = models.EmailField(max_length=255, unique=True, db_index=True)
is_verified = models.BooleanField(default=False)
is_active = models.BooleanField(default=True)
is_staff = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
auth_provider = models.CharField(
max_length=255, blank=False,
null=False, default=AUTH_PROVIDERS.get('email'))
groups = models.ManyToManyField(
Group,
verbose_name='groups',
blank=True,
help_text=
'The groups this user belongs to. A user will get all permissions '
'granted to each of their groups.'
,
related_name="my_auth_user_set",
related_query_name="user",
)
user_permissions = models.ManyToManyField(
Permission,
verbose_name='user permissions',
blank=True,
help_text='Specific permissions for this user.',
related_name="my_auth_user_set",
related_query_name="user",
)
USERNAME_FIELD = 'email'
REQUIRED_FIELDS = ['username']
objects = UserManager()
def __str__(self):
return self.email
def tokens(self):
refresh = RefreshToken.for_user(self)
return {
'refresh': str(refresh),
'access': str(refresh.access_token)
}
My custom user model: models.py
from django.db import models
# Create your models here.
from django.contrib.auth.models import (
AbstractBaseUser, BaseUserManager, PermissionsMixin, Group, Permission)
from django.db import models
from rest_framework_simplejwt.tokens import RefreshToken
class UserManager(BaseUserManager):
def create_user(self, username, email, password=None):
if username is None:
raise TypeError('Users should have a username')
if email is None:
raise TypeError('Users should have a Email')
user = self.model(username=username, email=self.normalize_email(email))
user.set_password(password)
user.save()
return user
def create_superuser(self, username, email, password=None):
if password is None:
raise TypeError('Password should not be none')
user = self.create_user(username, email, password)
user.is_superuser = True
user.is_staff = True
user.save()
return user
AUTH_PROVIDERS = {'google': 'google', 'email': 'email'}
class User(AbstractBaseUser, PermissionsMixin):
username = models.CharField(max_length=255, unique=True, db_index=True)
email = models.EmailField(max_length=255, unique=True, db_index=True)
is_verified = models.BooleanField(default=False)
is_active = models.BooleanField(default=True)
is_staff = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
auth_provider = models.CharField(
max_length=255, blank=False,
null=False, default=AUTH_PROVIDERS.get('email'))
groups = models.ManyToManyField(
Group,
verbose_name='groups',
blank=True,
help_text=
'The groups this user belongs to. A user will get all permissions '
'granted to each of their groups.'
,
related_name="my_auth_user_set",
related_query_name="user",
)
user_permissions = models.ManyToManyField(
Permission,
verbose_name='user permissions',
blank=True,
help_text='Specific permissions for this user.',
related_name="my_auth_user_set",
related_query_name="user",
)
USERNAME_FIELD = 'email'
REQUIRED_FIELDS = ['username']
objects = UserManager()
def __str__(self):
return self.email
def tokens(self):
print('is an instance of User', isinstance(self, User)) # returns True
refresh = RefreshToken.for_user(self)
return {
'refresh': str(refresh),
'access': str(refresh.access_token)
}
This is register.py
from .email_backend import EmailBackend
from .models import User
import os
import random
from rest_framework.exceptions import AuthenticationFailed
from dotenv import load_dotenv
load_dotenv('.env')
backend = EmailBackend()
def generate_username(name):
username = "".join(name.split(' ')).lower()
if not User.objects.filter(email=username).exists():
return username
else:
random_username = username + str(random.randint(0, 1000))
return generate_username(random_username)
def register_social_user(provider, user_id, email, name):
filtered_user_by_email = User.objects.filter(email=email)
if filtered_user_by_email.exists():
if provider == filtered_user_by_email[0].auth_provider:
print('email', backend.authenticate(email=email))
registered_user = backend.authenticate(email=email, password=os.getenv('SOCIAL_SECRET'))
print('user is already registered', registered_user)
if registered_user is not None:
return {
'username': registered_user.username,
'email': registered_user.email,
'tokens': registered_user.tokens()}
else:
user = {
'username': generate_username(name), 'email': email}
user = User.objects.create_user(**user)
user.is_verified = True
user.auth_provider = provider
user.save()
new_user = backend.authenticate(email=email)
return {
'email': new_user.email,
'username': new_user.username,
'tokens': new_user.tokens()
}
else:
raise AuthenticationFailed(
detail='Please continue your login using ' + filtered_user_by_email[0].auth_provider)
else:
user = {
'username': generate_username(name), 'email': email,
'password': os.getenv('SOCIAL_SECRET')}
user = User.objects.create_user(**user)
user.is_verified = True
user.auth_provider = provider
user.save()
return {
'email': user.email,
'username': user.username,
'tokens': user.tokens()
}
Custom authentication backend, email_backend.py:
from .models import User
from django.contrib.auth.backends import ModelBackend
class EmailBackend(ModelBackend):
def authenticate(self, request=None, username=None, email=None, password=None, **kwargs):
print("email backend", User.objects.get(email=email).check_password(password))
try:
user = User.objects.get(email=email)
except User.DoesNotExist:
return None
else:
if user.check_password(password):
return user
return None
requirements.txt
Django
djangorestframework
django-rest-auth
django-tailwind
django-tailwind[reload]
django-cors-headers
djangorestframework-simplejwt
drf-yasg
google-api-core
google-api-python-client
google-auth
google-auth-httplib2
googleapis-common-protos
gunicorn
Pillow
python-dotenv
urls.py
from django.contrib import admin
from django.urls import include, path
from django.conf import settings
from rest_framework.routers import DefaultRouter
from posts.views import PostViewSet, CommentViewSet
from users.views import UserViewSet
from auth.views import GoogleSocialAuthView
router = DefaultRouter()
router.register(r'users', UserViewSet)
router.register(r'posts', PostViewSet)
router.register(r'comments', CommentViewSet)
urlpatterns = [
path('admin/', admin.site.urls),
path('google/', GoogleSocialAuthView.as_view()),
path('', include(router.urls)),
path('auth/', include('auth.urls')),
]
if settings.DEBUG:
from django.conf.urls.static import static
urlpatterns += static (settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)
urlpatterns += static (settings.STATIC_URL, document_root=settings.STATIC_ROOT)