Django如何实现分表的Customer/Dealer模型及JWT鉴权接口

你当前的代码存在3个硬伤，直接跑不通认证逻辑：

• Customer和Dealer是普通模型，没有继承AbstractBaseUser和PermissionsMixin，无法被Django认证体系识别，也不能正常生成JWT
• 三个模型重复定义了大量相同字段，数据冗余且无法同步，甚至漏了address这类在REQUIRED_FIELDS里声明的字段
• create_superuser方法参数顺序传错，把password插到了mobile_number的位置，创建超级用户时会直接报错
• 没有做角色校验逻辑，无法实现登录入口隔离

推荐用单核心用户表+代理模型的方案实现需求，完全兼容Django认证逻辑，开发和维护成本远低于建三张独立用户表。

from django.db import models
from django.contrib.auth.models import AbstractBaseUser, BaseUserManager, PermissionsMixin
from rest_framework_simplejwt.tokens import RefreshToken

# 固定角色枚举，避免硬编码写错
class UserRoles(models.TextChoices):
    CUSTOMER = "Customer", "普通客户"
    DEALER = "Dealer", "经销商"

class UserManager(BaseUserManager):
    def _create_user(self, username, email, mobile_number, account_role, account_type, role, address, state, city, pincode, password, **extra_fields):
        if not email:
            raise ValueError("邮箱不能为空")
        email = self.normalize_email(email)
        user = self.model(
            username=username,
            email=email,
            mobile_number=mobile_number,
            account_role=account_role,
            account_type=account_type,
            role=role,
            address=address,
            state=state,
            city=city,
            pincode=pincode,
            **extra_fields
        )
        user.set_password(password)
        user.save(using=self._db)
        return user

    def create_user(self, username, email, mobile_number, account_role, account_type, role, address, state, city, pincode, password=None, **extra_fields):
        extra_fields.setdefault("is_staff", False)
        extra_fields.setdefault("is_superuser", False)
        return self._create_user(username, email, mobile_number, account_role, account_type, role, address, state, city, pincode, password, **extra_fields)

    def create_superuser(self, username, email, mobile_number, password, account_type, role, address, state, city, pincode, **extra_fields):
        extra_fields.setdefault("is_staff", True)
        extra_fields.setdefault("is_superuser", True)
        extra_fields.setdefault("is_verified", True)
        return self._create_user(
            username=username,
            email=email,
            mobile_number=mobile_number,
            account_role=UserRoles.CUSTOMER,
            account_type=account_type,
            role=role,
            address=address,
            state=state,
            city=city,
            pincode=pincode,
            password=password,
            **extra_fields
        )

# 核心用户表，存储所有用户的认证相关数据
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)
    mobile_number = models.CharField(max_length=12)
    account_role = models.CharField(max_length=50, choices=UserRoles.choices)
    account_type = models.CharField(max_length=50)
    role = models.CharField(max_length=50, null=True, blank=True)
    address = models.TextField()
    state = models.CharField(max_length=100)
    city = models.CharField(max_length=100)
    pincode = models.CharField(max_length=20)
    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)

    USERNAME_FIELD = 'email'
    REQUIRED_FIELDS = ['username', 'mobile_number', 'account_type', 'role', 'address', 'state', 'city', 'pincode']

    objects = UserManager()

    def tokens(self):
        refresh = RefreshToken.for_user(self)
        # 把角色写入token payload，后续校验不用反复查库
        refresh['account_role'] = self.account_role
        refresh['user_id'] = self.id
        return {
            'refresh': str(refresh),
            'access': str(refresh.access_token)
        }

# 客户代理模型，操作时自动过滤客户角色数据，相当于独立管理入口
class CustomerManager(UserManager):
    def get_queryset(self):
        return super().get_queryset().filter(account_role=UserRoles.CUSTOMER)

class Customer(User):
    objects = CustomerManager()
    class Meta:
        proxy = True
        verbose_name = "客户"

    def save(self, *args, **kwargs):
        self.account_role = UserRoles.CUSTOMER
        return super().save(*args, **kwargs)

# 经销商代理模型，操作时自动过滤经销商角色数据
class DealerManager(UserManager):
    def get_queryset(self):
        return super().get_queryset().filter(account_role=UserRoles.DEALER)

class Dealer(User):
    objects = DealerManager()
    class Meta:
        proxy = True
        verbose_name = "经销商"

    def save(self, *args, **kwargs):
        self.account_role = UserRoles.DEALER
        return super().save(*args, **kwargs)

改完模型后必须执行python manage.py makemigrations和python manage.py migrate生效。

在settings.py中添加以下配置：

• 配置自定义用户模型：AUTH_USER_MODEL = "你的应用名.User"
• 在INSTALLED_APPS中添加'rest_framework_simplejwt.token_blacklist'，支持登出时拉黑token
• 配置simplejwt默认认证类、token有效期等参数，根据业务需求调整即可

1. 登录序列化器（核心做角色校验） #

from rest_framework import serializers
from rest_framework.exceptions import AuthenticationFailed
from django.contrib.auth import authenticate
from .models import UserRoles

