跳到主要内容

中间件

中间件是 Django 请求/响应处理过程中的钩子框架，它可以在请求到达视图之前或响应返回客户端之前执行自定义代码。

中间件工作原理​

中间件按照定义的顺序依次处理请求，然后按照相反的顺序处理响应：

内置中间件​

Django 提供了多种内置中间件：

# settings.py
MIDDLEWARE = [
    # 安全中间件
    'django.middleware.security.SecurityMiddleware',
    
    # 会话中间件
    'django.contrib.sessions.middleware.SessionMiddleware',
    
    # CSRF 保护
    'django.middleware.csrf.CsrfViewMiddleware',
    
    # 认证中间件
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    
    # 消息中间件
    'django.contrib.messages.middleware.MessageMiddleware',
    
    # 点击劫持保护
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
    
    # 通用中间件
    'django.middleware.common.CommonMiddleware',
]

SecurityMiddleware​

提供安全相关的设置：

# settings.py
SECURE_SSL_REDIRECT = True              # 强制 HTTPS
SECURE_HSTS_SECONDS = 31536000          # HSTS 时间
SECURE_HSTS_INCLUDE_SUBDOMAINS = True   # 包含子域名
SECURE_HSTS_PRELOAD = True              # 预加载
SECURE_CONTENT_TYPE_NOSNIFF = True      # 防止 MIME 类型嗅探
SECURE_BROWSER_XSS_FILTER = True        # XSS 过滤
X_FRAME_OPTIONS = 'DENY'                # 防止点击劫持

SessionMiddleware​

管理用户会话：

# settings.py
SESSION_ENGINE = 'django.contrib.sessions.backends.db'  # 会话引擎
SESSION_COOKIE_NAME = 'sessionid'                       # Cookie 名称
SESSION_COOKIE_SECURE = True                            # 仅 HTTPS
SESSION_COOKIE_HTTPONLY = True                          # 防止 JS 访问
SESSION_EXPIRE_AT_BROWSER_CLOSE = True                  # 浏览器关闭时过期

CsrfViewMiddleware​

提供 CSRF 保护：

# 视图中豁免 CSRF
from django.views.decorators.csrf import csrf_exempt, csrf_protect


@csrf_exempt
def api_view(request):
    """豁免 CSRF 保护"""
    pass


@csrf_protect
def protected_view(request):
    """强制 CSRF 保护"""
    pass

自定义中间件​

函数式中间件​

# middleware.py
def simple_middleware(get_response):
    """简单的中间件"""
    
    # 初始化代码（服务器启动时执行一次）
    
    def middleware(request):
        # 请求处理前的代码
        
        response = get_response(request)  # 调用视图
        
        # 响应处理后的代码
        
        return response
    
    return middleware


# 完整示例
def timing_middleware(get_response):
    """记录请求处理时间"""
    import time
    
    def middleware(request):
        start_time = time.time()
        
        response = get_response(request)
        
        duration = time.time() - start_time
        response['X-Request-Duration'] = f'{duration:.3f}s'
        
        return response
    
    return middleware

类式中间件​

# middleware.py
class SimpleMiddleware:
    """类式中间件"""
    
    def __init__(self, get_response):
        self.get_response = get_response
        # 初始化代码
    
    def __call__(self, request):
        # 请求处理前的代码
        
        response = self.get_response(request)
        
        # 响应处理后的代码
        
        return response


# 完整示例
class RequestLoggingMiddleware:
    """请求日志中间件"""
    
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        import logging
        logger = logging.getLogger(__name__)
        
        # 记录请求
        logger.info(f'Request: {request.method} {request.path}')
        
        response = self.get_response(request)
        
        # 记录响应
        logger.info(f'Response: {response.status_code}')
        
        return response

钩子方法​

中间件可以定义特定的钩子方法：

class FullMiddleware:
    """完整的中间件示例"""
    
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        response = self.get_response(request)
        return response
    
    def process_request(self, request):
        """处理请求（在视图之前）"""
        # 返回 None 继续处理
        # 返回 HttpResponse 则中断处理
        pass
    
    def process_view(self, request, view_func, view_args, view_kwargs):
        """处理视图（在视图之前，但在 process_request 之后）"""
        pass
    
    def process_template_response(self, request, response):
        """处理模板响应"""
        return response
    
    def process_response(self, request, response):
        """处理响应（在视图之后）"""
        return response
    
    def process_exception(self, request, exception):
        """处理异常"""
        # 返回 None 让异常继续传播
        # 返回 HttpResponse 则处理异常
        pass

