章节5：认证权限与系统进阶

---

一、章节概述

本章聚焦 Django 项目中的认证授权体系与系统进阶功能。从 JWT Token 认证机制入手，深入讲解 RBAC 角色权限模型、用户系统扩展与第三方登录集成、文件上传与静态资源管理，以及 Admin 后台的深度定制化开发。这些知识是构建企业级 Django 应用的必备技能。

二、学习目标

目标分类	具体目标
知识目标	理解 JWT 认证的工作原理与优势；掌握 RBAC 权限模型的设计思想；了解 OAuth2.0 授权流程
技能目标	能集成 simplejwt 实现 JWT 认证；能设计并实现 RBAC 权限系统；能扩展 AbstractUser 自定义用户模型；能深度定制 Admin 后台
素养目标	培养安全优先的开发意识；理解企业级系统中的认证授权架构

三、核心知识点

---

3.1 JWT Token 认证机制

定义

JWT（JSON Web Token） 是一种基于 JSON 的开放标准（RFC 7519），用于在各方之间安全地传输信息。JWT Token 由三部分组成——Header（头部）、Payload（载荷）和 Signature（签名），以 . 分隔。服务器无需存储 Token 即可验证用户身份。

JWT 结构

# JWT 格式
header.payload.signature

# 示例
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.
eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.
SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c

部分	内容	说明
Header	{"alg": "HS256", "typ": "JWT"}	签名算法和类型
Payload	{"sub": "1", "username": "admin", "exp": 1700000000}	声明数据（用户信息、过期时间等）
Signature	对前两部分签名后的结果	防止数据被篡改

JWT 认证流程

客户端                             服务器
  │                                  │
  │  ── POST /api/token/ ──────────► │  用户名 + 密码
  │                                  │  验证用户
  │  ◄── {access, refresh} ──────── │  返回 JWT 令牌
  │                                  │
  │  ── GET /api/articles/ ─────────► │  Authorization: Bearer <access_token>
  │   (携带 access_token)            │  验证 Token
  │  ◄── 200 JSON ───────────────── │  返回数据
  │                                  │
  │  (access_token 过期)             │
  │  ── POST /api/token/refresh/ ───► │  携带 refresh_token
  │  ◄── {新 access_token} ───────── │  获取新的访问令牌

安装与配置

pip install djangorestframework-simplejwt

# settings.py —— JWT 配置

INSTALLED_APPS = [
    # ...
    'rest_framework',
    'rest_framework_simplejwt',
    'rest_framework_simplejwt.token_blacklist',  # Token 黑名单（可选）
]

from datetime import timedelta

SIMPLE_JWT = {
    # 签名密钥（默认使用 SECRET_KEY）
    'SIGNING_KEY': 'your-secret-key-here',
    
    # 访问 Token 配置
    'ACCESS_TOKEN_LIFETIME': timedelta(minutes=30),     # 访问令牌有效期
    'REFRESH_TOKEN_LIFETIME': timedelta(days=7),        # 刷新令牌有效期
    'ROTATE_REFRESH_TOKENS': True,       # 刷新时是否轮换 refresh_token
    'BLACKLIST_AFTER_ROTATION': True,    # 轮换后将旧 refresh_token 加入黑名单
    
    # 算法
    'ALGORITHM': 'HS256',
    
    # 认证
    'AUTH_HEADER_TYPES': ('Bearer',),    # Authorization: Bearer <token>
    'AUTH_HEADER_NAME': 'HTTP_AUTHORIZATION',
    
    # Token 中包含的字段
    'USER_ID_FIELD': 'id',
    'USER_ID_CLAIM': 'user_id',
    
    # Token 类型声明
    'TOKEN_TYPE_CLAIM': 'token_type',
    
    # 其他声明
    'JTI_CLAIM': 'jti',  # JWT ID（用于黑名单校验）
    
    # 序列化
    'TOKEN_USER_CLASS': 'rest_framework_simplejwt.models.TokenUser',
}

# DRF 全局认证配置
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework_simplejwt.authentication.JWTAuthentication',
        # 可以保留 Session 认证用于 Admin 后台
        'rest_framework.authentication.SessionAuthentication',
    ],
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticated',
    ],
}

URL 路由配置

# myproject/urls.py

from django.urls import path
from rest_framework_simplejwt.views import (
    TokenObtainPairView,       # 获取 Token 对（access + refresh）
    TokenRefreshView,          # 刷新 Token
    TokenVerifyView,           # 验证 Token
    TokenBlacklistView,        # 将 refresh_token 加入黑名单
)

urlpatterns = [
    # 登录获取 Token
    path('api/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'),
    # 刷新 Token
    path('api/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
    # 验证 Token
    path('api/token/verify/', TokenVerifyView.as_view(), name='token_verify'),
    # 登出（黑名单）
    path('api/token/blacklist/', TokenBlacklistView.as_view(), name='token_blacklist'),
]

自定义 Token

# blog/serializers.py —— 自定义 JWT Token

from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
from rest_framework_simplejwt.views import TokenObtainPairView

class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):
    """自定义 Token 返回内容"""
    
    @classmethod
    def get_token(cls, user):
        token = super().get_token(user)
        
        # 添加自定义声明（claims）
        token['username'] = user.username
        token['email'] = user.email
        token['is_staff'] = user.is_staff
        token['roles'] = list(user.groups.values_list('name', flat=True))
        
        return token
    
    def validate(self, attrs):
        """自定义验证（包含额外返回字段）"""
        data = super().validate(attrs)
        
        # 添加额外的返回数据
        data['user_id'] = self.user.id
        data['username'] = self.user.username
        data['email'] = self.user.email
        data['roles'] = list(self.user.groups.values_list('name', flat=True))
        
        return data

