Hi everyone,
I am doing a project that consists in creating an app, and I am using Django Rest Framework to create my APIs. I am getting this error, and just can’t find the solutions.
I am doing everything using django_cassandra_engine, because my database is in Cassandra.
I am creating the User Authentication.
models.py
from django_cassandra_engine.models import DjangoCassandraModel
from django.contrib.auth.models import BaseUserManager
from cassandra.cqlengine import columns
import uuid
import datetime
import bcrypt
class Conversations(DjangoCassandraModel):
id_conversation = columns.UUID(primary_key=True, default=uuid.uuid4)
id = columns.UUID(required=True) # Refers to the users ID
message_ids = columns.List(columns.UUID)
title = columns.Text()
created_at = columns.DateTime(default=datetime.datetime.utcnow)
class Meta:
db_table = "conversations"
class Message(DjangoCassandraModel):
id_message = columns.UUID(primary_key=True, default=uuid.uuid4)
id_conversation = columns.UUID(required=True)
sender = columns.Text(required=True)
content = columns.Text(required=True)
created_at = columns.DateTime(default=datetime.datetime.utcnow)
class Meta:
db_table = "messages"
class Users(DjangoCassandraModel):
id = columns.UUID(primary_key=True, default=uuid.uuid4) # User ID
email = columns.Text(required=True, index=True)
username = columns.Text(required=True, index=True)
password = columns.Text(required=True)
is_active = columns.Boolean(default=True)
is_staff = columns.Boolean(default=False)
is_superuser = columns.Boolean(default=False)
USERNAME_FIELD = 'username'
REQUIRED_FIELDS = ['email ', 'password']
class Meta:
db_table = 'users'
def set_password(self, password):
self.password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
def check_password(self, password):
return bcrypt.checkpw(password.encode('utf-8'), self.password.encode('utf-8'))
def get_full_name(self):
return self.username
def get_short_name(self):
return self.username
@property
def is_anonymous(self):
return False
@property
def is_authenticated(self):
return True
def __str__(self):
return self.username
views.py
# Importing from Django
from rest_framework.views import APIView
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.response import Response
from rest_framework import status
from rest_framework_simplejwt.tokens import RefreshToken
from django.conf import settings
# Importing from local folders
from .serializers import RegisterSerializer, UserSerializer, LoginSerializer
from .models import Users
from assistant.response import get_ai_response
# importing from Python
import bcrypt
import jwt
import datetime
import uuid
class ChatView(APIView):
"""Public Endpoint for sending messages to the chatbot."""
permission_classes = [AllowAny] # Without Authentication
def post(self, request):
user_message = request.data.get("message", "")
ai_reponse = get_ai_response(user_input=user_message)
return Response({"response": ai_reponse})
class ProtectedView(APIView):
"""Protected Route for testing authentication"""
permission_classes = [IsAuthenticated]
def get(self, request):
return Response({"message": "You have access!"}, status=status.HTTP_200_OK)
class RegisterView(APIView):
"""View to register news users."""
permission_classes = [AllowAny]
def post(self, request):
serializer = RegisterSerializer(data=request.data)
if serializer.is_valid():
hashed_password = bcrypt.hashpw(serializer.validated_data['password'].encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
user = Users.create(
id = uuid.uuid4(),
email = serializer.validated_data['email'].lower(),
username = serializer.validated_data['username'],
password = hashed_password
)
return Response({"message": "User registered successfully!", "user": UserSerializer(user).data}, status=status.HTTP_201_CREATED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
class LoginView(APIView):
"""View to login and obtain JWT token."""
permission_classes = [AllowAny]
def post(self, request):
serializer = LoginSerializer(data=request.data)
if not serializer.is_valid():
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
email = serializer.validated_data["email"]
password = serializer.validated_data["password"]
user = Users.objects.filter(email=email).first()
if not user or not user.check_password(password):
return Response({"error": "Invalid credentials!"}, status=status.HTTP_401_UNAUTHORIZED)
refresh = RefreshToken.for_user(user)
refresh['user_id'] = str(user.id)
access_token = str(refresh.access_token)
refresh_token = str(refresh)
return Response({
"message": "Login successful!",
"access_token": access_token,
"refresh_token": refresh_token
}, status=status.HTTP_200_OK)
serializers.py
from rest_framework import serializers
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
import bcrypt
from .models import Users, Message
import uuid
class MessageSerializer(serializers.ModelSerializer):
class Meta:
model = Message
fields = ['id_message', 'id_conversation', 'sender', 'content', 'created_at']
class UserSerializer(serializers.ModelSerializer):
id = serializers.UUIDField(format='hex_verbose')
email = serializers.CharField()
username = serializers.CharField()
is_active = serializers.BooleanField()
is_staff = serializers.BooleanField()
is_superuser = serializers.BooleanField()
class Meta:
model = Users
fields = ['id', 'email', 'username', 'is_active', 'is_staff', 'is_superuser']
class RegisterSerializer(serializers.Serializer):
email = serializers.EmailField()
username = serializers.CharField(max_length=255)
password = serializers.CharField(write_only=True)
def validate_email(self, value):
if Users.objects.filter(email=value).count() > 0:
raise serializers.ValidationError("This email is already in use.")
return value
def validate_username(self, value):
if Users.objects.filter(username=value).count() > 0:
raise serializers.ValidationError("This username is already in use.")
return value
def create(self, validated_data):
"""Create a user and encrypt the password before saving"""
hashed_password = bcrypt.hashpw(validated_data['password'].encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
user = Users.create(
id=uuid.uuid4(),
email=validated_data['email'].lower(),
username=validated_data['username'],
password=hashed_password
)
return user
class LoginSerializer(serializers.Serializer):
email = serializers.EmailField()
password = serializers.CharField(write_only=True)
def validate(self, data):
email = data.get('email')
password = data.get('password')
user = Users.objects.filter(email=email).first()
if not user:
raise ValueError("Invalid email or password!")
if not user.password or not bcrypt.checkpw(password.encode('utf-8'), user.password.encode('utf-8')):
raise serializers.ValidationError("Invalid email or password!")
data['user'] = user
return data
tests.py
from django_cassandra_engine.test import TestCase
from rest_framework.test import APIClient
from rest_framework import status
import uuid
class UserAuthTests(TestCase):
def setUp(self):
"""Initial Test Configuration."""
self.client_api = APIClient()
self.register_url = "/api/register/"
self.login_url = "/api/login/"
self.token_refresh_url = "/api/token/refresh/"
self.protected_url = "/api/protected/"
self.user_data = {
"email": "test@example.com",
"username": "TestUser",
"password": "testpass123"
}
def test_register_user(self):
"""Test to check if a user can register."""
response = self.client_api.post(self.register_url, self.user_data, format="json")
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertIn("user", response.data)
def test_login_user(self):
"""Login test and obtain JWT."""
# First, register a user
self.client_api.post(self.register_url, self.user_data, format='json')
# Now, try to Login
response = self.client_api.post(self.login_url, {
"email": self.user_data['email'],
"password": self.user_data['password']
}, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("access_token", response.data)
self.assertIn("refresh_token", response.data)
self.access_token = response.data['access_token']
self.refresh_token = response.data['refresh_token']
def test_access_protected_route(self):
"""Test to verify if a protected route demand authentication."""
self.test_login_user()
print(f"DEBUG: Access Token = {self.access_token}")
if not self.access_token:
self.fail("Access token was not generated correctly!")
self.client_api.credentials(HTTP_AUTHORIZATION=f"Bearer {self.access_token}")
response = self.client_api.get(self.protected_url) # Try to access a protected route
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_refresh_token(self):
"""Test to verify if the refresh token works."""
# Login to get the RefreshToken
self.test_login_user()
response = self.client_api.post(self.token_refresh_url, {
"refresh": str(self.refresh_token)
}, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("access", response.data)
authentication.py
from rest_framework_simplejwt.authentication import JWTAuthentication
from .models import Users
import uuid
class CassandraJWTAuthentication(JWTAuthentication):
"""Custom Authentication for SimpleJWT in Cassandra."""
def get_user(self, validated_token):
user_id = validated_token.get("user_id")
if not user_id:
return None
user_id = uuid.UUID(user_id)
user = Users.objects.filter(id=user_id).first()
if user:
return user
return None
settings.py
SIMPLE_JWT = {
"ACCESS_TOKEN_LIFETIME": timedelta(minutes=5),
"REFRESH_TOKEN_LIFETIME": timedelta(days=7),
"ROTATE_REFRESH_TOKENS": True,
"BLACKLIST_AFTER_ROTATION": True,
"AUTH_HEADER_TYPES": ("Bearer",),
"USER_ID_FIELD": "id", # The same value as the Users column
"USER_ID_CLAIM": "user_id",
"AUTH_TOKEN_CLASSES": ("rest_framework_simplejwt.tokens.AccessToken",),
"ALGORITHM": "HS256",
}
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'ptlaws_api.authentication.CassandraJWTAuthentication',
],
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticated',
],
}
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.messages',
'django.contrib.staticfiles',
# 3rd party apps
'corsheaders',
'rest_framework',
'django_cassandra_engine',
'rest_framework_simplejwt',
# My apps
'ptlaws_api.apps.PtlawsApiConfig',
]
The error I am getting:
ValueError: Field 'id' expected a number but got '65408306-b199-4ae8-9324-dc8ca0b9bf2c'.
----------------------------------------------------------------------
Ran 4 tests in 9.218s
FAILED (errors=1)
Destroying test database for alias 'default'...