每天40分玩转Django：Django中间件

Django中间件

一、今日学习内容概述

二、Django中间件基础

2.1 中间件工作原理

中间件是Django请求/响应处理的钩子框架。它是一个轻量级的、底层的插件系统，用于全局修改Django的输入或输出。每个中间件组件负责执行某些特定的功能。

2.2 中间件方法

class MiddlewareMixin:
    def __init__(self, get_response=None):
        self.get_response = get_response
        
    def __call__(self, request):
        response = None
        if hasattr(self, 'process_request'):
            response = self.process_request(request)
        response = response or self.get_response(request)
        if hasattr(self, 'process_response'):
            response = self.process_response(request, response)
        return response

中间件类可以实现以下方法：

1. process_request(request)
2. process_view(request, view_func, view_args, view_kwargs)
3. process_template_response(request, response)
4. process_exception(request, exception)
5. process_response(request, response)

三、自定义中间件示例

3.1 请求时间统计中间件

# middleware/timing.py
import time
from django.utils.deprecation import MiddlewareMixin
from django.core.cache import cache

class RequestTimingMiddleware(MiddlewareMixin):
    def process_request(self, request):
        request.start_time = time.time()
    
    def process_response(self, request, response):
        if hasattr(request, 'start_time'):
            total_time = time.time() - request.start_time
            response['X-Request-Time'] = f'{total_time:.2f}s'
            
            # 记录慢请求
            if total_time > 1.0:  # 超过1秒的请求
                cache.set(
                    f'slow_request_{request.path}_{time.time()}',
                    {
                        'path': request.path,
                        'method': request.method,
                        'time': total_time,
                        'user': str(request.user),
                    },
                    timeout=86400  # 24小时过期
                )
        return response

3.2 IP访问限制中间件

# middleware/ip_restriction.py
from django.core.exceptions import PermissionDenied
from django.core.cache import cache
from django.conf import settings

class IPRestrictionMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        # 白名单IP列表
        self.allowed_ips = getattr(settings, 'ALLOWED_IPS', [])
        
    def __call__(self, request):
        ip = self.get_client_ip(request)
        
        # 检查IP访问频率
        if self.is_ip_banned(ip):
            raise PermissionDenied('访问频率过高，请稍后再试')
            
        # 记录IP访问次数
        self.record_ip_request(ip)
        
        response = self.get_response(request)
        return response
    
    def get_client_ip(self, request):
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            ip = x_forwarded_for.split(',')[0]
        else:
            ip = request.META.get('REMOTE_ADDR')
        return ip
    
    def is_ip_banned(self, ip):
        if ip in self.allowed_ips:
            return False
            
        # 获取IP的访问次数
        key = f'ip_requests_{ip}'
        requests = cache.get(key, 0)
        
        # 如果1分钟内访问超过100次，则禁止访问
        return requests > 100
    
    def record_ip_request(self, ip):
        key = f'ip_requests_{ip}'
        timeout = 60  # 1分钟过期
        
        # 原子操作增加计数
        cache.add(key, 0, timeout)
        cache.incr(key)

3.3 用户行为日志中间件

# middleware/user_logging.py
import json
import logging
from django.utils.deprecation import MiddlewareMixin

logger = logging.getLogger('user_behavior')

class UserBehaviorMiddleware(MiddlewareMixin):
    def process_request(self, request):
        request.behavior_log = {
            'path': request.path,
            'method': request.method,
            'user_agent': request.META.get('HTTP_USER_AGENT', ''),
            'ip': self.get_client_ip(request),
            'user': str(request.user),
            'data': {}
        }
        
        # 记录请求数据
        if request.method in ['POST', 'PUT', 'PATCH']:
            try:
                request.behavior_log['data']['body'] = json.loads(request.body)
            except:
                request.behavior_log['data']['body'] = str(request.POST)
                
        if request.method == 'GET':
            request.behavior_log['data']['query'] = dict(request.GET)
    
    def process_response(self, request, response):
        if hasattr(request, 'behavior_log'):
            # 添加响应状态码
            request.behavior_log['status_code'] = response.status_code
            
            # 记录日志
            logger.info(
                'User Behavior',
                extra={'behavior': request.behavior_log}
            )
        return response
    
    def get_client_ip(self, request):
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            ip = x_forwarded_for.split(',')[0]
        else:
            ip = request.META.get('REMOTE_ADDR')
        return ip

四、中间件配置与使用

