python学习体系

Appearance

章节3：视图进阶与系统功能 ​

一、章节概述 ​

本章在视图函数的基础上深入进阶，学习 Django 的类视图（CBV）、会话管理（Cookie/Session）、中间件机制、信号系统、缓存策略以及异步任务队列等高级功能。这些知识将帮助学生构建更健壮、可维护且高性能的 Django 应用。

二、学习目标 ​

目标分类具体目标
知识目标理解 FBV 与 CBV 的区别及各自适用场景；掌握 Cookie 与 Session 的工作原理；理解中间件信号机制及缓存策略
技能目标能熟练使用 ListView/DetailView 等通用类视图；能自定义中间件和信号处理器；能配置 Redis 缓存并集成 Celery 异步任务
素养目标培养面向对象的设计思维；理解企业级 Web 应用的系统架构考量

三、核心知识点 ​

3.1 FBV 与 CBV 类视图 ​

定义 ​

FBV（Function-Based View） 是以函数形式定义的视图，简单直观，适合处理简单逻辑。CBV（Class-Based View） 是以类形式定义的视图，通过继承 Django 内置的通用视图类，能够以更少的代码实现常见模式（列表展示、详情展示、表单处理等）。Django 提供了丰富的通用类视图：View、ListView、DetailView、CreateView、UpdateView、DeleteView、FormView 等。

语法 ​

▶ 基础类视图 View

python
from django.views import View
from django.http import HttpResponse, JsonResponse
from django.shortcuts import render, get_object_or_404

# FBV 方式
def article_list_fbv(request):
    if request.method == 'GET':
        articles = Article.objects.all()
        return render(request, 'articles.html', {'articles': articles})
    elif request.method == 'POST':
        # 处理创建
        pass

# CBV 方式（等效）
class ArticleListView(View):
    """类视图通过不同方法处理不同 HTTP 方法"""
    
    def get(self, request):
        """处理 GET 请求"""
        articles = Article.objects.all()
        return render(request, 'articles.html', {'articles': articles})
    
    def post(self, request):
        """处理 POST 请求"""
        # 创建文章
        title = request.POST.get('title')
        content = request.POST.get('content')
        article = Article.objects.create(title=title, content=content)
        return JsonResponse({'id': article.id, 'title': article.title})
    
    def dispatch(self, request, *args, **kwargs):
        """请求分发钩子（可选重写）"""
        # 在调用 get/post 之前执行的通用逻辑
        if not request.user.is_authenticated:
            return HttpResponse('请先登录', status=401)
        return super().dispatch(request, *args, **kwargs)

# URL 配置
# path('articles/', ArticleListView.as_view(), name='article_list')

▶ ListView —— 列表视图

python
from django.views.generic import ListView
from .models import Article

class ArticleListView(ListView):
    """通用列表视图：自动查询并渲染列表模板"""
    
    # 必须指定：查询的模型
    model = Article
    
    # 可选：模板名称（默认：<app>/<model>_list.html → blog/article_list.html）
    template_name = 'blog/article_list.html'
    
    # 可选：上下文变量名（默认：object_list 或 article_list）
    context_object_name = 'articles'
    
    # 可选：过滤条件
    queryset = Article.objects.filter(is_published=True)
    
    # 可选：每页数量（启用分页）
    paginate_by = 10
    
    # 可选：默认排序
    ordering = ['-pub_date']
    
    # 可选：添加额外上下文数据
    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context['categories'] = Category.objects.all()
        context['page_title'] = '文章列表'
        return context
    
    # 可选：自定义查询集
    def get_queryset(self):
        queryset = super().get_queryset()
        keyword = self.request.GET.get('q')
        if keyword:
            queryset = queryset.filter(title__icontains=keyword)
        return queryset

▶ DetailView —— 详情视图

python
from django.views.generic import DetailView
from django.db.models import F

class ArticleDetailView(DetailView):
    """通用详情视图：自动查询单条记录并渲染详情模板"""
    
    model = Article
    template_name = 'blog/article_detail.html'
    context_object_name = 'article'
    
    # URL 中的 PK 参数名（默认：pk）
    # pk_url_kwarg = 'pk'
    
    def get_object(self, queryset=None):
        """获取对象（可在此处增加自定义逻辑）"""
        article = super().get_object(queryset)
        # 阅读量 +1（使用 F 表达式避免并发问题）
        Article.objects.filter(pk=article.pk).update(views=F('views') + 1)
        article.refresh_from_db()  # 刷新以获取最新 views
        return article
    
    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        # 上一篇、下一篇
        current = self.object
        context['prev_article'] = Article.objects.filter(
            pub_date__lt=current.pub_date
        ).first()
        context['next_article'] = Article.objects.filter(
            pub_date__gt=current.pub_date
        ).last()
        return context