# 使用自定义视图
class CustomTokenObtainPairView(TokenObtainPairView):
    serializer_class = CustomTokenObtainPairSerializer

在视图中使用

from rest_framework.permissions import IsAuthenticated
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.decorators import api_view, permission_classes

# 方式一：APIView
class UserInfoView(APIView):
    permission_classes = [IsAuthenticated]
    
    def get(self, request):
        """获取当前用户信息"""
        user = request.user
        return Response({
            'id': user.id,
            'username': user.username,
            'email': user.email,
            'is_staff': user.is_staff,
        })

# 方式二：函数视图装饰器
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def my_profile(request):
    """获取我的个人信息"""
    return Response({
        'username': request.user.username,
        'email': request.user.email,
    })

# 方式三：ViewSet
class ArticleViewSet(ModelViewSet):
    permission_classes = [IsAuthenticated]
    # ...

前端请求示例

// JavaScript 前端使用 JWT

// 登录
async function login(username, password) {
    const response = await fetch('/api/token/', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ username, password }),
    });
    const data = await response.json();
    
    // 保存 Token
    localStorage.setItem('access_token', data.access);
    localStorage.setItem('refresh_token', data.refresh);
    return data;
}

// 携带 Token 请求 API
async function fetchArticles() {
    const token = localStorage.getItem('access_token');
    
    const response = await fetch('/api/articles/', {
        headers: {
            'Authorization': `Bearer ${token}`,
            'Content-Type': 'application/json',
        },
    });
    return await response.json();
}

// 自动刷新 Token
async function fetchWithAuth(url, options = {}) {
    const token = localStorage.getItem('access_token');
    
    options.headers = {
        ...options.headers,
        'Authorization': `Bearer ${token}`,
    };
    
    let response = await fetch(url, options);
    
    if (response.status === 401) {
        // Token 过期，尝试刷新
        const refreshToken = localStorage.getItem('refresh_token');
        const refreshResponse = await fetch('/api/token/refresh/', {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({ refresh: refreshToken }),
        });
        
        if (refreshResponse.ok) {
            const data = await refreshResponse.json();
            localStorage.setItem('access_token', data.access);
            
            // 重试原请求
            options.headers['Authorization'] = `Bearer ${data.access}`;
            response = await fetch(url, options);
        }
    }
    
    return response;
}

---

3.2 RBAC 角色权限模型

定义

RBAC（Role-Based Access Control，基于角色的访问控制） 是将权限授予角色，再将角色分配给用户的权限管理模型。Django 内置了 Permission（权限）和 Group（角色组）模型，天然支持 RBAC。

用户 ───→ 角色（Group） ───→ 权限（Permission）
                              │
                              ├── 模型权限（view/add/change/delete）
                              ├── 自定义权限
                              └── 功能权限

Django 内置权限系统

▶ 模型权限（自动生成）

每个 Django 模型在 migrate 时自动生成 4 个基本权限：

权限代号（codename）	说明	URL 对应动作
view_article	查看文章	GET /articles/
add_article	添加文章	POST /articles/
change_article	修改文章	PUT /articles/{id}/
delete_article	删除文章	DELETE /articles/{id}/

# 查看所有权限
from django.contrib.auth.models import Permission

# 获取模型的所有权限
permissions = Permission.objects.filter(content_type__app_label='blog', content_type__model='article')
for perm in permissions:
    print(f"{perm.codename} → {perm.name}")
# 输出：
# view_article → Can view article
# add_article → Can add article
# change_article → Can change article
# delete_article → Can delete article

▶ 自定义权限

from django.db import models

class Article(models.Model):
    # ... 字段定义
    
    class Meta:
        permissions = [
            ('publish_article', '可以发布文章'),
            ('approve_comment', '可以审核评论'),
            ('export_articles', '可以导出文章数据'),
            ('view_statistics', '可以查看统计报表'),
        ]

# 添加自定义权限后需执行迁移
python manage.py makemigrations
python manage.py migrate

▶ Group（角色组）管理

from django.contrib.auth.models import Group, Permission
from django.contrib.auth import get_user_model

User = get_user_model()

# 创建角色
editor_group, created = Group.objects.get_or_create(name='编辑')
admin_group, _ = Group.objects.get_or_create(name='管理员')

# 为角色分配权限
article_permissions = Permission.objects.filter(
    content_type__app_label='blog',
    content_type__model='article',
)

# 编辑拥有查看、添加和修改权限
editor_group.permissions.add(*article_permissions.filter(
    codename__in=['view_article', 'add_article', 'change_article']
))

# 管理员拥有所有权限（含删除）
admin_group.permissions.add(*article_permissions)

# 添加自定义权限
publish_perm = Permission.objects.get(codename='publish_article')
editor_group.permissions.add(publish_perm)

# 为用户分配角色
user = User.objects.get(username='alice')
user.groups.add(editor_group)

# 检查用户是否属于角色
if user.groups.filter(name='编辑').exists():
    print('用户是编辑角色')

权限检查

# 方式一：在视图/模板中检查
from django.contrib.auth.decorators import permission_required
from django.contrib.auth.mixins import PermissionRequiredMixin

# FBV 方式
@permission_required('blog.publish_article', raise_exception=True)
def publish_article(request, pk):
    """发布文章（需要 publish_article 权限）"""
    article = get_object_or_404(Article, pk=pk)
    article.is_published = True
    article.save()
    return HttpResponse('已发布')

