factory/apps/system/serializers.py

292 lines
7.1 KiB
Python

import re
from django_celery_beat.models import PeriodicTask, CrontabSchedule, IntervalSchedule
from rest_framework import serializers
from django_celery_results.models import TaskResult
from apps.utils.serializers import CustomModelSerializer
from apps.utils.vars import EXCLUDE_FIELDS, EXCLUDE_FIELDS_BASE
from .models import (Dict, DictType, File, Dept, Permission, Post,
Role, User, UserPost)
class IntervalSerializer(CustomModelSerializer):
class Meta:
model = IntervalSchedule
fields = '__all__'
class CrontabSerializer(CustomModelSerializer):
class Meta:
model = CrontabSchedule
exclude = ['timezone']
class PTaskCreateUpdateSerializer(CustomModelSerializer):
class Meta:
model = PeriodicTask
fields = ['name', 'task', 'interval', 'crontab', 'args', 'kwargs']
class PTaskSerializer(CustomModelSerializer):
interval_ = IntervalSerializer(source='interval', read_only=True)
crontab_ = CrontabSerializer(source='crontab', read_only=True)
schedule = serializers.SerializerMethodField()
timetype = serializers.SerializerMethodField()
class Meta:
model = PeriodicTask
fields = '__all__'
@staticmethod
def setup_eager_loading(queryset):
""" Perform necessary eager loading of data. """
queryset = queryset.select_related('interval', 'crontab')
return queryset
def get_schedule(self, obj):
if obj.interval:
return obj.interval.__str__()
elif obj.crontab:
return obj.crontab.__str__()
return ''
def get_timetype(self, obj):
if obj.interval:
return 'interval'
elif obj.crontab:
return 'crontab'
return 'interval'
class FileSerializer(CustomModelSerializer):
class Meta:
model = File
fields = "__all__"
class DictTypeSerializer(CustomModelSerializer):
"""
数据字典类型序列化
"""
class Meta:
model = DictType
fields = '__all__'
class DictTypeCreateUpdateSerializer(CustomModelSerializer):
class Meta:
model = DictType
fields = ['name', 'code', 'parent']
class DictSerializer(CustomModelSerializer):
"""
数据字典序列化
"""
class Meta:
model = Dict
fields = '__all__'
class DictCreateUpdateSerializer(CustomModelSerializer):
"""
数据字典序列化
"""
class Meta:
model = Dict
exclude = EXCLUDE_FIELDS
class PostSerializer(CustomModelSerializer):
"""
岗位序列化
"""
class Meta:
model = Post
fields = '__all__'
class PostCreateUpdateSerializer(CustomModelSerializer):
"""
岗位序列化
"""
class Meta:
model = Post
exclude = EXCLUDE_FIELDS
class PostSimpleSerializer(CustomModelSerializer):
class Meta:
model = Post
fields = ['id', 'name', 'code']
class RoleSerializer(CustomModelSerializer):
"""
角色序列化
"""
class Meta:
model = Role
fields = '__all__'
class RoleSimpleSerializer(CustomModelSerializer):
class Meta:
model = Role
fields = ['id', 'name', 'code']
class RoleCreateUpdateSerializer(CustomModelSerializer):
"""
角色序列化
"""
class Meta:
model = Role
exclude = EXCLUDE_FIELDS
class PermissionSerializer(CustomModelSerializer):
"""
权限序列化
"""
class Meta:
model = Permission
fields = '__all__'
class PermissionCreateUpdateSerializer(CustomModelSerializer):
"""
权限序列化
"""
class Meta:
model = Permission
exclude = EXCLUDE_FIELDS_BASE
class DeptSimpleSerializer(CustomModelSerializer):
class Meta:
model = Dept
fields = ['id', 'name', 'type']
class DeptSerializer(CustomModelSerializer):
"""
组织架构序列化
"""
type = serializers.ChoiceField(
choices=Dept.dept_type_choices, default='部门')
class Meta:
model = Dept
fields = '__all__'
class DeptCreateUpdateSerializer(CustomModelSerializer):
"""
部门序列化
"""
class Meta:
model = Dept
exclude = EXCLUDE_FIELDS
class UserSimpleSerializer(CustomModelSerializer):
class Meta:
model = User
fields = ['id', 'username', 'name']
class UserListSerializer(CustomModelSerializer):
"""
用户列表序列化
"""
belong_dept_ = DeptSimpleSerializer(source='belong_dept', read_only=True)
posts_ = PostSimpleSerializer(source='posts', many=True)
class Meta:
model = User
exclude = ['password']
@staticmethod
def setup_eager_loading(queryset):
""" Perform necessary eager loading of data. """
queryset = queryset.select_related('superior', 'belong_dept')
queryset = queryset.prefetch_related('posts')
return queryset
class UserUpdateSerializer(CustomModelSerializer):
"""
用户编辑序列化
"""
phone = serializers.CharField(max_length=11, required=False)
class Meta:
model = User
fields = ['id', 'username', 'name', 'phone', 'email', 'belong_dept',
'avatar', 'is_active', 'is_superuser']
def validate_phone(self, phone):
re_phone = '^1[358]\d{9}$|^147\d{8}$|^176\d{8}$'
if not re.match(re_phone, phone):
raise serializers.ValidationError('手机号码不合法')
return phone
class UserCreateSerializer(CustomModelSerializer):
"""
创建用户序列化
"""
username = serializers.CharField(required=True)
phone = serializers.CharField(max_length=11, required=False)
class Meta:
model = User
fields = ['id', 'username', 'name', 'phone', 'email', 'belong_dept',
'avatar', 'is_active']
def validate_username(self, username):
if User.objects.filter(username=username):
raise serializers.ValidationError(username + ' 账号已存在')
return username
def validate_phone(self, phone):
re_phone = '^1[358]\d{9}$|^147\d{8}$|^176\d{8}$'
if not re.match(re_phone, phone):
raise serializers.ValidationError('手机号码不合法')
if User.objects.filter(phone=phone):
raise serializers.ValidationError('手机号已经被注册')
return phone
class PTaskResultSerializer(CustomModelSerializer):
class Meta:
model = TaskResult
fields = '__all__'
class UserPostSerializer(CustomModelSerializer):
"""
用户-岗位序列化
"""
user_ = UserSimpleSerializer(source='user', read_only=True)
post_ = PostSimpleSerializer(source='post', read_only=True)
dept_ = DeptSimpleSerializer(source='dept', read_only=True)
class Meta:
model = UserPost
fields = '__all__'
class UserPostCreateSerializer(CustomModelSerializer):
class Meta:
model = UserPost
exclude = EXCLUDE_FIELDS_BASE
class UserInfoSerializer(CustomModelSerializer):
posts_ = UserPostSerializer(source='post', read_only=True)
class Meta:
model = User
fields = ['id', 'username', 'name', 'posts_', 'avatar']