▶ CreateView / UpdateView / DeleteView

python
from django.views.generic import CreateView, UpdateView, DeleteView
from django.urls import reverse_lazy
from .forms import ArticleForm  # 需配合 Django 表单使用

class ArticleCreateView(CreateView):
    """创建视图：处理表单展示与提交"""
    model = Article
    form_class = ArticleForm          # 关联的表单类
    template_name = 'blog/article_form.html'
    success_url = reverse_lazy('blog:article_list')  # 成功后跳转
    
    def form_valid(self, form):
        """表单验证通过后的处理"""
        form.instance.author = self.request.user  # 设置作者为当前用户
        return super().form_valid(form)

class ArticleUpdateView(UpdateView):
    """更新视图：与 CreateView 类似，但预填充已有数据"""
    model = Article
    form_class = ArticleForm
    template_name = 'blog/article_form.html'
    
    def get_success_url(self):
        return reverse_lazy('blog:article_detail', kwargs={'pk': self.object.pk})

class ArticleDeleteView(DeleteView):
    """删除视图：需要 POST 确认删除"""
    model = Article
    template_name = 'blog/article_confirm_delete.html'
    success_url = reverse_lazy('blog:article_list')

▶ FormView 与表单处理

python
from django.views.generic import FormView
from django.contrib import messages
from .forms import ContactForm

class ContactFormView(FormView):
    """表单视图：展示表单 + 处理提交"""
    template_name = 'contact.html'
    form_class = ContactForm
    success_url = '/contact/success/'
    
    def form_valid(self, form):
        """表单验证通过"""
        form.send_email()  # 提交表单后发送邮件
        messages.success(self.request, '消息已发送！')
        return super().form_valid(form)
    
    def form_invalid(self, form):
        """表单验证失败"""
        messages.error(self.request, '请检查表单错误')
        return super().form_invalid(form)

▶ CBV 路由配置

python
# urls.py
from django.urls import path
from .views import (
    ArticleListView, ArticleDetailView,
    ArticleCreateView, ArticleUpdateView, ArticleDeleteView
)

app_name = 'blog'

urlpatterns = [
    path('', ArticleListView.as_view(), name='article_list'),
    path('<int:pk>/', ArticleDetailView.as_view(), name='article_detail'),
    path('create/', ArticleCreateView.as_view(), name='article_create'),
    path('<int:pk>/update/', ArticleUpdateView.as_view(), name='article_update'),
    path('<int:pk>/delete/', ArticleDeleteView.as_view(), name='article_delete'),
]

FBV vs CBV 对比 ​

对比维度FBVCBV
代码量简单场景少，复杂场景多继承通用视图，代码精简
复用性通过装饰器复用通过继承 Mixin 复用
HTTP 方法处理需手动判断 request.method自动分发到不同方法
学习曲线简单直观需要理解继承体系
灵活性完全控制受框架约定限制
适用场景简单或高度定制化的视图标准 CRUD 和列表展示

3.2 Cookie 与 Session 会话管理 ​

定义 ​

Cookie 是存储在客户端浏览器中的小数据块（通常 ≤4KB），由服务器通过 HTTP 响应头 Set-Cookie 设置，浏览器在后续请求中自动携带。Session 是存储在服务器端的用户会话数据，客户端仅保存 Session ID（通常存储在 Cookie 中），通过 Session ID 关联服务器端的数据。

语法 ​

▶ Cookie 操作

python
from django.http import HttpResponse

def set_cookie(request):
    """设置 Cookie"""
    response = HttpResponse('Cookie 已设置')
    
    # 基本设置
    response.set_cookie('username', 'alice')
    
    # 带过期时间的 Cookie（单位：秒）
    response.set_cookie('token', 'abcdef123', max_age=3600)  # 1 小时后过期
    
    # 带过期日期的 Cookie
    expires = datetime.datetime(2025, 12, 31, 23, 59, 59)
    response.set_cookie('preference', 'dark_mode', expires=expires)
    
    # 安全选项
    response.set_cookie(
        'session_id', 'xxx',
        max_age=86400,          # 24 小时
        httponly=True,          # 禁止 JavaScript 访问（防 XSS）
        secure=True,            # 仅 HTTPS 传输
        samesite='Lax',         # CSRF 防护
        path='/',               # 路径范围
        domain='.example.com',  # 域名范围
    )
    return response