# CBV 方式
class PublishArticleView(PermissionRequiredMixin, View):
    permission_required = 'blog.publish_article'
    # 或
    # permission_required = ['blog.view_article', 'blog.change_article']
    
    def handle_no_permission(self):
        from django.http import JsonResponse
        return JsonResponse({'error': '权限不足'}, status=403)
    
    def post(self, request, pk):
        article = get_object_or_404(Article, pk=pk)
        article.is_published = True
        article.save()
        return JsonResponse({'status': 'published'})


# 方式二：在代码中检查
def my_view(request):
    if request.user.has_perm('blog.publish_article'):
        # 有权限
        pass
    else:
        # 无权限
        from django.core.exceptions import PermissionDenied
        raise PermissionDenied('您没有发布文章的权限')

# 方式三：DRF 权限类
from rest_framework.permissions import BasePermission

class HasPublishPermission(BasePermission):
    """需要发布文章权限"""
    
    def has_permission(self, request, view):
        return request.user.has_perm('blog.publish_article')


class IsEditorOrAdmin(BasePermission):
    """编辑或管理员角色"""
    
    def has_permission(self, request, view):
        return request.user.groups.filter(
            name__in=['编辑', '管理员']
        ).exists()


class DynamicPermission(BasePermission):
    """动态权限：根据动作检查 Django 内置权限"""
    
    permission_map = {
        'list': '%(app_label)s.view_%(model_name)s',
        'retrieve': '%(app_label)s.view_%(model_name)s',
        'create': '%(app_label)s.add_%(model_name)s',
        'update': '%(app_label)s.change_%(model_name)s',
        'partial_update': '%(app_label)s.change_%(model_name)s',
        'destroy': '%(app_label)s.delete_%(model_name)s',
    }
    
    def has_permission(self, request, view):
        # 获取模型信息
        model_cls = getattr(view, 'queryset', None)
        if model_cls is None:
            model_cls = getattr(view, 'get_queryset')().model
        
        app_label = model_cls._meta.app_label
        model_name = model_cls._meta.model_name
        
        # 获取当前动作对应的权限
        permission_codename = self.permission_map.get(view.action, '')
        if not permission_codename:
            return True
        
        permission = permission_codename % {
            'app_label': app_label,
            'model_name': model_name,
        }
        
        return request.user.has_perm(permission)

完整 RBAC 实现示例

# blog/rbac.py —— RBAC 权限系统

from rest_framework.permissions import BasePermission, SAFE_METHODS
from django.contrib.auth.models import Group, Permission


def create_default_roles():
    """创建默认角色组及其权限"""
    
    # 管理员：所有权限
    admin_group, _ = Group.objects.get_or_create(name='管理员')
    admin_group.permissions.set(Permission.objects.filter(
        content_type__app_label='blog'
    ))
    
    # 编辑：查看、添加、修改、发布
    editor_group, _ = Group.objects.get_or_create(name='编辑')
    editor_perms = Permission.objects.filter(
        content_type__app_label='blog',
        codename__in=[
            'view_article', 'add_article', 'change_article',
            'view_category', 'add_category', 'change_category',
            'view_comment', 'change_comment',
            'publish_article', 'approve_comment',
        ]
    )
    editor_group.permissions.set(editor_perms)
    
    # 作者：只能管理自己的文章
    author_group, _ = Group.objects.get_or_create(name='作者')
    author_perms = Permission.objects.filter(
        content_type__app_label='blog',
        codename__in=[
            'view_article', 'add_article', 'change_article',
            'publish_article',
        ]
    )
    author_group.permissions.set(author_perms)
    
    # 读者：仅查看
    reader_group, _ = Group.objects.get_or_create(name='读者')
    reader_perms = Permission.objects.filter(
        content_type__app_label='blog',
        codename__in=['view_article', 'view_category', 'view_comment']
    )
    reader_group.permissions.set(reader_perms)


class RoleBasedPermission(BasePermission):
    """基于角色的 DRF 权限类"""
    
    def has_permission(self, request, view):
        user = request.user
        
        if not user.is_authenticated:
            return False
        
        if user.is_superuser:
            return True
        
        # 安全检查点
        if view.action == 'list':
            return user.has_perm('blog.view_article')
        elif view.action == 'create':
            return user.has_perm('blog.add_article')
        elif view.action in ('update', 'partial_update'):
            return user.has_perm('blog.change_article')
        elif view.action == 'destroy':
            return user.has_perm('blog.delete_article')
        
        return True
    
    def has_object_permission(self, request, view, obj):
        user = request.user
        
        if user.is_superuser:
            return True
        
        # 编辑和管理员可以操作所有对象
        if user.groups.filter(name__in=['管理员', '编辑']).exists():
            return True
        
        # 作者只能操作自己的文章
        if hasattr(obj, 'author'):
            return obj.author == user
        
        return True

---

3.3 用户系统与第三方登录

定义

Django 内置的 User 模型提供了基本的用户管理功能（用户名、密码、邮箱等）。但在实际项目中，通常需要扩展用户模型以支持手机号、头像等额外字段。OAuth2.0 是第三方登录的开放授权标准，允许用户通过微信、GitHub、Google 等第三方账号登录。

扩展用户模型

▶ 方式一：AbstractUser 扩展（推荐）

# blog/models.py

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

class User(AbstractUser):
    """自定义用户模型"""
    
    # 添加额外字段
    phone = models.CharField(max_length=20, unique=True, blank=True, null=True, verbose_name='手机号')
    avatar = models.ImageField(upload_to='avatars/', blank=True, null=True, verbose_name='头像')
    bio = models.TextField(max_length=500, blank=True, verbose_name='个人简介')
    birth_date = models.DateField(null=True, blank=True, verbose_name='出生日期')
    gender = models.CharField(
        max_length=10,
        choices=[('M', '男'), ('F', '女'), ('O', '其他')],
        blank=True,
        verbose_name='性别'
    )
    wechat_openid = models.CharField(max_length=100, blank=True, verbose_name='微信 OpenID')
    github_id = models.CharField(max_length=100, blank=True, verbose_name='GitHub ID')
    
    # 设置唯一认证字段
    EMAIL_FIELD = 'email'
    USERNAME_FIELD = 'username'  # 登录字段
    REQUIRED_FIELDS = ['email']  # createsuperuser 时必填
    
    class Meta:
        verbose_name = '用户'
        verbose_name_plural = '用户'
    
    def __str__(self):
        return self.username

