问询：如何简化Django Ninja中JWT认证与授权的实现流程

1. 扩展Django用户模型 #

首先扩展默认用户模型，支持邮箱作为登录标识，添加全名字段：

models.py

from django.contrib.auth.models import AbstractUser, BaseUserManager
from django.db import models

class CustomUserManager(BaseUserManager):
    def create_user(self, email, password=None, **extra_fields):
        if not email:
            raise ValueError('The Email field must be set')
        email = self.normalize_email(email)
        user = self.model(email=email, **extra_fields)
        user.set_password(password)
        user.save(using=self._db)
        return user

    def create_superuser(self, email, password=None, **extra_fields):
        extra_fields.setdefault('is_staff', True)
        extra_fields.setdefault('is_superuser', True)
        return self.create_user(email, password, **extra_fields)

class CustomUser(AbstractUser):
    username = None
    email = models.EmailField(unique=True)
    full_name = models.CharField(max_length=255)

    USERNAME_FIELD = 'email'
    REQUIRED_FIELDS = ['full_name']

    objects = CustomUserManager()

    def __str__(self):
        return self.email

在settings.py中配置自定义用户模型：

AUTH_USER_MODEL = 'yourapp.CustomUser'

2. 实现注册API #

先定义注册请求的Schema，再编写注册接口：

schemas.py

from ninja import Schema
from pydantic import EmailStr

class RegisterSchema(Schema):
    email: EmailStr
    password: str
    full_name: str

class TokenSchema(Schema):
    access_token: str
    token_type: str = "bearer"

api.py（或你的路由文件）

from ninja import NinjaAPI, Router
from django.contrib.auth import get_user_model
from .schemas import RegisterSchema, TokenSchema

api = NinjaAPI()
auth_router = Router()
User = get_user_model()

@auth_router.post("/register", response={201: dict})
def register(request, payload: RegisterSchema):
    if User.objects.filter(email=payload.email).exists():
        return 400, {"detail": "Email already registered"}
    
    user = User.objects.create_user(
        email=payload.email,
        password=payload.password,
        full_name=payload.full_name
    )
    return 201, {"detail": "User registered successfully", "user_id": user.id}

3. 手动实现JWT生成与验证 #

先安装依赖pyjwt：pip install pyjwt，然后编写JWT工具函数：

utils/jwt_utils.py

import jwt
from datetime import datetime, timedelta
from django.conf import settings
from django.contrib.auth import get_user_model

User = get_user_model()

def generate_jwt_token(user):
    payload = {
        "user_id": user.id,
        "email": user.email,
        "exp": datetime.utcnow() + timedelta(hours=24)  # 24小时过期
    }
    token = jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")
    return token

def validate_jwt_token(token):
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
        user = User.objects.get(id=payload["user_id"])
        return user
    except jwt.ExpiredSignatureError:
        return None
    except jwt.InvalidTokenError:
        return None
    except User.DoesNotExist:
        return None

4. 实现登录API #

在api.py中添加登录接口，验证用户身份并返回JWT令牌：

from django.contrib.auth import authenticate
from .utils.jwt_utils import generate_jwt_token

@auth_router.post("/login", response=TokenSchema)
def login(request, email: str, password: str):
    user = authenticate(request, username=email, password=password)
    if not user:
        return 401, {"detail": "Invalid credentials"}
    
    token = generate_jwt_token(user)
    return {"access_token": token}

5. 编写认证函数保护接口 #

创建一个认证函数，用于验证请求中的JWT令牌，然后在需要保护的接口上指定该函数：

api.py

from .utils.jwt_utils import validate_jwt_token

def jwt_auth(request):
    auth_header = request.META.get("HTTP_AUTHORIZATION")
    if not auth_header:
        return None
    try:
        token_type, token = auth_header.split()
        if token_type.lower() != "bearer":
            return None
    except ValueError:
        return None
    
    user = validate_jwt_token(token)
    return user

# 示例：受保护的接口
@auth_router.get("/protected", auth=jwt_auth)
def protected(request):
    return {
        "detail": "This is a protected endpoint",
        "user_id": request.auth.id,
        "user_email": request.auth.email
    }

最后记得把路由注册到主API：

api.add_router("/api", auth_router)

使用说明 #

• 注册：发送POST请求到/api/register，携带email、password、full_name
• 登录：发送POST请求到/api/login，携带email和password，获取access_token
• 访问受保护接口：在请求头中添加Authorization: Bearer <your_token>，再访问/api/protected

内容的提问来源于stack exchange，提问作者Krishnakant Mane