def get_cookie(request):
    """读取 Cookie"""
    username = request.COOKIES.get('username', '匿名用户')
    preference = request.COOKIES.get('preference', 'light_mode')
    return HttpResponse(f'用户: {username}, 偏好: {preference}')

def delete_cookie(request):
    """删除 Cookie"""
    response = HttpResponse('Cookie 已删除')
    response.delete_cookie('username')
    response.delete_cookie('token')
    return response

▶ Session 配置与操作

python
# settings.py —— Session 配置
# Django 默认使用数据库存储 Session（需执行 migrate）
INSTALLED_APPS = [
    'django.contrib.sessions',  # 默认已包含
]

# Session 引擎配置
SESSION_ENGINE = 'django.contrib.sessions.backends.db'  # 数据库存储（默认）
# SESSION_ENGINE = 'django.contrib.sessions.backends.cache'  # 缓存存储
# SESSION_ENGINE = 'django.contrib.sessions.backends.cached_db'  # 缓存+数据库双写
# SESSION_ENGINE = 'django.contrib.sessions.backends.file'  # 文件存储

SESSION_COOKIE_NAME = 'sessionid'           # Cookie 名称（默认）
SESSION_COOKIE_AGE = 1209600                # Session 过期时间（秒，默认 2 周）
SESSION_EXPIRE_AT_BROWSER_CLOSE = False     # 关闭浏览器是否过期
SESSION_SAVE_EVERY_REQUEST = False          # 每次请求都保存 Session
python
# views.py —— Session 操作

def session_set(request):
    """存储 Session 数据"""
    # 会话数据可以像字典一样操作
    request.session['user_id'] = 1
    request.session['username'] = 'alice'
    request.session['roles'] = ['admin', 'editor']
    request.session['preferences'] = {
        'theme': 'dark',
        'language': 'zh-CN',
    }
    
    # 设置过期时间
    request.session.set_expiry(3600)  # 1 小时后过期
    # request.session.set_expiry(0)    # 关闭浏览器时过期
    # request.session.set_expiry(None) # 使用全局设置
    
    return HttpResponse('Session 数据已保存')

def session_get(request):
    """读取 Session 数据"""
    user_id = request.session.get('user_id')
    username = request.session.get('username', '未登录')
    
    # 安全获取并设置默认值
    roles = request.session.get('roles', [])
    
    # 检查键是否存在
    if 'preferences' in request.session:
        theme = request.session['preferences'].get('theme')
    else:
        theme = 'default'
    
    return HttpResponse(f'用户: {username}, 角色: {roles}, 主题: {theme}')

def session_delete(request):
    """删除 Session 数据"""
    # 删除单个键
    if 'preferences' in request.session:
        del request.session['preferences']
    
    # 清空所有数据（保留 Session）
    request.session.flush()  # 删除 sesssion 数据并创建新 key
    
    return HttpResponse('Session 已清空')

def session_test(request):
    """Session 测试与调试"""
    # 测试 session 是否可用
    request.session['test_key'] = 'test_value'
    assert request.session['test_key'] == 'test_value'
    
    # 获取 session 中的所有键
    keys = list(request.session.keys())
    
    # 获取 session key
    session_key = request.session.session_key
    
    # 循环遍历所有会话数据
    for key, value in request.session.items():
        print(f"{key}: {value}")
    
    return HttpResponse(f'Session Key: {session_key}, Keys: {keys}')

▶ 基于 Session 的用户认证

python
from django.contrib.auth import authenticate, login, logout
from django.shortcuts import render, redirect
from django.contrib import messages

def user_login(request):
    """用户登录"""
    if request.method == 'POST':
        username = request.POST['username']
        password = request.POST['password']
        
        # 验证用户
        user = authenticate(request, username=username, password=password)
        
        if user is not None:
            # 登录（创建 Session）
            login(request, user)
            
            # 可选：在 Session 中存储额外信息
            request.session['login_ip'] = request.META.get('REMOTE_ADDR')
            request.session['login_time'] = str(timezone.now())
            
            messages.success(request, '登录成功！')
            return redirect('blog:article_list')
        else:
            messages.error(request, '用户名或密码错误')
    
    return render(request, 'blog/login.html')

def user_logout(request):
    """用户登出"""
    logout(request)  # 清除 session
    messages.info(request, '已退出登录')
    return redirect('blog:index')

3.3 中间件原理与自定义中间件 ​

定义 ​

中间件（Middleware） 是 Django 请求/响应处理过程中的钩子框架，是一个用于全局处理请求和响应的轻量级插件系统。每个中间件负责特定的功能（如认证、CSRF 保护、Session 管理等），请求在经过 URL 路由之前和视图处理之后都会经过中间件。