# 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',
    # 自定义中间件
    'middleware.timing.RequestTimingMiddleware',
    'middleware.ip_restriction.IPRestrictionMiddleware',
    'middleware.user_logging.UserBehaviorMiddleware',
]

五、中间件执行流程

六、高级应用场景

6.1 API认证中间件

# middleware/api_auth.py
import jwt
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.exceptions import PermissionDenied

User = get_user_model()

class JWTAuthMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        if request.path.startswith('/api/'):
            auth_header = request.META.get('HTTP_AUTHORIZATION', '')
            if not auth_header.startswith('Bearer '):
                raise PermissionDenied('无效的认证头')
            
            try:
                token = auth_header.split(' ')[1]
                payload = jwt.decode(
                    token,
                    settings.SECRET_KEY,
                    algorithms=['HS256']
                )
                user = User.objects.get(id=payload['user_id'])
                request.user = user
            except (jwt.InvalidTokenError, User.DoesNotExist):
                raise PermissionDenied('无效的token')
        
        response = self.get_response(request)
        return response

6.2 多语言支持中间件

# middleware/language.py
from django.conf import settings
from django.utils.translation import activate
from django.utils.deprecation import MiddlewareMixin

class LanguageMiddleware(MiddlewareMixin):
    def process_request(self, request):
        # 从请求头获取语言设置
        language = request.META.get('HTTP_ACCEPT_LANGUAGE', '')
        if language:
            language = language.split(',')[0].split(';')[0]
        
        # 检查是否支持该语言
        if language in [lang[0] for lang in settings.LANGUAGES]:
            activate(language)
        else:
            activate(settings.LANGUAGE_CODE)

6.3 异常处理中间件

# middleware/error_handler.py
import traceback
import logging
from django.http import JsonResponse
from django.conf import settings

logger = logging.getLogger('django.request')

class ErrorHandlerMiddleware:
    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):
        if settings.DEBUG:
            return None
            
        # 记录错误日志
        logger.error(
            'Unhandled Exception',
            exc_info=(type(exception), exception, exception.__traceback__),
            extra={
                'status_code': 500,
                'request': request,
            }
        )
        
        # API请求返回JSON响应
        if request.path.startswith('/api/'):
            return JsonResponse({
                'error': 'Internal Server Error',
                'message': str(exception)
            }, status=500)
        
        # 普通请求重定向到错误页面
        from django.shortcuts import render
        return render(request, '500.html', status=500)

七、性能注意事项

1. 中间件执行顺序优化

   • 将最常用的中间件放在前面
   • 将耗时的操作放在必要的时候执行

2. 缓存使用

   • 合理使用缓存避免重复计算
   • 设置适当的缓存过期时间

3. 异步处理

   • 将耗时操作放在异步任务中处理
   • 使用消息队列处理日志记录等操作

八、测试中间件

# tests/test_middleware.py
from django.test import TestCase, Client
from django.urls import reverse
from django.core.cache import cache

class RequestTimingMiddlewareTests(TestCase):
    def setUp(self):
        self.client = Client()
    
    def test_timing_header(self):
        response = self.client.get(reverse('home'))
        self.assertTrue('X-Request-Time' in response.headers)
        
class IPRestrictionMiddlewareTests(TestCase):
    def setUp(self):
        self.client = Client()
        cache.clear()
    
    def test_ip_restriction(self):
        # 模拟频繁访问
        for _ in range(101):
            self.client.get(reverse('home'))
        
        # 下一个请求应该被禁止
        response = self.client.get(reverse('home'))
        self.assertEqual(response.status_code, 403)

九、最佳实践建议

1. 中间件设计原则

   • 单一职责原则
   • 保持简单，每个中间件只做一件事
   • 适当的错误处理和日志记录

2. 代码组织

   • 将相关的中间件放在同一个包中
   • 使用清晰的命名约定
   • 编写完整的文档注释

3. 性能优化

   • 避免在每个请求中执行重复操作
   • 合理使用缓存
   • 监控中间件性能影响

4. 安全考虑

   • 注意敏感信息的处理
   • 实现适当的访问控制
   • 防止中间件被滥用

十、总结

通过今天的学习，我们掌握了：

1. Django中间件的工作原理和执行流程
2. 如何开发自定义中间件
3. 中间件的常见应用场景
4. 中间件性能优化和测试方法

怎么样今天的内容还满意吗？再次感谢朋友们的观看，关注GZH：凡人的AI工具箱，回复666，送您价值199的AI大礼包。最后，祝您早日实现财务自由，还请给个赞，谢谢！