Django 中间件是请求/响应处理链中的「插件」，在视图执行前后执行；认证系统则提供完整的用户管理功能。本文讲解两者的原理和使用方法。

中间件工作原理

Django 请求处理流程：

请求 → MIDDLEWARE_1.process_request() → MIDDLEWARE_2.process_request() → ...
                                                              ↓
                                                         视图函数
                                                              ↓
响应 ← MIDDLEWARE_1.process_response() ← MIDDLEWARE_2.process_response() ← ...

内置中间件

# myproject/settings.py
MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",   # 安全相关headers
    "django.contrib.sessions.middleware.SessionMiddleware",  # 会话管理
    "django.middleware.common.CommonMiddleware",        # URL处理、CSRF
    "django.middleware.csrf.CsrfViewMiddleware",        # CSRF保护
    "django.contrib.auth.middleware.AuthenticationMiddleware",  # 用户认证
    "django.contrib.messages.middleware.MessageMiddleware",  # 消息框架
    "django.middleware.clickjacking.XFrameOptionsMiddleware",  # X-Frame-Options
    "django.middleware.http.ConditionalGetMiddleware",  # 条件GET/ETag
]

自定义中间件

# blog/middleware.py
from django.shortcuts import redirect
from django.http import JsonResponse
from django.utils.timezone import now
from django.conf import settings
import time
class RequestTimerMiddleware:
    """记录每个请求的处理时间"""
    def __init__(self, get_response):
        self.get_response = get_response
def __call__(self, request):
        start = time.time()
        response = self.get_response(request)
        duration = time.time() - start
        response["X-Request-Duration"] = f"{duration:.3f}s"
        return response
class IPRateLimitMiddleware:
    """简单的 IP 频率限制"""
    def __init__(self, get_response):
        self.get_response = get_response
        self.request_counts = {}
        self.limit = 100  # 100 requests
        self.window = 60  # per 60 seconds
def __call__(self, request):
        ip = request.META.get("REMOTE_ADDR")
        now_time = time.time()
        self.request_counts.setdefault(ip, [])
        # 清理过期记录
        self.request_counts[ip] = [t for t in self.request_counts[ip] if now_time - t < self.window]
        if len(self.request_counts[ip]) >= self.limit:
            return JsonResponse({"error": "请求过于频繁"}, status=429)
        self.request_counts[ip].append(now_time)
        return self.get_response(request)
class LoginRequiredMiddleware:
    """强制登录中间件（排除白名单路径）"""
    def __init__(self, get_response):
        self.get_response = get_response
        self.exempt_urls = ["/admin/", "/api/health/", "/about/"]
def __call__(self, request):
        if not request.user.is_authenticated:
            for exempt in self.exempt_urls:
                if request.path.startswith(exempt):
                    return self.get_response(request)
            return redirect(f"/accounts/login/?next={request.path}")
        return self.get_response(request)
class VisitCounterMiddleware:
    """访客统计中间件"""
    def __init__(self, get_response):
        self.get_response = get_response
def process_view(self, request, view_func, view_args, view_kwargs):
        """在视图函数调用前执行"""
        request.visit_time = now()
        return None  # 返回 None 继续处理，返回 response 则短路
def process_template_response(self, request, response):
        """模板响应处理（可选）"""
        if hasattr(request, "visit_time"):
            response.context_data["visit_time"] = request.visit_time
        return response
settings.py 注册
MIDDLEWARE += ["blog.middleware.RequestTimerMiddleware", ...]</code></pre><h2>Django 认证系统</h2><pre><code># 内置 User 模型字段
username, password, email, first_name, last_name
is_active, is_staff, is_superuser, date_joined, last_login
from django.contrib.auth.models import User, Group, Permission
from django.contrib.auth import authenticate, login, logout, get_user_model
from django.contrib.auth.decorators import login_required, permission_required
from django.contrib.auth.mixins import LoginRequiredMixin, PermissionRequiredMixin
认证操作
def do_login(request):
    username = request.POST.get("username")
    password = request.POST.get("password")
    user = authenticate(request, username=username, password=password)
    if user is not None:
        login(request, user)  # 创建 session
        return redirect("dashboard")
    return render(request, "login.html", {"error": "用户名或密码错误"})