语法 ​

▶ Django 内置中间件

python
# settings.py —— 默认中间件（按顺序执行）
MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',              # 安全防护
    'django.contrib.sessions.middleware.SessionMiddleware',       # Session 支持
    'django.middleware.common.CommonMiddleware',                  # 通用功能（URL 规范化等）
    'django.middleware.csrf.CsrfViewMiddleware',                  # CSRF 保护
    'django.contrib.auth.middleware.AuthenticationMiddleware',    # 用户认证
    'django.contrib.messages.middleware.MessageMiddleware',      # 消息提示
    'django.middleware.clickjacking.XFrameOptionsMiddleware',    # 点击劫持防护
]

▶ 中间件执行顺序

请求进入
    │
    ▼
SecurityMiddleware.process_request(request)
SessionMiddleware.process_request(request)
CommonMiddleware.process_request(request)
CsrfViewMiddleware.process_request(request)
AuthenticationMiddleware.process_request(request)
MessageMiddleware.process_request(request)
XFrameOptionsMiddleware.process_request(request)
    │
    ▼
URL 路由 → 视图函数
    │
    ▼
XFrameOptionsMiddleware.process_response(request, response)
MessageMiddleware.process_response(request, response)
AuthenticationMiddleware.process_response(request, response)
CsrfViewMiddleware.process_response(request, response)
CommonMiddleware.process_response(request, response)
SessionMiddleware.process_response(request, response)
SecurityMiddleware.process_response(request, response)
    │
    ▼
响应返回

▶ 自定义中间件

python
# blog/middleware.py —— 自定义中间件

import time
import logging
from django.http import HttpResponse
from django.utils.deprecation import MiddlewareMixin

logger = logging.getLogger(__name__)


# 方式一：使用 MiddlewareMixin（兼容新旧两种写法）
class RequestTimeMiddleware(MiddlewareMixin):
    """记录每个请求的处理时间"""
    
    def process_request(self, request):
        """请求处理前执行"""
        request.start_time = time.time()
    
    def process_view(self, request, view_func, view_args, view_kwargs):
        """视图函数调用前执行"""
        # 可以在此处检查权限、添加额外数据等
        pass
    
    def process_response(self, request, response):
        """响应返回前执行"""
        if hasattr(request, 'start_time'):
            duration = time.time() - request.start_time
            logger.info(f'请求 {request.path} 耗时: {duration:.3f}秒')
            response['X-Request-Time'] = f'{duration:.3f}s'
        return response
    
    def process_exception(self, request, exception):
        """视图函数抛出异常时执行"""
        logger.error(f'请求 {request.path} 发生异常: {exception}')
        return None  # 返回 None 则继续向上抛异常


# 方式二：函数式中间件（Django 1.10+ 推荐）
class BlockSuspiciousIPMiddleware:
    """拦截可疑 IP 的中间件"""
    
    def __init__(self, get_response):
        self.get_response = get_response
        # 初始化时执行一次（服务器启动时）
        self.blocked_ips = ['192.168.1.100', '10.0.0.50']
    
    def __call__(self, request):
        # 请求处理前的代码
        ip = request.META.get('REMOTE_ADDR')
        if ip in self.blocked_ips:
            return HttpResponse('您的 IP 已被禁止访问', status=403)
        
        if request.path.startswith('/admin/') and not request.path.endswith('/login/'):
            if not request.user.is_authenticated:
                logger.warning(f'未认证用户尝试访问管理后台: {ip}')
        
        # 调用下一个中间件或视图
        response = self.get_response(request)
        
        # 响应返回后的代码
        response['X-Request-IP'] = ip
        
        return response


class MaintenanceModeMiddleware:
    """维护模式中间件"""
    
    def __init__(self, get_response):
        self.get_response = get_response
        self.maintenance_mode = False  # 可通过数据库或环境变量控制
    
    def __call__(self, request):
        if self.maintenance_mode and not request.path.startswith('/admin/'):
            return HttpResponse(
                '<h1>网站维护中</h1><p>请稍后再访问</p>',
                status=503
            )
        return self.get_response(request)