# settings.py —— 使用自定义用户模型

AUTH_USER_MODEL = 'blog.User'  # 格式：app_label.ModelName

# 重要：必须先在第一次 migrate 前设置 AUTH_USER_MODEL
python manage.py makemigrations
python manage.py migrate

▶ 方式二：OneToOneField 扩展（适用于已有项目）

# 如果项目已经迁移过，无法使用 AbstractUser
# 使用一对一关联扩展

class Profile(models.Model):
    user = models.OneToOneField(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        related_name='profile',
    )
    phone = models.CharField(max_length=20, blank=True)
    avatar = models.ImageField(upload_to='avatars/', blank=True)
    bio = models.TextField(max_length=500, blank=True)
    
    def __str__(self):
        return f'{self.user.username} 的个人资料'

▶ 用户注册序列化器

# blog/serializers.py

from rest_framework import serializers
from django.contrib.auth import get_user_model
from django.contrib.auth.password_validation import validate_password

User = get_user_model()

class UserRegistrationSerializer(serializers.ModelSerializer):
    """用户注册序列化器"""
    password = serializers.CharField(
        write_only=True,
        required=True,
        validators=[validate_password],  # 使用 Django 密码验证器
        style={'input_type': 'password'},
    )
    password2 = serializers.CharField(
        write_only=True,
        required=True,
        style={'input_type': 'password'},
        label='确认密码',
    )
    
    class Meta:
        model = User
        fields = ['id', 'username', 'email', 'phone', 'password', 'password2']
        extra_kwargs = {
            'email': {'required': True},
        }
    
    def validate(self, attrs):
        if attrs['password'] != attrs['password2']:
            raise serializers.ValidationError({"password2": "两次密码不一致"})
        return attrs
    
    def create(self, validated_data):
        validated_data.pop('password2')
        user = User.objects.create_user(**validated_data)
        return user

class UserSerializer(serializers.ModelSerializer):
    """用户信息序列化器"""
    class Meta:
        model = User
        fields = ['id', 'username', 'email', 'phone', 'avatar', 'bio', 'birth_date', 'gender']
        read_only_fields = ['id']

# blog/views.py —— 用户注册视图

from rest_framework import generics, status
from rest_framework.response import Response
from rest_framework.permissions import AllowAny
from .serializers import UserRegistrationSerializer

class UserRegistrationView(generics.CreateAPIView):
    """用户注册"""
    queryset = User.objects.all()
    permission_classes = [AllowAny]
    serializer_class = UserRegistrationSerializer
    
    def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        user = serializer.save()
        
        # 注册成功后自动生成 Token
        from rest_framework_simplejwt.tokens import RefreshToken
        refresh = RefreshToken.for_user(user)
        
        return Response({
            'user': UserSerializer(user).data,
            'refresh': str(refresh),
            'access': str(refresh.access_token),
        }, status=status.HTTP_201_CREATED)

OAuth2 第三方登录

▶ 使用 django-allauth（推荐的第三方登录库）

pip install django-allauth

# settings.py

INSTALLED_APPS = [
    # ...
    'django.contrib.sites',
    'allauth',
    'allauth.account',
    'allauth.socialaccount',
    'allauth.socialaccount.providers.github',
    'allauth.socialaccount.providers.weixin',
    'allauth.socialaccount.providers.google',
]

AUTHENTICATION_BACKENDS = [
    'django.contrib.auth.backends.ModelBackend',
    'allauth.account.auth_backends.AuthenticationBackend',
]

SITE_ID = 1

# allauth 配置
ACCOUNT_EMAIL_VERIFICATION = 'optional'  # 邮箱验证：mandatory/optional/none
ACCOUNT_AUTHENTICATION_METHOD = 'username_email'  # 用户名或邮箱登录
ACCOUNT_EMAIL_REQUIRED = True
ACCOUNT_USERNAME_REQUIRED = True
ACCOUNT_LOGOUT_ON_GET = True
ACCOUNT_LOGIN_REDIRECT_URL = '/'
ACCOUNT_LOGOUT_REDIRECT_URL = '/'

SOCIALACCOUNT_PROVIDERS = {
    'github': {
        'APP': {
            'client_id': 'your-github-client-id',
            'secret': 'your-github-client-secret',
            'key': '',
        },
        'SCOPE': ['user:email'],
    },
    'weixin': {
        'APP': {
            'client_id': 'your-wechat-app-id',
            'secret': 'your-wechat-app-secret',
        },
    },
}

# myproject/urls.py

urlpatterns = [
    # ...
    path('accounts/', include('allauth.urls')),
]

▶ 手动实现 OAuth2 登录（以 GitHub 为例）

# blog/views.py —— 手动实现 GitHub OAuth 登录

import requests
from django.conf import settings
from django.contrib.auth import login, get_user_model
from rest_framework import status
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import AllowAny
from rest_framework_simplejwt.tokens import RefreshToken

User = get_user_model()