class CustomerLoginSerializer(serializers.Serializer):
    email = serializers.EmailField()
    password = serializers.CharField(write_only=True)

    def validate(self, attrs):
        user = authenticate(email=attrs['email'], password=attrs['password'])
        if not user:
            raise AuthenticationFailed("账号或密码错误")
        if not user.is_active:
            raise AuthenticationFailed("账号已被禁用")
        # 核心隔离逻辑：非客户角色直接拒绝
        if user.account_role != UserRoles.CUSTOMER:
            raise AuthenticationFailed("请使用对应角色入口登录")
        if not user.is_verified:
            raise AuthenticationFailed("请先完成账号验证")
        return {"email": user.email, "username": user.username, "tokens": user.tokens()}

class DealerLoginSerializer(serializers.Serializer):
    email = serializers.EmailField()
    password = serializers.CharField(write_only=True)

    def validate(self, attrs):
        user = authenticate(email=attrs['email'], password=attrs['password'])
        if not user:
            raise AuthenticationFailed("账号或密码错误")
        if not user.is_active:
            raise AuthenticationFailed("账号已被禁用")
        # 核心隔离逻辑：非经销商角色直接拒绝
        if user.account_role != UserRoles.DEALER:
            raise AuthenticationFailed("请使用对应角色入口登录")
        if not user.is_verified:
            raise AuthenticationFailed("请先完成账号验证")
        return {"email": user.email, "username": user.username, "tokens": user.tokens()}

2. 接口视图实现 #

注册接口分别调用Customer.objects.create_user和Dealer.objects.create_user，自动绑定对应角色，不需要手动传account_role参数：

from rest_framework import generics, status
from rest_framework.response import Response
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework_simplejwt.tokens import RefreshToken
from .models import Customer, Dealer, User
from .serializers import CustomerLoginSerializer, DealerLoginSerializer
from django.contrib.auth.tokens import PasswordResetTokenGenerator, default_token_generator
from django.utils.http import urlsafe_base64_encode, urlsafe_base64_decode
from django.utils.encoding import smart_bytes, smart_str
from django.core.mail import send_mail

# 客户注册
class CustomerRegisterView(generics.GenericAPIView):
    permission_classes = [AllowAny]
    def post(self, request):
        # 自行编写客户注册序列化器，校验字段格式
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        user = Customer.objects.create_user(**serializer.validated_data)
        # 此处添加发送验证邮件逻辑，生成验证token和uidb64拼接成验证链接
        return Response({"msg": "注册成功，请查收验证邮件"}, status=status.HTTP_201_CREATED)

# 客户登录
class CustomerLoginView(generics.GenericAPIView):
    permission_classes = [AllowAny]
    serializer_class = CustomerLoginSerializer
    def post(self, request):
        serializer = self.serializer_class(data=request.data)
        serializer.is_valid(raise_exception=True)
        return Response(serializer.validated_data, status=status.HTTP_200_OK)

# 经销商注册
class DealerRegisterView(generics.GenericAPIView):
    permission_classes = [AllowAny]
    def post(self, request):
        # 自行编写经销商注册序列化器，校验字段格式
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        user = Dealer.objects.create_user(**serializer.validated_data)
        # 此处添加发送验证邮件逻辑
        return Response({"msg": "注册成功，请查收验证邮件"}, status=status.HTTP_201_CREATED)

# 经销商登录
class DealerLoginView(generics.GenericAPIView):
    permission_classes = [AllowAny]
    serializer_class = DealerLoginSerializer
    def post(self, request):
        serializer = self.serializer_class(data=request.data)
        serializer.is_valid(raise_exception=True)
        return Response(serializer.validated_data, status=status.HTTP_200_OK)

# 通用登出
class LogoutView(generics.GenericAPIView):
    permission_classes = [IsAuthenticated]
    def post(self, request):
        try:
            refresh_token = request.data["refresh"]
            token = RefreshToken(refresh_token)
            token.blacklist()
            return Response({"msg": "登出成功"}, status=status.HTTP_205_RESET_CONTENT)
        except:
            return Response({"error": "无效的token"}, status=status.HTTP_400_BAD_REQUEST)

# 账号验证
class VerifyEmailView(generics.GenericAPIView):
    permission_classes = [AllowAny]
    def get(self, request, uidb64, token):
        try:
            user_id = smart_str(urlsafe_base64_decode(uidb64))
            user = User.objects.get(id=user_id)
            if not default_token_generator.check_token(user, token):
                return Response({"error": "验证链接无效或已过期"}, status=status.HTTP_400_BAD_REQUEST)
            user.is_verified = True
            user.save()
            return Response({"msg": "账号验证成功"}, status=status.HTTP_200_OK)
        except:
            return Response({"error": "验证链接无效"}, status=status.HTTP_400_BAD_REQUEST)

# 密码重置接口（发重置邮件、确认重置）逻辑和邮箱验证一致，不需要区分角色，通用即可

内容的提问来源于stack exchange，提问作者Alan Wake