class CorsMiddleware:
    """跨域资源共享（CORS）中间件"""
    
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        response = self.get_response(request)
        
        # 添加 CORS 头
        response['Access-Control-Allow-Origin'] = '*'
        response['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS'
        response['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
        
        # 处理预检请求
        if request.method == 'OPTIONS':
            return response
        
        return response

▶ 注册自定义中间件

python
# 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',
    
    # 自定义中间件（按顺序添加）
    'blog.middleware.RequestTimeMiddleware',
    'blog.middleware.BlockSuspiciousIPMiddleware',
]

3.4 信号机制与缓存系统 ​

定义 ​

信号（Signals） 是 Django 内置的发布-订阅模式实现，允许在特定事件发生时通知其他部分的代码。常见内置信号包括数据模型保存前后、删除前后、用户登录/登出等。缓存系统 用于临时存储计算结果或数据库查询结果，减少重复计算和数据库查询，显著提升应用性能。

语法 ​

▶ 常用内置信号

信号触发时机参数
pre_save模型 save() 调用前sender, instance, raw, using, update_fields
post_save模型 save() 调用后sender, instance, created, raw, using, update_fields
pre_delete模型 delete() 调用前sender, instance, using
post_delete模型 delete() 调用后sender, instance, using
m2m_changedManyToMany 字段变化时sender, instance, action, reverse, model, pk_set
user_logged_in用户登录成功sender, request, user
user_logged_out用户登出sender, request, user
user_login_failed用户登录失败sender, credentials, request
request_startedHTTP 请求开始时sender, environ
request_finishedHTTP 请求结束时sender

▶ 信号接收器定义

python
# blog/signals.py —— 信号处理器

import logging
from django.db.models.signals import post_save, post_delete, pre_save
from django.contrib.auth.signals import user_logged_in, user_logged_out
from django.dispatch import receiver
from django.core.cache import cache
from .models import Article

logger = logging.getLogger(__name__)


# 方式一：使用 @receiver 装饰器（推荐）
@receiver(post_save, sender=Article)
def article_saved_handler(sender, instance, created, **kwargs):
    """文章保存后的处理"""
    if created:
        logger.info(f'新文章创建: {instance.title} (ID: {instance.pk})')
    else:
        logger.info(f'文章更新: {instance.title}')
    
    # 清除文章列表缓存
    cache.delete('article_list_page_1')
    cache.delete_pattern('article_list_*')  # 清除所有文章列表缓存


@receiver(post_delete, sender=Article)
def article_deleted_handler(sender, instance, **kwargs):
    """文章删除后的处理"""
    logger.info(f'文章删除: {instance.title} (ID: {instance.pk})')
    cache.delete('article_list_page_1')


# 方式二：定义接收器函数，然后在 apps.py 中连接
def clear_category_cache(sender, instance, **kwargs):
    """清除分类缓存"""
    cache.delete('categories_all')

def update_article_count(sender, instance, created, **kwargs):
    """更新文章计数"""
    if created:
        from django.db.models import F
        # 更新分类的文章总数
        Category.objects.filter(pk=instance.category_id).update(
            article_count=F('article_count') + 1
        )


# 处理用户认证信号
@receiver(user_logged_in)
def user_logged_in_handler(sender, request, user, **kwargs):
    """用户登录成功"""
    logger.info(f'用户登录: {user.username}')
    request.session['last_login'] = str(timezone.now())


@receiver(user_logged_out)
def user_logged_out_handler(sender, request, user, **kwargs):
    """用户登出"""
    logger.info(f'用户登出: {user.username}')

▶ 信号注册（apps.py）

python
# blog/apps.py
from django.apps import AppConfig

class BlogConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'blog'
    verbose_name = '博客管理'
    
    def ready(self):
        """应用就绪时注册信号"""
        import blog.signals  # 导入信号处理器

▶ 自定义信号

python
# blog/signals.py —— 自定义信号
import django.dispatch

# 定义自定义信号
article_viewed = django.dispatch.Signal()
comment_created = django.dispatch.Signal()

# 发送信号
def article_detail(request, pk):
    article = get_object_or_404(Article, pk=pk)
    
    # 发送自定义信号
    article_viewed.send(
        sender=Article,
        instance=article,
        request=request,
        ip=request.META.get('REMOTE_ADDR'),
    )
    return render(request, 'blog/detail.html', {'article': article})

# 接收自定义信号
@receiver(article_viewed, sender=Article)
def article_viewed_handler(sender, instance, request, ip, **kwargs):
    """文章被浏览时增加阅读计数"""
    logger.info(f'文章被浏览: {instance.title}, IP: {ip}')
    Article.objects.filter(pk=instance.pk).update(views=F('views') + 1)

▶ 缓存系统配置

python
# settings.py —— 缓存配置

# 方式一：本地内存缓存（开发/测试用）
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
        'LOCATION': 'unique-snowflake',
    }
}

# 方式二：文件系统缓存
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.filebased.FileBasedCache',
        'LOCATION': BASE_DIR / 'cache',
    }
}