class GitHubLoginView(APIView):
    """GitHub OAuth2 登录"""
    permission_classes = [AllowAny]
    
    def get(self, request):
        # 获取 GitHub 授权 URL
        github_auth_url = (
            f"https://github.com/login/oauth/authorize"
            f"?client_id={settings.GITHUB_CLIENT_ID}"
            f"&redirect_uri={settings.GITHUB_REDIRECT_URI}"
            f"&scope=user:email"
        )
        return Response({'auth_url': github_auth_url})
    
    def post(self, request):
        """接收 GitHub 回调后的 code，完成登录"""
        code = request.data.get('code')
        
        if not code:
            return Response({'error': '缺少 code 参数'}, status=status.HTTP_400_BAD_REQUEST)
        
        # 1. 交换 code 获取 access_token
        token_response = requests.post(
            'https://github.com/login/oauth/access_token',
            data={
                'client_id': settings.GITHUB_CLIENT_ID,
                'client_secret': settings.GITHUB_CLIENT_SECRET,
                'code': code,
            },
            headers={'Accept': 'application/json'},
        )
        token_data = token_response.json()
        access_token = token_data.get('access_token')
        
        if not access_token:
            return Response({'error': '获取 access_token 失败'}, status=status.HTTP_400_BAD_REQUEST)
        
        # 2. 获取 GitHub 用户信息
        user_response = requests.get(
            'https://api.github.com/user',
            headers={
                'Authorization': f'token {access_token}',
                'Accept': 'application/json',
            },
        )
        github_user = user_response.json()
        
        # 3. 获取用户邮箱
        email_response = requests.get(
            'https://api.github.com/user/emails',
            headers={'Authorization': f'token {access_token}'},
        )
        emails = email_response.json()
        primary_email = next(
            (e['email'] for e in emails if e.get('primary')),
            emails[0]['email'] if emails else None,
        )
        
        # 4. 查找或创建用户
        github_id = str(github_user['id'])
        
        try:
            user = User.objects.get(github_id=github_id)
        except User.DoesNotExist:
            # 创建新用户
            username = github_user.get('login', f'github_{github_id}')
            user = User.objects.create(
                username=username,
                email=primary_email or '',
                github_id=github_id,
            )
        
        # 5. 生成 JWT Token
        refresh = RefreshToken.for_user(user)
        
        return Response({
            'user': UserSerializer(user).data,
            'access': str(refresh.access_token),
            'refresh': str(refresh),
        })

---

3.4 文件上传与静态资源管理

定义

文件上传 是 Web 应用中常见的需求（头像、图片、附件等），Django 通过 FileField 和 ImageField 字段类型提供便捷的文件存储方案。静态资源 包括 CSS、JavaScript、图片等前端文件，Django 通过 staticfiles 应用统一管理。

文件上传配置

# settings.py

import os

# 文件上传路径
MEDIA_URL = '/media/'                     # URL 前缀
MEDIA_ROOT = BASE_DIR / 'media'           # 文件存储的本地路径

# 文件上传大小限制（在 Nginx 层也需配置）
DATA_UPLOAD_MAX_MEMORY_SIZE = 5242880     # 5MB（内存中处理的最大大小）
FILE_UPLOAD_MAX_MEMORY_SIZE = 5242880     # 5MB
FILE_UPLOAD_PERMISSIONS = 0o644           # 上传文件的权限
FILE_UPLOAD_DIRECTORY_PERMISSIONS = 0o755 # 上传目录的权限

# myproject/urls.py —— 开发环境提供媒体文件访问

from django.conf import settings
from django.conf.urls.static import static

urlpatterns = [
    # ... 其他路由
]

if settings.DEBUG:
    urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

▶ 模型中使用文件字段

from django.db import models

class Article(models.Model):
    # ...
    cover_image = models.ImageField(
        upload_to='articles/covers/',          # 上传到 media/articles/covers/
        blank=True,
        null=True,
        verbose_name='封面图片',
        help_text='建议尺寸：1200×630像素',
    )
    attachment = models.FileField(
        upload_to='articles/attachments/',
        blank=True,
        null=True,
        verbose_name='附件',
    )

class User(AbstractUser):
    avatar = models.ImageField(
        upload_to='avatars/%Y/%m/%d/',         # 按日期分目录
        default='avatars/default.png',
        verbose_name='头像',
    )

▶ 自定义文件上传路径

# blog/utils.py —— 自定义上传路径

import os
import uuid

def user_avatar_path(instance, filename):
    """根据用户 ID 生成头像存储路径"""
    ext = filename.split('.')[-1]
    new_filename = f'{uuid.uuid4().hex}.{ext}'
    return f'avatars/user_{instance.user.id}/{new_filename}'

def article_cover_path(instance, filename):
    """根据文章 ID 生成封面图存储路径"""
    ext = filename.split('.')[-1]
    new_filename = f'cover_{instance.id}_{uuid.uuid4().hex[:8]}.{ext}'
    return f'articles/{instance.id}/{new_filename}'


# 在模型中使用
class Article(models.Model):
    cover_image = models.ImageField(
        upload_to=article_cover_path,  # 使用函数
        blank=True,
        null=True,
    )

▶ 文件上传 API

# blog/serializers.py —— 文件上传序列化器

from rest_framework import serializers

class FileUploadSerializer(serializers.Serializer):
    file = serializers.FileField()
    description = serializers.CharField(required=False, max_length=255)

class ImageUploadSerializer(serializers.Serializer):
    image = serializers.ImageField()
    alt_text = serializers.CharField(required=False, max_length=255)


# blog/views.py —— 文件上传视图

from rest_framework.parsers import MultiPartParser, FormParser
from rest_framework import status