def do_logout(request):
    logout(request)  # 清除 session
    return redirect("home")
基于类的登录视图
from django.contrib.auth.views import LoginView, LogoutView, PasswordChangeView
from django.urls import reverse_lazy
class CustomLoginView(LoginView):
    template_name = "accounts/login.html"
    redirect_authenticated_user = True
    success_url = reverse_lazy("dashboard")
class CustomPasswordChangeView(PasswordChangeView):
    template_name = "accounts/password_change.html"
    success_url = reverse_lazy("password_change_done")

用户注册与权限管理

# blog/views.py
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group, Permission
from django.contrib.contenttypes.models import ContentType

User = get_user_model()
def register(request):
    if request.method == "POST":
        username = request.POST.get("username")
        email = request.POST.get("email")
        password = request.POST.get("password")
        password2 = request.POST.get("password2")
        if password != password2:
            return render(request, "register.html", {"error": "两次密码不一致"})
        if User.objects.filter(username=username).exists():
            return render(request, "register.html", {"error": "用户名已存在"})
        user = User.objects.create_user(username=username, email=email, password=password)
        login(request, user)
        return redirect("dashboard")
    return render(request, "register.html")
def assign_permissions(request, user_id):
    """给用户分配权限"""
    user = User.objects.get(id=user_id)
    # 方式1：直接分配权限
    user.user_permissions.add(
        Permission.objects.get(codename="change_article"),
        Permission.objects.get(codename="delete_article"),
    )
    # 方式2：分配到用户组
    editors = Group.objects.get(name="editors")
    user.groups.add(editors)
    # 检查权限
    if request.user.has_perm("blog.change_article"):
        pass  # 有编辑文章的权限
    return redirect("user_detail", user_id=user.id)
@login_required
@permission_required("blog.change_article", raise_exception=True)
def edit_article(request, article_id):
    """装饰器双重保护"""
    pass

扩展 User 模型

# blog/models.py
from django.contrib.auth.models import AbstractUser

class UserProfile(AbstractUser):
    """扩展默认 User 模型"""
    GENDER_CHOICES = [("M", "男"), ("F", "女"), ("O", "其他")]
    bio = models.TextField(blank=True, verbose_name="个人简介")
    gender = models.CharField(max_length=1, choices=GENDER_CHOICES, blank=True)
    birthday = models.DateField(null=True, blank=True)
    phone = models.CharField(max_length=20, blank=True)
    avatar = models.ImageField(upload_to="avatars/", null=True, blank=True)
    credit_score = models.IntegerField(default=100)  # 信用分
class Meta:
        verbose_name = "用户资料"
        verbose_name_plural = verbose_name
settings.py 指定自定义用户模型
AUTH_USER_MODEL = "blog.UserProfile"</code></pre><h2>常见问题</h2><ul><li><strong>Q: 中间件的 __init__、process_request、process_response 分别在什么时候执行？</strong><br>A：__init__ 在服务器启动时执行一次（做初始化）；每个请求执行一次 process_request（从顶到底）；视图返回 response 后执行 process_response（从底到顶）。</li><li><strong>Q: 多个中间件的 process_request 执行顺序是什么？</strong><br>A：按照 MIDDLEWARE 列表中的顺序，从上到下执行。AuthenticationMiddleware 在 SessionMiddleware 之后是因为它依赖 session 数据。</li><li><strong>Q: 自定义用户模型后，原有数据怎么办？</strong><br>A：必须在一开始项目时设定 AUTH_USER_MODEL，已经有数据后不能直接迁移。最佳实践是始终使用 get_user_model() 而不是直接引用 User。</li><li><strong>Q: 如何在中间件中访问数据库？</strong><br>A：中间件的 __init__ 中不要访问数据库（应用还未完全加载），在 __call__ 或 process_request 中可以安全访问。异步中间件需特别注意连接管理。</li></ul><h2>延伸阅读</h2><ul><li><a href="https://docs.djangoproject.com/zh-hans/5.0/topics/http/middleware/">Django 中间件官方文档</a></li><li><a href="https://docs.djangoproject.com/zh-hans/5.0/topics/auth/">Django 认证官方文档</a></li><li>Django REST Framework 认证机制详解</li></ul>