# 方式三：数据库缓存
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.db.DatabaseCache',
        'LOCATION': 'django_cache_table',
    }
}

# 方式四：Redis 缓存（生产环境推荐）
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',  # Django 4.0+
        'LOCATION': 'redis://127.0.0.1:6379/1',  # 使用 DB 1
        'OPTIONS': {
            'password': 'your_password',  # 如果有密码
            'timeout': 5,  # 连接超时
        },
        'KEY_PREFIX': 'blog',  # 键前缀
        'TIMEOUT': 300,  # 默认过期时间（秒）
    }
}

# 方式五：使用 django-redis（功能更丰富）
# pip install django-redis
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'OPTIONS': {
            'CLIENT_CLASS': 'django_redis.client.DefaultClient',
            'PASSWORD': 'your_password',
            'CONNECTION_POOL_CLASS': 'redis.BlockingConnectionPool',
            'CONNECTION_POOL_CLASS_KWARGS': {
                'max_connections': 50,
                'timeout': 20,
            },
            'MAX_ENTRIES': 1000,  # 最大缓存条目数
        },
        'KEY_PREFIX': 'blog',
    }
}

▶ 缓存操作

python
from django.core.cache import cache
from django.views.decorators.cache import cache_page
from django.views.decorators.vary import vary_on_cookie

# ---- 基本缓存操作 ----

# 设置缓存
cache.set('article_count', 100, timeout=300)  # 5 分钟过期
cache.set_many({
    'categories': ['Python', 'Django', 'JavaScript'],
    'tags': ['tutorial', 'guide'],
}, timeout=600)

# 获取缓存
count = cache.get('article_count', 0)  # 不存在返回默认值 0
data = cache.get_many(['categories', 'tags'])

# 检查键是否存在
if cache.has_key('article_count'):
    print('缓存存在')

# 递增/递减
cache.incr('article_count')      # +1
cache.incr('article_count', 10)  # +10
cache.decr('article_count')      # -1

# 删除缓存
cache.delete('article_count')
cache.delete_many(['categories', 'tags'])
cache.clear()  # 清空所有缓存（慎用）

# 模式删除（需 django-redis）
# cache.delete_pattern('article_list_*')

# ---- 缓存装饰器 ----

@cache_page(60 * 15)  # 缓存整个视图 15 分钟
def article_list(request):
    """文章列表（整个视图响应被缓存）"""
    articles = Article.objects.filter(is_published=True)
    return render(request, 'blog/article_list.html', {'articles': articles})


@vary_on_cookie  # 按 Cookie 区分缓存
@cache_page(60 * 15)
def article_detail(request, pk):
    """文章详情（不同用户的缓存不同）"""
    article = get_object_or_404(Article, pk=pk)
    return render(request, 'blog/article_detail.html', {'article': article})


# ---- 模板片段缓存 ----

"""
{% load cache %}

{% cache 300 sidebar %}
    {# 缓存侧栏 5 分钟 #}
    {% for category in categories %}
        <li>{{ category.name }}</li>
    {% endfor %}
{% endcache %}

{% cache 600 "article_detail" article.pk %}
    {# 使用动态键缓存 #}
    <h1>{{ article.title }}</h1>
    <p>{{ article.content }}</p>
{% endcache %}
"""


# ---- 低级别缓存使用模式 ----

def get_article_with_cache(pk):
    """带缓存的文章查询"""
    cache_key = f'article_detail_{pk}'
    
    article = cache.get(cache_key)
    if article is None:
        # 缓存未命中，从数据库查询
        article = get_object_or_404(Article, pk=pk)
        cache.set(cache_key, article, timeout=300)
    
    return article


def get_popular_articles(limit=10):
    """获取热门文章（缓存 1 小时）"""
    cache_key = f'popular_articles_{limit}'
    
    articles = cache.get(cache_key)
    if not articles:
        articles = list(
            Article.objects.filter(is_published=True)
            .order_by('-views')[:limit]
        )
        cache.set(cache_key, articles, timeout=3600)
    
    return articles

3.5 Celery 异步任务队列 ​

定义 ​

Celery 是一个分布式任务队列，用于在 Django 应用外异步执行耗时操作（如发送邮件、图片处理、生成报表、调用第三方 API 等）。它使用消息代理（Broker，如 Redis/RabbitMQ）来协调任务分发，Worker 进程负责执行任务。

架构 ​

用户请求
    │
    ▼
Django 视图 ──► 调用 task.delay() ──► 任务加入 Broker (Redis/RabbitMQ)
                                            │
                                            ▼
                                        Celery Worker
                                            │
                                    ┌───────┼───────┐
                                    │       │       │
                                    ▼       ▼       ▼
                                 发送邮件  处理图片  生成报表