class FileUploadView(APIView):
    """通用文件上传接口"""
    parser_classes = [MultiPartParser, FormParser]
    permission_classes = [IsAuthenticated]
    
    def post(self, request, format=None):
        file = request.FILES.get('file')
        
        if not file:
            return Response({'error': '请选择文件'}, status=status.HTTP_400_BAD_REQUEST)
        
        # 校验文件大小（5MB）
        if file.size > 5 * 1024 * 1024:
            return Response({'error': '文件大小不能超过 5MB'}, status=status.HTTP_400_BAD_REQUEST)
        
        # 校验文件类型
        allowed_types = ['image/jpeg', 'image/png', 'image/gif', 'application/pdf']
        if file.content_type not in allowed_types:
            return Response({'error': '不支持的文件类型'}, status=status.HTTP_400_BAD_REQUEST)
        
        # 保存文件到本地或云存储
        from django.core.files.storage import default_storage
        from django.core.files.base import ContentFile
        
        path = default_storage.save(f'uploads/{file.name}', ContentFile(file.read()))
        file_url = default_storage.url(path)
        
        return Response({
            'url': file_url,
            'name': file.name,
            'size': file.size,
            'content_type': file.content_type,
        }, status=status.HTTP_201_CREATED)


class AvatarUploadView(APIView):
    """用户头像上传"""
    parser_classes = [MultiPartParser]
    permission_classes = [IsAuthenticated]
    
    def post(self, request):
        user = request.user
        avatar = request.FILES.get('avatar')
        
        if not avatar:
            return Response({'error': '请选择头像图片'}, status=400)
        
        # 校验图片
        if avatar.size > 2 * 1024 * 1024:
            return Response({'error': '头像图片不能超过 2MB'}, status=400)
        
        # 保存头像
        user.avatar = avatar
        user.save()
        
        return Response({
            'avatar_url': user.avatar.url,
            'message': '头像更新成功',
        })

静态资源管理

# settings.py —— 静态资源配置

# 开发环境
STATIC_URL = '/static/'                                 # URL 前缀
STATICFILES_DIRS = [BASE_DIR / 'static']                # 开发时的静态文件目录
STATIC_ROOT = BASE_DIR / 'staticfiles'                  # 收集后的静态文件目录（生产）

# 可选：使用 Whitenoise 为生产环境提供静态文件服务
# pip install whitenoise
MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'whitenoise.middleware.WhiteNoiseMiddleware',  # 放在 SecurityMiddleware 之后
    # ...
]

# Whitenoise 配置
STATICFILES_STORAGE = 'whitenoise.storage.CompressedManifestStaticFilesStorage'

# 收集静态文件（部署时执行）
python manage.py collectstatic

# 收集结果示例
# staticfiles/
# ├── admin/
# │   ├── css/
# │   ├── js/
# │   └── ...
# ├── css/
# │   └── style.css
# └── js/
#     └── main.js

<!-- 模板中使用静态文件 -->
{% load static %}

<!DOCTYPE html>
<html>
<head>
    <link rel="stylesheet" href="{% static 'css/style.css' %}">
    <link rel="icon" href="{% static 'favicon.ico' %}">
</head>
<body>
    <img src="{% static 'images/logo.png' %}" alt="Logo">
    <script src="{% static 'js/main.js' %}"></script>
</body>
</html>

---

3.5 Admin 后台定制化开发

定义

Django Admin 是 Django 自动生成的后台管理界面，基于模型注册实现数据的增删改查。通过定制 ModelAdmin，可以深度控制后台的展示、过滤、搜索、编辑等功能，构建功能完善的后台管理系统。

语法

▶ 基础注册

# blog/admin.py

from django.contrib import admin
from .models import Article, Category, Tag, Comment

# 简单注册
admin.site.register(Tag)

# 使用注册装饰器
@admin.register(Article)
class ArticleAdmin(admin.ModelAdmin):
    pass

@admin.register(Category)
class CategoryAdmin(admin.ModelAdmin):
    pass

▶ ModelAdmin 核心配置

