我在 Django 应用程序中尝试创建包含子任务和用户的任务时遇到问题。由于验证错误,特别是子任务和用户字段的验证错误,任务创建失败。这是我发送到后端的有效负载:
{
"id": null,
"title": "aa",
"description": "aaaa",
"due_to": "2024-06-27T22:00:00.000Z",
"created": null,
"updated": null,
"priority": "LOW",
"category": "TECHNICAL_TASK",
"status": "TO_DO",
"subtasks": [
{
"task_id": null,
"description": "s1",
"is_done": false
},
{
"task_id": null,
"description": "s2",
"is_done": false
}
],
"users": [
{
"id": 6
},
{
"id": 7
}
]
}
当我尝试创建任务时,我从后端收到以下错误消息:
{
"subtasks": [
{
"task_id": [
"This field may not be null."
]
},
{
"task_id": [
"This field may not be null."
]
}
],
"users": [
{
"email": [
"This field is required."
],
"password": [
"This field is required."
],
"name": [
"This field is required."
]
},
{
"email": [
"This field is required."
],
"password": [
"This field is required."
],
"name": [
"This field is required."
]
}
]
}
我怀疑这个问题与我在 Django 应用程序中处理嵌套序列化器和模型关系的方式有关。我正在使用 Django Rest Framework (DRF) 来处理数据的序列化和反序列化。
这是我的相关代码:
序列化器.py
class UserSerializer(serializers.ModelSerializer):
"""Serializer for the user object."""
class Meta:
model = get_user_model()
fields = ['id', 'email', 'password', 'name', 'phone_number', 'avatar_color']
extra_kwargs = {'password': {'write_only': True, 'style': {'input_type': 'password'}, 'min_length': 6},
'avatar_color': {'read_only': True}
}
def create(self, validated_data):
"""Create and return a user with encrypted password."""
return get_user_model().objects.create_user(**validated_data)
def update(self, instance, validated_data):
"""Update and return user."""
password = validated_data.pop('password', None)
user = super().update(instance, validated_data)
if password:
user.set_password(password)
user.save()
return user
class AuthTokenSerializer(serializers.Serializer):
"""Serializer for the user auth token."""
email = serializers.EmailField()
password = serializers.CharField(
style={'input_type': 'password'},
trim_whitespace=False,
)
def validate(self, attrs):
"""Validate and authenticate the user."""
email = attrs.get('email')
password = attrs.get('password')
user = authenticate(
request=self.context.get('request'),
username=email,
password=password,
)
if not user:
msg = _('Unable to authenticate with provided credentials.')
raise serializers.ValidationError(msg, code='authorization')
attrs['user'] = user
return attrs
class TaskSerializer(serializers.ModelSerializer):
"""Serializes a task object"""
subtasks = SubtaskSerializer(many=True, required=False)
users = UserSerializer(many=True, required=False)
class Meta:
model = Task
fields = ['id', 'title', 'description', 'due_to', 'created', 'updated', 'priority', 'category', 'status',
'subtasks', 'users']
read_only_fields = ['created', 'updated']
def create(self, validated_data):
users_data = validated_data.pop('users', None)
subtasks_data = validated_data.pop('subtasks', None)
task = Task.objects.create(**validated_data)
if users_data:
for user_data in users_data:
user = User.objects.get(id=user_data['id'])
task.users.add(user)
if subtasks_data:
for subtask_data in subtasks_data:
subtask_data['task_id'] = task.id
SubtaskSerializer().create(validated_data=subtask_data)
return task
class SubtaskSerializer(serializers.ModelSerializer):
task_id = serializers.IntegerField(write_only=True, required=False)
class Meta:
model = Subtask
fields = ['id', 'task_id', 'description', 'is_done']
read_only_fields = ['id']
def create(self, validated_data):
task_id = validated_data.pop('task_id', None)
task = Task.objects.get(id=task_id)
subtask = Subtask.objects.create(task=task, **validated_data)
return subtask
模型.py
class UserManager(BaseUserManager):
"""Manager for user"""
def create_user(self, email, name, password=None, **extra_fields):
"""Create, save and return a new user."""
if not email:
raise ValueError('User must have an email address.')
user = self.model(email=self.normalize_email(email), name=name, **extra_fields)
user.set_password(password)
user.save(using=self._db)
return user
def create_superuser(self, email, name, password):
"""Create and save a new superuser with given details"""
user = self.create_user(email, name, password)
user.is_superuser = True
user.is_staff = True
user.save(using=self._db)
return user
class User(AbstractBaseUser, PermissionsMixin):
"""Database model for users in the system"""
email = models.EmailField(unique=True)
name = models.CharField(max_length=50)
is_active = models.BooleanField(default=True)
is_staff = models.BooleanField(default=False)
phone_number = models.CharField(max_length=20, blank=True, null=True)
avatar_color = models.CharField(max_length=7)
objects = UserManager()
USERNAME_FIELD = 'email'
REQUIRED_FIELDS = ['name']
def save(self, *args, **kwargs):
if not self.pk:
self.avatar_color = random.choice([
'#FF5733', '#C70039', '#900C3F', '#581845',
'#8E44AD', '#1F618D', '#008000', '#A52A2A', '#000080'
])
super().save(*args, **kwargs)
def get_full_name(self):
"""Retrieve full name for user"""
return self.name
def __str__(self):
"""Return string representation of user"""
return self.email
class Task(models.Model):
class Priority(models.TextChoices):
LOW = "LOW"
MEDIUM = "MEDIUM"
URGENT = "URGENT"
class Category(models.TextChoices):
TECHNICAL_TASK = "TECHNICAL_TASK"
USER_STORY = "USER_STORY"
class TaskStatus(models.TextChoices):
TO_DO = "TO_DO"
AWAIT_FEEDBACK = "AWAIT_FEEDBACK"
IN_PROGRESS = "IN_PROGRESS"
DONE = "DONE"
id = models.AutoField(primary_key=True)
title = models.TextField()
description = models.TextField(blank=True, null=True)
due_to = models.DateTimeField()
created = models.DateTimeField()
updated = models.DateTimeField(auto_now_add=True)
priority = models.TextField(choices=Priority.choices)
category = models.TextField(choices=Category.choices)
status = models.TextField(choices=TaskStatus.choices)
users = models.ManyToManyField(User, blank=True, null=True, related_name='tasks')
def save(self, *args, **kwargs):
if not self.id:
self.created = timezone.now()
super().save(*args, **kwargs)
def __str__(self):
return f"{self.title}"
class Subtask(models.Model):
"""Subtask object."""
id = models.AutoField(primary_key=True)
description = models.TextField()
is_done = models.BooleanField(default=False)
task = models.ForeignKey(Task, on_delete=models.CASCADE, related_name='subtasks')
def __str__(self):
return self.description
当你开始用 drf 重写
create
validate
时,除非它绝对合理(相当小众),否则你正在破坏 DRF 希望你使用的模式。
在您的情况下,DRF 的行为与
patch
vs put
vs post
等不同...
对于嵌套关系,您必须指定那些嵌套序列化器,例如:
Non-nested: task_id = serializers.IntegerField(write_only=True, required=False)
Nested: task = TaskSerializer(required=False)
最后:
编写单元测试来隔离和测试您感兴趣的功能,编写您想要的最小垂直切片并测试它,然后扩展它,不要尝试一次完成所有工作。
一个关键要点是单元测试可能看起来很乏味,但从短期和长期来看它们总是节省时间,因为它们避免了这些令人头痛的问题。