安装与配置 ​

bash
# 安装 Celery 和 Redis 作为 Broker
pip install celery[redis]
python
# myproject/celery.py —— Celery 实例配置

import os
from celery import Celery

# 设置 Django 默认配置模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

app = Celery('myproject')

# 从 Django settings 读取配置（所有 CELERY_ 开头的配置项）
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自动发现所有 app 中的 tasks.py
app.autodiscover_tasks()


@app.task(bind=True, ignore_result=True)
def debug_task(self):
    """调试任务"""
    print(f'Request: {self.request!r}')
python
# myproject/__init__.py —— 确保 Celery 随 Django 启动

from .celery import app as celery_app

__all__ = ('celery_app',)
python
# settings.py —— Celery 配置

# Redis 作为消息代理
CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'

# 结果后端（存储任务执行结果）
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'

# 时区设置
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_ENABLE_UTC = False

# 任务序列化
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']

# 任务结果过期时间（秒）
CELERY_RESULT_EXPIRES = 3600 * 24  # 1 天

# Worker 并发数
CELERY_WORKER_CONCURRENCY = 4

# 任务重试
CELERY_TASK_ACKS_LATE = True  # 任务完成后再确认
CELERY_TASK_REJECT_ON_WORKER_LOST = True
CELERY_TASK_RETRY_LIMIT = 3  # 最大重试次数

任务定义 ​

python
# blog/tasks.py —— 异步任务定义

from celery import shared_task
from django.core.mail import send_mail
from django.conf import settings
from django.utils import timezone
import logging

logger = logging.getLogger(__name__)


@shared_task
def send_async_email(subject, message, recipient_list):
    """异步发送邮件"""
    try:
        send_mail(
            subject=subject,
            message=message,
            from_email=settings.DEFAULT_FROM_EMAIL,
            recipient_list=recipient_list,
            fail_silently=False,
        )
        logger.info(f'邮件发送成功: {recipient_list}')
        return {'status': 'success', 'recipients': recipient_list}
    except Exception as e:
        logger.error(f'邮件发送失败: {e}')
        raise  # 触发 Celery 重试


@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def process_image(self, image_path, operations):
    """处理图片（带重试机制）"""
    try:
        from PIL import Image
        img = Image.open(image_path)
        
        for op in operations:
            if op['type'] == 'resize':
                img = img.resize((op['width'], op['height']))
            elif op['type'] == 'thumbnail':
                img.thumbnail((op['width'], op['height']))
        
        img.save(image_path)
        logger.info(f'图片处理完成: {image_path}')
        return {'status': 'success', 'path': image_path}
    
    except Exception as exc:
        logger.error(f'图片处理失败: {exc}')
        # 重试任务
        raise self.retry(exc=exc)


@shared_task
def generate_weekly_report():
    """生成周报（定时任务）"""
    from datetime import timedelta
    from .models import Article
    
    week_ago = timezone.now() - timedelta(days=7)
    
    stats = {
        'new_articles': Article.objects.filter(created_at__gte=week_ago).count(),
        'total_views': Article.objects.filter(created_at__gte=week_ago).aggregate(
            total=Sum('views')
        )['total'] or 0,
        'top_articles': list(
            Article.objects.filter(is_published=True)
            .order_by('-views')[:10]
            .values('title', 'views')
        ),
    }
    
    # 将报表保存到文件或发送邮件
    logger.info(f'周报生成完成: {stats}')
    return stats


@shared_task
def cleanup_expired_sessions():
    """清理过期 Session"""
    from django.contrib.sessions.models import Session
    deleted_count = Session.objects.filter(expire_date__lt=timezone.now()).count()
    Session.objects.filter(expire_date__lt=timezone.now()).delete()
    logger.info(f'清理过期 Session: {deleted_count} 条')
    return {'deleted': deleted_count}

任务调用 ​

python
# blog/views.py —— 在视图中调用异步任务

from .tasks import send_async_email, process_image
from django.http import JsonResponse

def register_user(request):
    """用户注册"""
    if request.method == 'POST':
        # ... 处理用户注册逻辑
        
        # 异步发送欢迎邮件（不阻塞响应）
        send_async_email.delay(
            subject='欢迎注册！',
            message=f'亲爱的 {username}，欢迎加入！',
            recipient_list=[email],
        )
        
        return JsonResponse({'status': 'success', 'message': '注册成功！'})