常用中间件示例​

IP 黑名单中间件​

class IPBlacklistMiddleware:
    """IP 黑名单中间件"""
    
    BLACKLIST = [
        '192.168.1.100',
        '10.0.0.50',
    ]
    
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        ip = self.get_client_ip(request)
        
        if ip in self.BLACKLIST:
            from django.http import HttpResponseForbidden
            return HttpResponseForbidden('访问被拒绝')
        
        return self.get_response(request)
    
    def get_client_ip(self, request):
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            return x_forwarded_for.split(',')[0]
        return request.META.get('REMOTE_ADDR')

用户活动记录中间件​

class UserActivityMiddleware:
    """记录用户活动"""
    
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        if request.user.is_authenticated:
            from django.utils import timezone
            request.user.last_activity = timezone.now()
            request.user.save(update_fields=['last_activity'])
        
        return self.get_response(request)

请求限流中间件​

from django.core.cache import cache
from django.http import HttpResponseTooManyRequests


class RateLimitMiddleware:
    """请求限流中间件"""
    
    def __init__(self, get_response):
        self.get_response = get_response
        self.rate_limit = 100  # 每分钟最多 100 次请求
        self.window = 60       # 时间窗口（秒）
    
    def __call__(self, request):
        ip = self.get_client_ip(request)
        key = f'rate_limit:{ip}'
        
        # 获取当前请求次数
        count = cache.get(key, 0)
        
        if count >= self.rate_limit:
            return HttpResponseTooManyRequests('请求过于频繁，请稍后再试')
        
        # 增加计数
        cache.set(key, count + 1, self.window)
        
        return self.get_response(request)
    
    def get_client_ip(self, request):
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            return x_forwarded_for.split(',')[0]
        return request.META.get('REMOTE_ADDR')

异常处理中间件​

class ExceptionHandlerMiddleware:
    """统一异常处理"""
    
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        return self.get_response(request)
    
    def process_exception(self, request, exception):
        import logging
        logger = logging.getLogger(__name__)
        
        logger.exception(f'Unhandled exception: {exception}')
        
        from django.http import JsonResponse
        
        if request.path.startswith('/api/'):
            return JsonResponse({
                'error': '服务器内部错误',
                'detail': str(exception) if settings.DEBUG else None,
            }, status=500)
        
        return None

CORS 中间件​

class CORSMiddleware:
    """跨域资源共享中间件"""
    
    def __init__(self, get_response):
        self.get_response = get_response
        self.allowed_origins = [
            'http://localhost:3000',
            'https://example.com',
        ]
    
    def __call__(self, request):
        response = self.get_response(request)
        
        origin = request.META.get('HTTP_ORIGIN')
        
        if origin in self.allowed_origins:
            response['Access-Control-Allow-Origin'] = origin
            response['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS'
            response['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
            response['Access-Control-Allow-Credentials'] = 'true'
        
        # 处理预检请求
        if request.method == 'OPTIONS':
            response.status_code = 200
        
        return response

注册中间件​

# settings.py
MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
    
    # 自定义中间件
    'myapp.middleware.RequestLoggingMiddleware',
    'myapp.middleware.RateLimitMiddleware',
    'myapp.middleware.IPBlacklistMiddleware',
]

中间件执行顺序​

中间件的执行顺序非常重要：

  1. 请求阶段：按照 MIDDLEWARE 列表从上到下执行
  2. 响应阶段：按照 MIDDLEWARE 列表从下到上执行
# 请求顺序
SecurityMiddleware → SessionMiddleware → CommonMiddleware → ... → View

# 响应顺序
View → ... → CommonMiddleware → SessionMiddleware → SecurityMiddleware

中间件最佳实践​

1. 保持中间件简洁​

# 好的做法
class SimpleMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        # 单一职责
        self.log_request(request)
        return self.get_response(request)

# 不好的做法
class BloatedMiddleware:
    # 包含太多功能
    pass

2. 使用缓存优化性能​

class CachedMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.cache = {}  # 或使用 Django 缓存

3. 条件性处理​

class ConditionalMiddleware:
    def __call__(self, request):
        # 只处理特定路径
        if request.path.startswith('/api/'):
            # 处理 API 请求
            pass
        
        return self.get_response(request)

4. 异常处理​

class SafeMiddleware:
    def __call__(self, request):
        try:
            # 可能出错的操作
            pass
        except Exception as e:
            import logging
            logging.exception(e)
        
        return self.get_response(request)