@admin.register(Article)
class ArticleAdmin(admin.ModelAdmin):
    """文章管理后台"""
    
    # --- 列表页配置 ---
    
    # 列表展示字段
    list_display = [
        'id', 'title', 'category', 'author', 
        'views', 'is_published', 'pub_date',
    ]
    
    # 可点击进入编辑页的字段
    list_display_links = ['title']
    
    # 右侧过滤栏
    list_filter = [
        'is_published', 
        'category', 
        'pub_date',
        ('author', admin.RelatedOnlyFieldListFilter),  # 仅显示有文章的作者的过滤
    ]
    
    # 顶部搜索框
    search_fields = ['title', 'content', 'author__username']
    
    # 列表页可编辑字段（不进入编辑页即可修改）
    list_editable = ['is_published', 'views']
    
    # 每页显示数量
    list_per_page = 20
    
    # 日期层级导航（顶部）
    date_hierarchy = 'pub_date'
    
    # 排序（覆盖模型的 Meta.ordering）
    ordering = ['-pub_date']
    
    # 列表页批量操作
    actions = ['make_published', 'make_unpublished', 'export_as_csv']
    
    # --- 编辑页配置 ---
    
    # 字段分组
    fieldsets = [
        ('基本信息', {
            'fields': ['title', 'category', 'author', 'tags'],
            'classes': ['wide'],  # 额外 CSS 类
        }),
        ('内容', {
            'fields': ['content', 'cover_image'],
            'classes': ['collapse'],  # 可折叠
        }),
        ('发布设置', {
            'fields': ['is_published', 'pub_date', 'views'],
            'classes': ['collapse'],
            'description': '文章的发布相关设置',  # 分组说明
        }),
    ]
    
    # 水平排列的 M2M 字段（改善 ManyToMany 的 UI）
    filter_horizontal = ['tags']
    # filter_vertical = ['tags']  # 垂直排列
    
    # 外键下拉框改为搜索框（数据量大时）
    autocomplete_fields = ['author', 'category']
    
    # 只读字段
    readonly_fields = ['pub_date', 'update_date', 'views']
    
    # 预填充字段（根据标题自动生成 slug）
    prepopulated_fields = {'slug': ('title',)}
    
    # --- 方法字段 ---
    
    def comment_count(self, obj):
        """自定义显示字段：评论数量"""
        return obj.comments.count()
    comment_count.short_description = '评论数'
    comment_count.admin_order_field = 'comments__count'  # 可按此字段排序
    
    def colored_status(self, obj):
        """带颜色的状态显示"""
        if obj.is_published:
            return format_html('<span style="color:green">✓ 已发布</span>')
        return format_html('<span style="color:gray">✗ 草稿</span>')
    colored_status.short_description = '状态'
    colored_status.allow_tags = True
    
    # 在 list_display 中添加自定义字段
    list_display = ['id', 'title', 'category', 'author', 'views', 
                    'colored_status', 'comment_count', 'pub_date']
    
    # --- 批量操作 ---
    
    @admin.action(description='标记为已发布')
    def make_published(self, request, queryset):
        updated = queryset.update(is_published=True)
        self.message_user(request, f'已成功发布 {updated} 篇文章')
    
    @admin.action(description='标记为草稿')
    def make_unpublished(self, request, queryset):
        updated = queryset.update(is_published=False)
        self.message_user(request, f'已标记 {updated} 篇文章为草稿')
    
    @admin.action(description='导出为 CSV')
    def export_as_csv(self, request, queryset):
        import csv
        from django.http import HttpResponse
        
        response = HttpResponse(content_type='text/csv')
        response['Content-Disposition'] = 'attachment; filename="articles.csv"'
        
        writer = csv.writer(response)
        writer.writerow(['ID', '标题', '分类', '作者', '阅读量', '发布时间'])
        
        for article in queryset:
            writer.writerow([
                article.id, article.title,
                article.category.name, article.author.username,
                article.views, article.pub_date,
            ])
        
        return response
    
    # --- 权限控制 ---
    
    def get_queryset(self, request):
        """限制管理员只能看到自己创建的文章"""
        qs = super().get_queryset(request)
        if request.user.is_superuser:
            return qs
        return qs.filter(author=request.user)
    
    def save_model(self, request, obj, form, change):
        """保存时自动设置作者"""
        if not change:  # 创建时
            obj.author = request.user
        obj.save()
    
    # --- 内联编辑 ---
    
    class Media:
        """自定义 CSS/JS"""
        css = {
            'all': ('css/admin-custom.css',)
        }
        js = ('js/admin-custom.js',)

▶ 内联模型（InlineModelAdmin）

class CommentInline(admin.TabularInline):
    """评论内联（表格形式）"""
    model = Comment
    extra = 1  # 额外空白行数
    fields = ['author', 'content', 'is_approved', 'created_at']
    readonly_fields = ['created_at']
    autocomplete_fields = ['author']

class CommentInlineStacked(admin.StackedInline):
    """评论内联（堆叠形式）"""
    model = Comment
    extra = 0
    fields = ['content', 'is_approved']

@admin.register(Article)
class ArticleAdmin(admin.ModelAdmin):
    inlines = [CommentInline]

@admin.register(Category)
class CategoryAdmin(admin.ModelAdmin):
    inlines = [CommentInlineStacked]

▶ 自定义 Admin 站点

# blog/admin.py —— 自定义 Admin 站点

from django.contrib.admin import AdminSite
from django.contrib import admin
from django.utils.translation import gettext_lazy as _

class CustomAdminSite(AdminSite):
    """自定义管理后台站点"""
    
    site_header = '博客管理系统'          # 页面标题
    site_title = '博客管理后台'           # 浏览器标签标题
    site_url = '/'                       # "查看站点"链接
    index_title = '管理首页'             # 首页标题
    empty_value_display = '-'            # 空值显示
    enable_nav_sidebar = True            # 启用侧边栏
    
    # 自定义首页
    def index(self, request, extra_context=None):
        context = {
            'custom_data': {
                'total_articles': Article.objects.count(),
                'published_articles': Article.objects.filter(is_published=True).count(),
                'total_users': get_user_model().objects.count(),
                'total_comments': Comment.objects.count(),
            },
        }
        if extra_context:
            context.update(extra_context)
        return super().index(request, extra_context=context)

# 创建实例
custom_admin_site = CustomAdminSite(name='custom_admin')

# 使用自定义站点注册
custom_admin_site.register(Article, ArticleAdmin)
custom_admin_site.register(Category, CategoryAdmin)
custom_admin_site.register(Tag)
custom_admin_site.register(Comment, CommentAdmin)

# urls.py
# from blog.admin import custom_admin_site
# urlpatterns = [
#     path('admin/', custom_admin_site.urls),
# ]

▶ Admin 仪表盘增强

# blog/admin.py —— 首页添加自定义看板

from django.db.models import Count
from django.utils import timezone
from datetime import timedelta

@admin.register(Article)
class ArticleAdmin(admin.ModelAdmin):
    # ... 其他配置
    
    def changelist_view(self, request, extra_context=None):
        """列表页添加统计信息"""
        context = {
            'stats': {
                'total': Article.objects.count(),
                'published': Article.objects.filter(is_published=True).count(),
                'draft': Article.objects.filter(is_published=False).count(),
                'today': Article.objects.filter(
                    pub_date__date=timezone.now().date()
                ).count(),
                'week': Article.objects.filter(
                    pub_date__gte=timezone.now() - timedelta(days=7)
                ).count(),
            },
            'top_categories': Category.objects.annotate(
                article_count=Count('articles')
            ).order_by('-article_count')[:5],
        }
        if extra_context:
            context.update(extra_context)
        return super().changelist_view(request, extra_context=context)