def upload_avatar(request):
    """上传头像"""
    if request.method == 'POST' and request.FILES.get('avatar'):
        avatar = request.FILES['avatar']
        # 保存文件
        with open(f'media/avatars/{avatar.name}', 'wb+') as f:
            for chunk in avatar.chunks():
                f.write(chunk)
        
        # 异步处理：生成缩略图
        process_image.delay(
            f'media/avatars/{avatar.name}',
            [{'type': 'thumbnail', 'width': 200, 'height': 200}],
        )
        
        return JsonResponse({'status': 'success'})

定时任务（Celery Beat） ​

python
# settings.py —— 定时任务配置

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    # 每周一生成周报
    'generate-weekly-report': {
        'task': 'blog.tasks.generate_weekly_report',
        'schedule': crontab(hour=9, minute=0, day_of_week=1),  # 周一早上 9:00
        'args': (),
    },
    # 每天凌晨清理过期 Session
    'cleanup-sessions-daily': {
        'task': 'blog.tasks.cleanup_expired_sessions',
        'schedule': crontab(hour=3, minute=0),  # 每天凌晨 3:00
    },
    # 每隔 30 分钟执行一次
    'check-health': {
        'task': 'blog.tasks.check_system_health',
        'schedule': 1800.0,  # 秒为单位
    },
}

启动 Celery ​

bash
# 启动 Worker（处理异步任务）
celery -A myproject worker -l info

# 启动 Worker（生产模式，守护进程）
celery -A myproject worker -l info --concurrency=4 --max-tasks-per-child=1000

# 启动 Beat（定时任务调度器）
celery -A myproject beat -l info

# 同时启动 Worker 和 Beat
celery -A myproject worker -B -l info

# 查看任务状态
celery -A myproject status

# 查看正在运行的 Worker
celery -A myproject inspect active

# 清除所有任务
celery -A myproject purge

任务状态与结果 ​

python
# 在视图中查询任务状态
from celery.result import AsyncResult
from blog.tasks import send_async_email

def check_task_status(request, task_id):
    """查询异步任务状态"""
    result = AsyncResult(task_id)
    
    status_info = {
        'PENDING': '等待执行',
        'STARTED': '正在执行',
        'SUCCESS': '执行成功',
        'FAILURE': '执行失败',
        'RETRY': '重试中',
        'REVOKED': '已撤销',
    }
    
    response_data = {
        'task_id': task_id,
        'status': result.status,
        'status_display': status_info.get(result.status, '未知'),
    }
    
    if result.successful():
        response_data['result'] = result.result  # 任务返回值
    elif result.failed():
        response_data['error'] = str(result.result)  # 异常信息
    
    return JsonResponse(response_data)

四、小结 ​

知识点核心要点
FBV vs CBVFBV 函数式直观；CBV 通过继承复用代码；ListView/DetailView/CreateView 等通用视图大幅减少样板代码
Cookie 与 SessionCookie 存客户端（≤4KB），Session 存服务器；Cookie 通过 set_cookie() 操作，Session 通过 request.session 字典操作
中间件每个请求/响应都会经过中间件链；可自定义 process_request/process_view/process_response/process_exception
信号机制发布-订阅模式；post_save/post_delete 等内置信号在模型事件触发；在 apps.py 中注册信号
缓存系统Redis 是生产环境首选；cache.get/set 低级别缓存；@cache_page 视图级别缓存；模板片段缓存
Celery 异步任务处理耗时操作（邮件、图片处理）；delay() 调用任务；Celery Beat 实现定时任务

五、练习 ​

基础练习 ​

  1. CBV 实践：使用 ListView 和 DetailView 重构章节1中的博客文章列表和详情视图
  2. Session 计数器：创建一个视图，使用 Session 记录用户的访问次数，每次刷新页面次数 +1

进阶练习 ​

  1. 自定义中间件：编写一个 RateLimitMiddleware，限制单个 IP 每分钟最多请求 60 次，超过返回 429 状态码
  2. 信号应用：使用 post_save 信号，当文章被创建时自动发送新文章通知邮件（使用 Celery 异步发送）

挑战练习 ​

  1. 完整的异步任务系统：
    • 配置 Celery + Redis
    • 定义一个任务：批量生成文章摘要（调用 AI API 或本地处理）
    • 在 Admin 后台添加"批量生成摘要"操作按钮
    • 显示任务执行进度和结果

参考代码（练习 2 Session 计数器） ​

python
def visit_counter(request):
    count = request.session.get('visit_count', 0)
    count += 1
    request.session['visit_count'] = count
    return HttpResponse(f'您已访问本页面 {count} 次')

Python 学习资料

仅供学习使用