▶ 完整示例：文章管理后台

from django.contrib import admin
from django.utils.html import format_html
from django.db.models import Count
from .models import Article, Category, Tag, Comment


class CommentInline(admin.TabularInline):
    model = Comment
    extra = 0
    fields = ['author', 'content_short', 'is_approved', 'created_at']
    readonly_fields = ['created_at']
    autocomplete_fields = ['author']
    
    def content_short(self, obj):
        return obj.content[:50] + '...' if len(obj.content) > 50 else obj.content
    content_short.short_description = '评论内容'


@admin.register(Article)
class ArticleAdmin(admin.ModelAdmin):
    list_display = [
        'id', 'title', 'category', 'author', 'views',
        'cover_preview', 'status_display', 'comment_count', 'pub_date',
    ]
    list_display_links = ['title']
    list_filter = ['is_published', 'category', 'pub_date', 'author']
    search_fields = ['title', 'content']
    list_per_page = 20
    date_hierarchy = 'pub_date'
    ordering = ['-pub_date']
    filter_horizontal = ['tags']
    autocomplete_fields = ['author', 'category']
    readonly_fields = ['pub_date', 'update_date', 'preview_link']
    prepopulated_fields = {'slug': ('title',)}
    inlines = [CommentInline]
    
    fieldsets = [
        ('基本信息', {
            'fields': ['title', 'slug', 'category', 'author', 'tags'],
        }),
        ('内容', {
            'fields': ['content', 'cover_image'],
        }),
        ('发布设置', {
            'fields': ['is_published', 'views', 'pub_date'],
            'classes': ['collapse'],
        }),
    ]
    
    def cover_preview(self, obj):
        if obj.cover_image:
            return format_html(
                '<img src="{}" style="width:80px;height:45px;object-fit:cover;border-radius:4px;" />',
                obj.cover_image.url,
            )
        return format_html('<span style="color:gray">无封面</span>')
    cover_preview.short_description = '封面'
    
    def status_display(self, obj):
        if obj.is_published:
            return format_html(
                '<span style="color:#28a745;font-weight:bold;">✓ 已发布</span>'
            )
        return format_html(
            '<span style="color:#6c757d;">✗ 草稿</span>'
        )
    status_display.short_description = '状态'
    
    def comment_count(self, obj):
        count = obj.comments.count()
        url = f'/admin/blog/comment/?article__id__exact={obj.id}'
        return format_html('<a href="{}">{} 条</a>', url, count)
    comment_count.short_description = '评论'
    
    def preview_link(self, obj):
        if obj.pk:
            return format_html(
                '<a href="/blog/{}/" target="_blank">预览文章 →</a>', obj.pk
            )
        return '-'
    preview_link.short_description = '预览'
    
    @admin.action(description='批量发布选中文章')
    def make_published(self, request, queryset):
        updated = queryset.update(is_published=True)
        self.message_user(request, f'已发布 {updated} 篇文章')
    
    @admin.action(description='批量取消发布')
    def make_unpublished(self, request, queryset):
        updated = queryset.update(is_published=False)
        self.message_user(request, f'已将 {updated} 篇文章设为草稿')
    
    actions = ['make_published', 'make_unpublished']
    
    def save_model(self, request, obj, form, change):
        if not change:
            obj.author = request.user
        super().save_model(request, obj, form, change)
    
    def get_queryset(self, request):
        qs = super().get_queryset(request)
        if not request.user.is_superuser:
            qs = qs.filter(author=request.user)
        return qs

---

四、小结

知识点	核心要点
JWT 认证	三部分结构（Header.Payload.Signature）；access_token 短时效、refresh_token 长时效；djangorestframework-simplejwt 实现签发/刷新/验证/黑名单
RBAC 权限模型	用户 → 组（角色） → 权限；Django 内置 Permission/Group 模型；@permission_required 和 PermissionRequiredMixin 检查权限
用户系统扩展	AbstractUser 扩展（推荐，需在首次 migrate 前设置）；OneToOneField 扩展（已有项目）；django-allauth 实现第三方登录
文件上传与静态资源	ImageField/FileField 文件字段；MEDIA 配置处理用户上传；collectstatic 收集静态文件；Whitenoise 生产环境静态文件服务
Admin 后台定制	list_display/list_filter/search_fields 列表配置；fieldsets 编辑页分组；inlines 内联编辑；自定义动作和权限控制

五、练习

基础练习

1. JWT 认证：为博客 API 集成 JWT 认证，配置 access_token 15 分钟有效、refresh_token 7 天有效，并实现 Token 自动刷新
2. Admin 配置：将文章管理后台配置完善，包含列表展示、搜索过滤、字段分组和批量操作

进阶练习

3. RBAC 权限：创建"管理员"、"编辑"、"作者"三个角色组，分配对应权限，并在 DRF 视图集中实现基于角色的访问控制
4. 用户注册：实现完整的用户注册 API，包含邮箱、手机号、密码验证，注册成功后自动返回 JWT Token

挑战练习

5. 集成三方登录：
   • 注册 GitHub OAuth App，获取 Client ID 和 Secret
   • 实现 GitHub OAuth2 登录流程
   • 登录成功返回 JWT Token
   • 如果是新用户，自动创建账号并绑定 GitHub ID

参考代码（练习 1 JWT 配置）

SIMPLE_JWT = {
    'ACCESS_TOKEN_LIFETIME': timedelta(minutes=15),
    'REFRESH_TOKEN_LIFETIME': timedelta(days=7),
    'ROTATE_REFRESH_TOKENS': True,
    'BLACKLIST_AFTER_ROTATION': True,
}