第18章 Django Web开发

学习目标

完成本章学习后，读者应能够：

  1. 理解Django架构：掌握MTV（Model-Template-View）设计模式的原理与实现机制
  2. 精通ORM系统：运用Django ORM实现复杂查询、关系映射、迁移管理与性能优化
  3. 掌握视图系统：熟练使用函数视图与类视图，理解请求-响应处理流程
  4. 设计URL体系：实现RESTful URL设计、命名路由与反向解析
  5. 构建模板系统：掌握模板继承、自定义标签与过滤器、上下文处理器
  6. 实现表单处理：运用ModelForm、Form验证与文件上传处理
  7. 定制Admin后台：深度定制Admin界面，实现数据管理与业务逻辑集成
  8. 构建认证体系：实现用户认证、权限控制、分组管理与自定义认证后端
  9. 掌握中间件机制：理解中间件执行链，实现自定义中间件
  10. 完成生产部署：掌握Django的安全配置、性能优化与部署策略

18.1 Django架构与设计哲学

18.1.1 MTV架构模式

Django采用MTV（Model-Template-View）架构，是MVC模式的变体：

MVC组件Django对应职责
ModelModel数据访问层，定义数据结构与数据库交互
ViewTemplate表现层，负责数据展示与渲染
ControllerView业务逻辑层，处理请求与返回响应
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
Django请求处理流程：

HTTP请求
    │
    ▼
┌──────────┐
│  URLconf  │  URL路由匹配
└─────┬────┘
      │
      ▼
┌──────────┐
│   View    │  业务逻辑处理
└─────┬────┘
      │
      ├──────────────┐
      ▼              ▼
┌──────────┐  ┌──────────┐
│  Model    │  │ Template │
│ 数据访问  │  │ 渲染展示  │
└──────────┘  └──────────┘
      │              │
      └──────┬───────┘
             ▼
        HTTP响应

18.1.2 Django核心设计原则

  1. DRY（Don’t Repeat Yourself）：通过ORM、中间件、混入类等机制消除重复
  2. 松耦合：各层之间通过明确定义的接口交互
  3. 快速开发：内置Admin、Auth、ORM等开箱即用的组件
  4. 显式优于隐式：配置明确，行为可预测
  5. 安全优先：默认防御XSS、CSRF、SQL注入、点击劫持等攻击

18.1.3 项目创建与结构

1
2
3
4
5
6
pip install django
django-admin startproject mysite
cd mysite
python manage.py startapp blog
python manage.py startapp users
python manage.py startapp api

大型项目推荐结构：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
mysite/
├── manage.py
├── requirements/
│   ├── base.txt
│   ├── development.txt
│   └── production.txt
├── config/
│   ├── __init__.py
│   ├── settings/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── development.py
│   │   ├── production.py
│   │   └── testing.py
│   ├── urls.py
│   ├── wsgi.py
│   └── asgi.py
├── apps/
│   ├── __init__.py
│   ├── blog/
│   │   ├── __init__.py
│   │   ├── models.py
│   │   ├── views.py
│   │   ├── urls.py
│   │   ├── forms.py
│   │   ├── admin.py
│   │   ├── apps.py
│   │   ├── services.py
│   │   ├── migrations/
│   │   └── templates/blog/
│   └── users/
│       ├── __init__.py
│       ├── models.py
│       ├── views.py
│       └── ...
├── templates/
│   ├── base.html
│   └── components/
├── static/
│   ├── css/
│   ├── js/
│   └── images/
├── media/
└── tests/
    ├── __init__.py
    ├── conftest.py
    ├── test_blog.py
    └── test_users.py

18.1.4 多环境配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import os
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent.parent

SECRET_KEY = os.environ.get("SECRET_KEY", "django-insecure-dev-key")

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "django.contrib.sites",
    "apps.blog",
    "apps.users",
]

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",
]

ROOT_URLCONF = "config.urls"
WSGI_APPLICATION = "config.wsgi.application"

DATABASES = {
    "default": {
        "ENGINE": os.environ.get("DB_ENGINE", "django.db.backends.sqlite3"),
        "NAME": os.environ.get("DB_NAME", BASE_DIR / "db.sqlite3"),
        "USER": os.environ.get("DB_USER", ""),
        "PASSWORD": os.environ.get("DB_PASSWORD", ""),
        "HOST": os.environ.get("DB_HOST", ""),
        "PORT": os.environ.get("DB_PORT", ""),
        "OPTIONS": {
            "connect_timeout": 10,
        },
    }
}

LANGUAGE_CODE = "zh-hans"
TIME_ZONE = "Asia/Shanghai"
USE_I18N = True
USE_TZ = True

STATIC_URL = "/static/"
STATIC_ROOT = BASE_DIR / "staticfiles"
STATICFILES_DIRS = [BASE_DIR / "static"]

MEDIA_URL = "/media/"
MEDIA_ROOT = BASE_DIR / "media"

DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
SITE_ID = 1

开发环境配置 config/settings/development.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from .base import *

DEBUG = True
ALLOWED_HOSTS = ["*"]

INSTALLED_APPS += [
    "django_extensions",
    "debug_toolbar",
]

MIDDLEWARE = [
    "debug_toolbar.middleware.DebugToolbarMiddleware",
] + MIDDLEWARE

INTERNAL_IPS = ["127.0.0.1"]

DATABASES["default"]["NAME"] = BASE_DIR / "dev.db"

生产环境配置 config/settings/production.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from .base import *

DEBUG = False
ALLOWED_HOSTS = os.environ.get("ALLOWED_HOSTS", "").split(",")

SECURE_BROWSER_XSS_FILTER = True
SECURE_CONTENT_TYPE_NOSNIFF = True
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True
SECURE_HSTS_SECONDS = 31536000
SECURE_SSL_REDIRECT = True
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True
X_FRAME_OPTIONS = "DENY"

DATABASES["default"]["CONN_MAX_AGE"] = 60
CACHES = {
    "default": {
        "BACKEND": "django.core.cache.backends.redis.RedisCache",
        "LOCATION": os.environ.get("REDIS_URL", "redis://localhost:6379/0"),
    }
}

18.2 模型与ORM

18.2.1 模型定义与字段类型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
from django.db import models
from django.contrib.auth.models import AbstractUser
from django.utils.text import slugify
from django.urls import reverse


class User(AbstractUser):
    bio = models.TextField(blank=True)
    avatar = models.ImageField(upload_to="avatars/", blank=True)
    website = models.URLField(blank=True)
    following = models.ManyToManyField(
        "self", symmetrical=False, related_name="followers", blank=True
    )

    def __str__(self):
        return self.username


class Category(models.Model):
    name = models.CharField(max_length=100, unique=True)
    slug = models.SlugField(max_length=100, unique=True)
    description = models.TextField(blank=True)
    parent = models.ForeignKey(
        "self", on_delete=models.CASCADE, null=True, blank=True, related_name="children"
    )
    order = models.PositiveIntegerField(default=0)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        verbose_name = "分类"
        verbose_name_plural = "分类"
        ordering = ["order", "name"]

    def __str__(self):
        return self.name

    def save(self, *args, **kwargs):
        if not self.slug:
            self.slug = slugify(self.name)
        super().save(*args, **kwargs)


class Tag(models.Model):
    name = models.CharField(max_length=50, unique=True)
    slug = models.SlugField(max_length=50, unique=True)

    class Meta:
        verbose_name = "标签"
        verbose_name_plural = "标签"
        ordering = ["name"]

    def __str__(self):
        return self.name


class PublishedManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().filter(status=Post.Status.PUBLISHED)


class Post(models.Model):
    class Status(models.TextChoices):
        DRAFT = "DF", "草稿"
        PUBLISHED = "PB", "已发布"
        ARCHIVED = "AR", "已归档"

    title = models.CharField(max_length=200)
    slug = models.SlugField(max_length=200, unique=True)
    content = models.TextField()
    summary = models.CharField(max_length=500, blank=True)
    status = models.CharField(
        max_length=2, choices=Status.choices, default=Status.DRAFT
    )
    author = models.ForeignKey(
        User, on_delete=models.CASCADE, related_name="blog_posts"
    )
    category = models.ForeignKey(
        Category, on_delete=models.SET_NULL, null=True, blank=True, related_name="posts"
    )
    tags = models.ManyToManyField(Tag, blank=True, related_name="posts")
    featured_image = models.ImageField(upload_to="posts/%Y/%m/", blank=True)
    view_count = models.PositiveIntegerField(default=0)
    is_featured = models.BooleanField(default=False)
    published_at = models.DateTimeField(null=True, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    objects = models.Manager()
    published = PublishedManager()

    class Meta:
        verbose_name = "文章"
        verbose_name_plural = "文章"
        ordering = ["-published_at"]
        indexes = [
            models.Index(fields=["-published_at"]),
            models.Index(fields=["status"]),
            models.Index(fields=["slug"]),
        ]

    def __str__(self):
        return self.title

    def get_absolute_url(self):
        return reverse("blog:post_detail", args=[self.slug])

    def save(self, *args, **kwargs):
        if not self.slug:
            self.slug = slugify(self.title)
        if self.status == self.Status.PUBLISHED and not self.published_at:
            from django.utils import timezone
            self.published_at = timezone.now()
        super().save(*args, **kwargs)


class Comment(models.Model):
    post = models.ForeignKey(Post, on_delete=models.CASCADE, related_name="comments")
    author = models.ForeignKey(
        User, on_delete=models.CASCADE, related_name="comments"
    )
    parent = models.ForeignKey(
        "self", on_delete=models.CASCADE, null=True, blank=True, related_name="replies"
    )
    content = models.TextField()
    is_approved = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        verbose_name = "评论"
        verbose_name_plural = "评论"
        ordering = ["created_at"]

    def __str__(self):
        return f"Comment by {self.author} on {self.post}"

18.2.2 高级ORM查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
from django.db.models import Q, F, Count, Avg, Sum, Max, Min, Case, When, Value
from django.db.models.functions import TruncMonth, Coalesce


posts = Post.published.all()


posts = Post.published.filter(
    Q(title__icontains="django") | Q(content__icontains="django")
)


posts = Post.published.filter(
    title__icontains="django",
    category__name="Web开发",
)


posts = Post.published.filter(
    Q(title__icontains="django") & ~Q(status=Post.Status.ARCHIVED)
)


Post.published.filter(view_count__gt=F("author__post_count"))


posts = Post.published.annotate(
    comment_count=Count("comments", filter=Q(comments__is_approved=True)),
    avg_rating=Avg("ratings__score"),
).filter(comment_count__gt=5)


Post.published.annotate(
    month=TruncMonth("published_at")
).values("month").annotate(
    count=Count("id")
).order_by("-month")


Post.published.aggregate(
    total_posts=Count("id"),
    avg_views=Avg("view_count"),
    max_views=Max("view_count"),
)


Post.published.annotate(
    status_label=Case(
        When(view_count__gt=1000, then=Value("热门")),
        When(view_count__gt=100, then=Value("普通")),
        default=Value("冷门"),
    )
)


Post.published.select_related("author", "category").all()


Post.published.prefetch_related("tags", "comments").all()


from django.db.models import Prefetch

Post.published.select_related("author").prefetch_related(
    Prefetch("comments", queryset=Comment.objects.filter(is_approved=True))
).all()


Post.published.filter(author__in=authors).only("title", "slug", "published_at")


Post.published.defer("content")


Post.published.bulk_create([
    Post(title=f"Post {i}", content=f"Content {i}", author=user, status="PB")
    for i in range(100)
])


Post.published.filter(status="DF").update(status="PB")


Post.published.filter(view_count=0).delete()

18.2.3 数据库迁移

1
2
3
4
5
6
7
python manage.py makemigrations
python manage.py makemigrations blog --name add_post_summary
python manage.py migrate
python manage.py migrate blog 0003
python manage.py showmigrations
python manage.py sqlmigrate blog 0001
python manage.py makemigrations --empty blog --name populate_slug

数据迁移示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from django.db import migrations


def populate_slugs(apps, schema_editor):
    Post = apps.get_model("blog", "Post")
    for post in Post.objects.filter(slug=""):
        post.slug = slugify(post.title)
        post.save(update_fields=["slug"])


def reverse_populate(apps, schema_editor):
    pass


class Migration(migrations.Migration):
    dependencies = [
        ("blog", "0003_add_post_slug"),
    ]

    operations = [
        migrations.RunPython(populate_slugs, reverse_populate),
    ]

18.3 视图系统

18.3.1 函数视图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
from django.shortcuts import render, get_object_or_404, redirect
from django.http import JsonResponse
from django.contrib.auth.decorators import login_required
from django.views.decorators.http import require_http_methods, require_GET
from django.core.paginator import Paginator
from django.db.models import F
from .models import Post, Comment
from .forms import PostForm, CommentForm


@require_GET
def post_list(request):
    page_num = request.GET.get("page", 1)
    tag_slug = request.GET.get("tag")
    category_slug = request.GET.get("category")

    posts = Post.published.select_related("author", "category").prefetch_related("tags")

    if tag_slug:
        posts = posts.filter(tags__slug=tag_slug)
    if category_slug:
        posts = posts.filter(category__slug=category_slug)

    paginator = Paginator(posts, 10)
    page_obj = paginator.get_page(page_num)

    return render(request, "blog/post_list.html", {
        "page_obj": page_obj,
        "tag_slug": tag_slug,
        "category_slug": category_slug,
    })


@require_GET
def post_detail(request, slug):
    post = get_object_or_404(
        Post.published.select_related("author", "category").prefetch_related("tags"),
        slug=slug,
    )
    Post.published.filter(pk=post.pk).update(view_count=F("view_count") + 1)

    comments = post.comments.filter(is_approved=True, parent__isnull=True)
    comment_form = CommentForm()

    return render(request, "blog/post_detail.html", {
        "post": post,
        "comments": comments,
        "comment_form": comment_form,
    })


@login_required
@require_http_methods(["GET", "POST"])
def post_create(request):
    if request.method == "POST":
        form = PostForm(request.POST, request.FILES)
        if form.is_valid():
            post = form.save(commit=False)
            post.author = request.user
            post.save()
            form.save_m2m()
            return redirect(post.get_absolute_url())
    else:
        form = PostForm()

    return render(request, "blog/post_form.html", {"form": form, "action": "创建"})


@login_required
@require_http_methods(["GET", "POST"])
def post_update(request, slug):
    post = get_object_or_404(Post, slug=slug, author=request.user)

    if request.method == "POST":
        form = PostForm(request.POST, request.FILES, instance=post)
        if form.is_valid():
            form.save()
            return redirect(post.get_absolute_url())
    else:
        form = PostForm(instance=post)

    return render(request, "blog/post_form.html", {"form": form, "action": "编辑"})

18.3.2 类视图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
from django.views.generic import (
    ListView,
    DetailView,
    CreateView,
    UpdateView,
    DeleteView,
)
from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin
from django.urls import reverse_lazy
from django.db.models import F
from .models import Post, Comment
from .forms import PostForm, CommentForm


class PostListView(ListView):
    model = Post
    template_name = "blog/post_list.html"
    context_object_name = "posts"
    paginate_by = 10

    def get_queryset(self):
        queryset = Post.published.select_related(
            "author", "category"
        ).prefetch_related("tags")

        tag_slug = self.kwargs.get("tag_slug")
        if tag_slug:
            queryset = queryset.filter(tags__slug=tag_slug)

        category_slug = self.kwargs.get("category_slug")
        if category_slug:
            queryset = queryset.filter(category__slug=category_slug)

        return queryset

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["tag_slug"] = self.kwargs.get("tag_slug")
        context["category_slug"] = self.kwargs.get("category_slug")
        return context


class PostDetailView(DetailView):
    model = Post
    template_name = "blog/post_detail.html"
    context_object_name = "post"

    def get_queryset(self):
        return Post.published.select_related(
            "author", "category"
        ).prefetch_related("tags", "comments")

    def get_object(self, queryset=None):
        obj = super().get_object(queryset)
        Post.objects.filter(pk=obj.pk).update(view_count=F("view_count") + 1)
        return obj

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["comments"] = self.object.comments.filter(
            is_approved=True, parent__isnull=True
        )
        context["comment_form"] = CommentForm()
        return context


class PostCreateView(LoginRequiredMixin, CreateView):
    model = Post
    form_class = PostForm
    template_name = "blog/post_form.html"

    def form_valid(self, form):
        form.instance.author = self.request.user
        return super().form_valid(form)

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["action"] = "创建"
        return context


class PostUpdateView(LoginRequiredMixin, UserPassesTestMixin, UpdateView):
    model = Post
    form_class = PostForm
    template_name = "blog/post_form.html"

    def test_func(self):
        post = self.get_object()
        return self.request.user == post.author or self.request.user.is_staff

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["action"] = "编辑"
        return context


class PostDeleteView(LoginRequiredMixin, UserPassesTestMixin, DeleteView):
    model = Post
    template_name = "blog/post_confirm_delete.html"
    success_url = reverse_lazy("blog:post_list")

    def test_func(self):
        post = self.get_object()
        return self.request.user == post.author or self.request.user.is_staff

18.3.3 通用混入类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from django.http import JsonResponse


class AjaxFormMixin:
    def form_invalid(self, form):
        response = super().form_invalid(form)
        if self.request.headers.get("x-requested-with") == "XMLHttpRequest":
            return JsonResponse(form.errors, status=400)
        return response

    def form_valid(self, form):
        response = super().form_valid(form)
        if self.request.headers.get("x-requested-with") == "XMLHttpRequest":
            return JsonResponse({"success": True, "redirect_url": self.get_success_url()})
        return response


class OwnerRequiredMixin:
    def test_func(self):
        obj = self.get_object()
        return self.request.user == getattr(obj, "author", None) or self.request.user.is_staff


class CacheControlMixin:
    cache_timeout = 300

    def dispatch(self, request, *args, **kwargs):
        response = super().dispatch(request, *args, **kwargs)
        response["Cache-Control"] = f"max-age={self.cache_timeout}"
        return response

18.4 URL配置

18.4.1 URL路由设计

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from django.urls import path
from . import views

app_name = "blog"

urlpatterns = [
    path("", views.PostListView.as_view(), name="post_list"),
    path("search/", views.PostSearchView.as_view(), name="post_search"),
    path("category/<slug:slug>/", views.CategoryPostListView.as_view(), name="category_posts"),
    path("tag/<slug:slug>/", views.TagPostListView.as_view(), name="tag_posts"),
    path("post/<slug:slug>/", views.PostDetailView.as_view(), name="post_detail"),
    path("post/new/", views.PostCreateView.as_view(), name="post_create"),
    path("post/<slug:slug>/edit/", views.PostUpdateView.as_view(), name="post_update"),
    path("post/<slug:slug>/delete/", views.PostDeleteView.as_view(), name="post_delete"),
    path("post/<slug:slug>/comment/", views.CommentCreateView.as_view(), name="comment_create"),
]

主URL配置 config/urls.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from django.contrib import admin
from django.urls import path, include
from django.conf import settings
from django.conf.urls.static import static

urlpatterns = [
    path("admin/", admin.site.urls),
    path("", include("apps.blog.urls", namespace="blog")),
    path("users/", include("apps.users.urls", namespace="users")),
    path("api/v1/", include("apps.api.urls", namespace="api")),
]

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

handler404 = "apps.core.views.custom_404"
handler500 = "apps.core.views.custom_500"

18.5 模板系统

18.5.1 模板继承与组件化

基础模板 templates/base.html：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>{% block title %}MySite{% endblock %}</title>
    {% load static %}
    <link rel="stylesheet" href="{% static 'css/style.css' %}">
    {% block extra_css %}{% endblock %}
</head>
<body>
    {% include "components/navbar.html" %}

    <main class="container">
        {% if messages %}
        {% include "components/messages.html" %}
        {% endif %}

        {% block content %}{% endblock %}
    </main>

    {% include "components/footer.html" %}

    <script src="{% static 'js/main.js' %}"></script>
    {% block extra_js %}{% endblock %}
</body>
</html>

文章列表模板 blog/templates/blog/post_list.html：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
{% extends "base.html" %}
{% load static %}

{% block title %}文章列表 - {{ block.super }}{% endblock %}

{% block content %}
<div class="post-list">
    <h1>文章列表</h1>

    {% for post in page_obj %}
    <article class="post-card">
        <h2><a href="{{ post.get_absolute_url }}">{{ post.title }}</a></h2>
        <div class="post-meta">
            <span class="author">{{ post.author.username }}</span>
            <span class="date">{{ post.published_at|date:"Y-m-d" }}</span>
            <span class="views">{{ post.view_count }} 次阅读</span>
        </div>
        <p class="summary">{{ post.summary|default:post.content|truncatewords:50 }}</p>
        <div class="tags">
            {% for tag in post.tags.all %}
            <a href="{% url 'blog:tag_posts' slug=tag.slug %}" class="tag">{{ tag.name }}</a>
            {% endfor %}
        </div>
    </article>
    {% empty %}
    <p class="empty">暂无文章。</p>
    {% endfor %}

    {% if page_obj.has_other_pages %}
    {% include "components/pagination.html" with page_obj=page_obj %}
    {% endif %}
</div>
{% endblock %}

18.5.2 自定义模板标签与过滤器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
from django import template
from django.utils import timezone

register = template.Library()


@register.filter(name="time_ago")
def time_ago(value):
    if value is None:
        return ""
    now = timezone.now()
    diff = now - value
    seconds = int(diff.total_seconds())

    intervals = [
        (31536000, "年"),
        (2592000, "个月"),
        (604800, "周"),
        (86400, "天"),
        (3600, "小时"),
        (60, "分钟"),
        (1, "秒"),
    ]

    for interval, label in intervals:
        count = seconds // interval
        if count > 0:
            return f"{count}{label}前"
    return "刚刚"


@register.filter(name="truncate_html")
def truncate_html(value, length=200):
    import re
    text = re.sub(r"<[^>]+>", "", value)
    if len(text) <= length:
        return value
    return text[:length].rsplit(" ", 1)[0] + "..."


@register.inclusion_tag("components/post_card.html")
def render_post_card(post, show_author=True):
    return {"post": post, "show_author": show_author}


@register.simple_tag
def query_transform(request, **kwargs):
    updated = request.GET.copy()
    for key, value in kwargs.items():
        if value:
            updated[key] = value
        elif key in updated:
            del updated[key]
    return updated.urlencode()

18.6 表单系统

18.6.1 ModelForm与表单验证

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
from django import forms
from django.core.exceptions import ValidationError
from .models import Post, Comment


class PostForm(forms.ModelForm):
    class Meta:
        model = Post
        fields = ["title", "content", "summary", "category", "tags", "featured_image", "status"]
        widgets = {
            "title": forms.TextInput(attrs={"class": "form-control", "placeholder": "文章标题"}),
            "content": forms.Textarea(attrs={"class": "form-control", "rows": 15}),
            "summary": forms.TextInput(attrs={"class": "form-control", "placeholder": "文章摘要"}),
            "category": forms.Select(attrs={"class": "form-select"}),
            "tags": forms.CheckboxSelectMultiple(),
            "status": forms.Select(attrs={"class": "form-select"}),
        }

    def clean_title(self):
        title = self.cleaned_data.get("title", "")
        if len(title) < 5:
            raise ValidationError("标题长度不能少于5个字符")
        return title

    def clean(self):
        cleaned_data = super().clean()
        status = cleaned_data.get("status")
        category = cleaned_data.get("category")

        if status == Post.Status.PUBLISHED and not category:
            self.add_error("category", "发布文章必须选择分类")

        return cleaned_data


class CommentForm(forms.ModelForm):
    class Meta:
        model = Comment
        fields = ["content", "parent"]
        widgets = {
            "content": forms.Textarea(attrs={
                "class": "form-control",
                "rows": 4,
                "placeholder": "写下你的评论...",
            }),
            "parent": forms.HiddenInput(),
        }

    def clean_content(self):
        content = self.cleaned_data.get("content", "")
        if len(content) < 5:
            raise ValidationError("评论内容不能少于5个字符")
        return content


class ContactForm(forms.Form):
    name = forms.CharField(
        max_length=100,
        widget=forms.TextInput(attrs={"class": "form-control", "placeholder": "姓名"}),
    )
    email = forms.EmailField(
        widget=forms.EmailInput(attrs={"class": "form-control", "placeholder": "邮箱"}),
    )
    subject = forms.CharField(
        max_length=200,
        widget=forms.TextInput(attrs={"class": "form-control", "placeholder": "主题"}),
    )
    message = forms.CharField(
        widget=forms.Textarea(attrs={"class": "form-control", "rows": 6, "placeholder": "消息内容"}),
    )

18.6.2 文件上传处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import os
import uuid
from django import forms
from django.core.exceptions import ValidationError


def validate_file_size(value):
    max_size = 5 * 1024 * 1024
    if value.size > max_size:
        raise ValidationError(f"文件大小不能超过5MB")


def validate_file_extension(value):
    allowed_extensions = [".jpg", ".jpeg", ".png", ".gif", ".pdf", ".doc", ".docx"]
    ext = os.path.splitext(value.name)[1].lower()
    if ext not in allowed_extensions:
        raise ValidationError(f"不支持的文件类型: {ext}")


class UploadForm(forms.Form):
    file = forms.FileField(
        validators=[validate_file_size, validate_file_extension],
        widget=forms.FileInput(attrs={"class": "form-control", "accept": ".jpg,.jpeg,.png,.gif,.pdf"}),
    )


def upload_file(request):
    if request.method == "POST":
        form = UploadForm(request.POST, request.FILES)
        if form.is_valid():
            uploaded_file = request.FILES["file"]
            ext = os.path.splitext(uploaded_file.name)[1]
            filename = f"{uuid.uuid4().hex}{ext}"
            filepath = os.path.join("uploads", filename)

            from django.core.files.storage import default_storage
            saved_path = default_storage.save(filepath, uploaded_file)

            return JsonResponse({"url": default_storage.url(saved_path)})
    else:
        form = UploadForm()

    return render(request, "upload.html", {"form": form})

18.7 Admin后台定制

18.7.1 ModelAdmin配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
from django.contrib import admin
from django.utils.html import format_html
from django.urls import reverse
from .models import Post, Category, Tag, Comment


@admin.register(Post)
class PostAdmin(admin.ModelAdmin):
    list_display = [
        "title", "author", "category", "status",
        "view_count", "is_featured", "published_at",
    ]
    list_display_links = ["title"]
    list_editable = ["status", "is_featured"]
    list_filter = ["status", "is_featured", "category", "created_at"]
    search_fields = ["title", "content", "author__username"]
    prepopulated_fields = {"slug": ("title",)}
    raw_id_fields = ["author"]
    date_hierarchy = "published_at"
    ordering = ["-published_at"]
    readonly_fields = ["view_count", "created_at", "updated_at"]
    save_on_top = True
    list_per_page = 25
    actions = ["make_published", "make_draft", "mark_featured"]

    fieldsets = (
        ("基本信息", {
            "fields": ("title", "slug", "author", "status", "is_featured"),
        }),
        ("内容", {
            "fields": ("content", "summary", "featured_image"),
            "classes": ("wide",),
        }),
        ("分类与标签", {
            "fields": ("category", "tags"),
        }),
        ("统计信息", {
            "fields": ("view_count", "published_at", "created_at", "updated_at"),
            "classes": ("collapse",),
        }),
    )

    filter_horizontal = ["tags"]

    @admin.action(description="标记为已发布")
    def make_published(self, request, queryset):
        from django.utils import timezone
        updated = queryset.filter(status=Post.Status.DRAFT).update(
            status=Post.Status.PUBLISHED, published_at=timezone.now()
        )
        self.message_user(request, f"成功发布 {updated} 篇文章")

    @admin.action(description="标记为草稿")
    def make_draft(self, request, queryset):
        updated = queryset.update(status=Post.Status.DRAFT)
        self.message_user(request, f"成功将 {updated} 篇文章设为草稿")

    @admin.action(description="标记为精选")
    def mark_featured(self, request, queryset):
        updated = queryset.update(is_featured=True)
        self.message_user(request, f"成功将 {updated} 篇文章设为精选")


@admin.register(Category)
class CategoryAdmin(admin.ModelAdmin):
    list_display = ["name", "slug", "parent", "order", "post_count"]
    prepopulated_fields = {"slug": ("name",)}
    ordering = ["order", "name"]

    def post_count(self, obj):
        return obj.posts.count()
    post_count.short_description = "文章数"


@admin.register(Comment)
class CommentAdmin(admin.ModelAdmin):
    list_display = ["author", "post", "content_preview", "is_approved", "created_at"]
    list_editable = ["is_approved"]
    list_filter = ["is_approved", "created_at"]
    search_fields = ["content", "author__username"]
    actions = ["approve_comments"]

    def content_preview(self, obj):
        return obj.content[:50] + "..." if len(obj.content) > 50 else obj.content
    content_preview.short_description = "评论内容"

    @admin.action(description="审核通过")
    def approve_comments(self, request, queryset):
        updated = queryset.update(is_approved=True)
        self.message_user(request, f"成功审核 {updated} 条评论")

18.7.2 Admin站点定制

1
2
3
4
5
6
from django.contrib import admin

admin.site.site_header = "博客管理系统"
admin.site.site_title = "博客管理"
admin.site.index_title = "欢迎来到博客管理后台"
admin.site.enable_nav_sidebar = True

18.8 认证与授权

18.8.1 用户认证视图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
from django.shortcuts import render, redirect
from django.contrib.auth import login, logout, authenticate
from django.contrib.auth.decorators import login_required
from django.contrib.auth.forms import UserCreationForm, AuthenticationForm
from django.views.generic import CreateView
from django.urls import reverse_lazy
from .forms import CustomUserCreationForm, ProfileUpdateForm
from .models import User


class RegisterView(CreateView):
    form_class = CustomUserCreationForm
    template_name = "users/register.html"
    success_url = reverse_lazy("users:login")

    def form_valid(self, form):
        response = super().form_valid(form)
        user = authenticate(
            self.request,
            username=form.cleaned_data["username"],
            password=form.cleaned_data["password1"],
        )
        if user:
            login(self.request, user)
        return response


def login_view(request):
    if request.method == "POST":
        form = AuthenticationForm(request, data=request.POST)
        if form.is_valid():
            user = form.get_user()
            login(request, user)
            next_url = request.GET.get("next", "blog:post_list")
            return redirect(next_url)
    else:
        form = AuthenticationForm()
    return render(request, "users/login.html", {"form": form})


def logout_view(request):
    logout(request)
    return redirect("blog:post_list")


@login_required
def profile_view(request):
    if request.method == "POST":
        form = ProfileUpdateForm(request.POST, request.FILES, instance=request.user)
        if form.is_valid():
            form.save()
            return redirect("users:profile")
    else:
        form = ProfileUpdateForm(instance=request.user)
    return render(request, "users/profile.html", {"form": form})

18.8.2 自定义认证后端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from django.contrib.auth.backends import ModelBackend
from django.contrib.auth import get_user_model
from django.db.models import Q

UserModel = get_user_model()


class EmailOrUsernameModelBackend(ModelBackend):
    def authenticate(self, request, username=None, password=None, **kwargs):
        try:
            user = UserModel.objects.get(
                Q(username=username) | Q(email=username)
            )
        except UserModel.DoesNotExist:
            return None

        if user.check_password(password) and self.user_can_authenticate(user):
            return user
        return None

配置：

1
2
3
4
AUTHENTICATION_BACKENDS = [
    "apps.users.backends.EmailOrUsernameModelBackend",
    "django.contrib.auth.backends.ModelBackend",
]

18.8.3 权限与分组

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from django.contrib.auth.models import Permission, Group
from django.contrib.contenttypes.models import ContentType
from .models import Post


content_type = ContentType.objects.get_for_model(Post)

edit_any_post = Permission.objects.create(
    codename="edit_any_post",
    name="Can edit any post",
    content_type=content_type,
)

publish_post = Permission.objects.create(
    codename="publish_post",
    name="Can publish post",
    content_type=content_type,
)

editors, _ = Group.objects.get_or_create(name="Editors")
editors.permissions.add(edit_any_post, publish_post)

user.groups.add(editors)


if request.user.has_perm("blog.edit_any_post"):
    pass

if request.user.has_perm("blog.publish_post"):
    pass

18.9 中间件

18.9.1 中间件执行机制

Django中间件采用洋葱模型，请求从外到内经过每个中间件的process_request，响应从内到外经过process_response：

1
2
3
4
5
6
请求 →  Middleware1.process_request
      →  Middleware2.process_request
      →  View
      →  Middleware2.process_response
      →  Middleware1.process_response
      → 响应

18.9.2 自定义中间件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import time
import logging
from django.http import JsonResponse

logger = logging.getLogger(__name__)


class RequestTimingMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        start_time = time.time()

        response = self.get_response(request)

        duration = time.time() - start_time
        response["X-Request-Duration"] = f"{duration:.3f}s"

        if duration > 1.0:
            logger.warning(
                "Slow request: %s %s took %.3fs",
                request.method, request.path, duration,
            )

        return response


class APIRateLimitMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.request_counts = {}

    def __call__(self, request):
        if request.path.startswith("/api/"):
            ip = self.get_client_ip(request)
            key = f"{ip}:{int(time.time() / 60)}"

            self.request_counts[key] = self.request_counts.get(key, 0) + 1

            if self.request_counts[key] > 60:
                return JsonResponse(
                    {"error": "请求过于频繁，请稍后再试"}, status=429
                )

        return self.get_response(request)

    @staticmethod
    def get_client_ip(request):
        x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
        if x_forwarded_for:
            return x_forwarded_for.split(",")[0].strip()
        return request.META.get("REMOTE_ADDR")


class MaintenanceModeMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        import os
        if os.environ.get("MAINTENANCE_MODE") == "1":
            if not request.user.is_staff:
                from django.http import HttpResponse
                return HttpResponse(
                    "系统维护中，请稍后再试", status=503,
                )
        return self.get_response(request)

18.10 信号

18.10.1 内置信号

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from django.db.models.signals import post_save, pre_delete, m2m_changed
from django.dispatch import receiver
from django.contrib.auth import user_logged_in
from .models import Post, Comment, User


@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    if created:
        Profile.objects.create(user=instance)


@receiver(post_save, sender=Comment)
def notify_comment_author(sender, instance, created, **kwargs):
    if created and instance.post.author != instance.author:
        from .services import send_notification
        send_notification(
            recipient=instance.post.author,
            message=f"{instance.author} 评论了你的文章《{instance.post.title}》",
        )


@receiver(pre_delete, sender=Post)
def cleanup_post_files(sender, instance, **kwargs):
    if instance.featured_image:
        from django.core.files.storage import default_storage
        if default_storage.exists(instance.featured_image.name):
            default_storage.delete(instance.featured_image.name)


@receiver(user_logged_in)
def update_last_login_ip(sender, request, user, **kwargs):
    x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
    ip = x_forwarded_for.split(",")[0].strip() if x_forwarded_for else request.META.get("REMOTE_ADDR")
    user.last_login_ip = ip
    user.save(update_fields=["last_login_ip"])

18.10.2 自定义信号

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from django.dispatch import Signal

post_published = Signal()
post_viewed = Signal()


from django.dispatch import receiver
from .signals import post_published

@receiver(post_published)
def on_post_published(sender, post, **kwargs):
    from .services import update_search_index, notify_subscribers
    update_search_index(post)
    notify_subscribers(post)


from .signals import post_published

class PostDetailView(DetailView):
    def get_object(self, queryset=None):
        obj = super().get_object(queryset)
        if obj.status == Post.Status.PUBLISHED:
            post_viewed.send(
                sender=self.__class__,
                post=obj,
                request=self.request,
            )
        return obj

18.11 缓存

18.11.1 缓存配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
CACHES = {
    "default": {
        "BACKEND": "django.core.cache.backends.redis.RedisCache",
        "LOCATION": "redis://localhost:6379/0",
        "TIMEOUT": 300,
        "KEY_PREFIX": "mysite",
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
        },
    }
}

CACHE_MIDDLEWARE_ALIAS = "default"
CACHE_MIDDLEWARE_SECONDS = 600
CACHE_MIDDLEWARE_KEY_PREFIX = "mysite"

18.11.2 缓存使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from django.views.decorators.cache import cache_page, cache_control, never_cache
from django.core.cache import cache
from django.utils.decorators import method_decorator


@cache_page(60 * 5)
def post_list(request):
    posts = Post.published.select_related("author", "category").all()
    return render(request, "blog/post_list.html", {"posts": posts})


@method_decorator(cache_page(60 * 15), name="dispatch")
class PostDetailView(DetailView):
    model = Post


@never_cache
def user_dashboard(request):
    return render(request, "dashboard.html")


def get_sidebar_data():
    key = "sidebar:categories"
    data = cache.get(key)
    if data is None:
        data = Category.objects.annotate(
            post_count=Count("posts")
        ).filter(post_count__gt=0)
        cache.set(key, data, timeout=60 * 30)
    return data


def get_post_with_cache(slug):
    key = f"post:{slug}"
    post = cache.get(key)
    if post is None:
        post = get_object_or_404(
            Post.published.select_related("author", "category"),
            slug=slug,
        )
        cache.set(key, post, timeout=60 * 15)
    return post

18.12 测试

18.12.1 测试配置与工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from django.test import TestCase, Client, RequestFactory
from django.contrib.auth import get_user_model
from django.urls import reverse
from ..models import Post, Category

User = get_user_model()


class PostModelTest(TestCase):
    @classmethod
    def setUpTestData(cls):
        cls.user = User.objects.create_user(
            username="testauthor", password="TestPass123!"
        )
        cls.category = Category.objects.create(
            name="Python", slug="python"
        )
        cls.post = Post.objects.create(
            title="Test Post",
            slug="test-post",
            content="Test content for the post.",
            author=cls.user,
            category=cls.category,
            status=Post.Status.PUBLISHED,
        )

    def test_post_str(self):
        self.assertEqual(str(self.post), "Test Post")

    def test_post_absolute_url(self):
        self.assertEqual(
            self.post.get_absolute_url(),
            reverse("blog:post_detail", args=[self.post.slug]),
        )

    def test_published_manager(self):
        published_posts = Post.published.all()
        self.assertIn(self.post, published_posts)

    def test_draft_not_in_published(self):
        draft = Post.objects.create(
            title="Draft", slug="draft", content="Draft content",
            author=self.user, status=Post.Status.DRAFT,
        )
        published_posts = Post.published.all()
        self.assertNotIn(draft, published_posts)

18.12.2 视图测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class PostViewTest(TestCase):
    @classmethod
    def setUpTestData(cls):
        cls.user = User.objects.create_user(
            username="testauthor", password="TestPass123!"
        )
        cls.category = Category.objects.create(name="Python", slug="python")
        cls.post = Post.objects.create(
            title="Test Post", slug="test-post",
            content="Content", author=cls.user,
            category=cls.category, status=Post.Status.PUBLISHED,
        )

    def test_post_list_view(self):
        response = self.client.get(reverse("blog:post_list"))
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Test Post")

    def test_post_detail_view(self):
        response = self.client.get(
            reverse("blog:post_detail", args=[self.post.slug])
        )
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Test Post")

    def test_post_create_requires_login(self):
        response = self.client.get(reverse("blog:post_create"))
        self.assertEqual(response.status_code, 302)
        self.assertTrue(response.url.startswith("/users/login/"))

    def test_post_create_authenticated(self):
        self.client.login(username="testauthor", password="TestPass123!")
        response = self.client.get(reverse("blog:post_create"))
        self.assertEqual(response.status_code, 200)

    def test_post_create_post(self):
        self.client.login(username="testauthor", password="TestPass123!")
        response = self.client.post(reverse("blog:post_create"), {
            "title": "New Post",
            "content": "New content",
            "status": Post.Status.DRAFT,
        })
        self.assertEqual(response.status_code, 302)
        self.assertTrue(Post.objects.filter(title="New Post").exists())

    def test_post_update_by_author(self):
        self.client.login(username="testauthor", password="TestPass123!")
        response = self.client.post(
            reverse("blog:post_update", args=[self.post.slug]),
            {"title": "Updated Title", "content": "Updated content"},
        )
        self.assertEqual(response.status_code, 302)
        self.post.refresh_from_db()
        self.assertEqual(self.post.title, "Updated Title")

    def test_post_update_by_other_user_forbidden(self):
        other = User.objects.create_user(username="other", password="OtherPass123!")
        self.client.login(username="other", password="OtherPass123!")
        response = self.client.get(
            reverse("blog:post_update", args=[self.post.slug])
        )
        self.assertEqual(response.status_code, 403)

18.12.3 API测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
from rest_framework.test import APITestCase


class PostAPITest(APITestCase):
    @classmethod
    def setUpTestData(cls):
        cls.user = User.objects.create_user(
            username="apiuser", password="ApiPass123!"
        )
        cls.category = Category.objects.create(name="API", slug="api")
        cls.post = Post.objects.create(
            title="API Post", slug="api-post",
            content="API content", author=cls.user,
            category=cls.category, status=Post.Status.PUBLISHED,
        )

    def test_list_posts(self):
        response = self.client.get("/api/v1/posts/")
        self.assertEqual(response.status_code, 200)

    def test_retrieve_post(self):
        response = self.client.get(f"/api/v1/posts/{self.post.slug}/")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data["title"], "API Post")

    def test_create_post_unauthenticated(self):
        response = self.client.post("/api/v1/posts/", {
            "title": "Unauthorized", "content": "No auth",
        })
        self.assertEqual(response.status_code, 401)

    def test_create_post_authenticated(self):
        self.client.force_authenticate(user=self.user)
        response = self.client.post("/api/v1/posts/", {
            "title": "New API Post",
            "content": "New content",
            "status": "PB",
        })
        self.assertEqual(response.status_code, 201)

18.13 生产部署

18.13.1 安全配置清单

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
DEBUG = False
SECRET_KEY = os.environ["SECRET_KEY"]
ALLOWED_HOSTS = ["example.com", "www.example.com"]

SECURE_SSL_REDIRECT = True
SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")
SECURE_HSTS_SECONDS = 31536000
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True
SECURE_BROWSER_XSS_FILTER = True
SECURE_CONTENT_TYPE_NOSNIFF = True
SECURE_REFERRER_POLICY = "strict-origin-when-cross-origin"

SESSION_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_AGE = 3600 * 24 * 7

CSRF_COOKIE_SECURE = True
CSRF_COOKIE_HTTPONLY = True

X_FRAME_OPTIONS = "DENY"

PASSWORD_HASHERS = [
    "django.contrib.auth.hashers.Argon2PasswordHasher",
    "django.contrib.auth.hashers.PBKDF2PasswordHasher",
]

18.13.2 Gunicorn部署

1
2
3
4
5
6
7
8
pip install gunicorn
gunicorn config.wsgi:application \
    --bind 0.0.0.0:8000 \
    --workers 4 \
    --threads 2 \
    --timeout 120 \
    --access-logfile - \
    --error-logfile -

18.13.3 Docker部署

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
FROM python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

RUN python manage.py collectstatic --noinput

EXPOSE 8000

CMD ["gunicorn", "config.wsgi:application", "--bind", "0.0.0.0:8000", "--workers", "4"]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
version: "3.8"
services:
  web:
    build: .
    command: gunicorn config.wsgi:application --bind 0.0.0.0:8000 --workers 4
    volumes:
      - staticfiles:/app/staticfiles
      - mediafiles:/app/media
    environment:
      - DJANGO_SETTINGS_MODULE=config.settings.production
      - SECRET_KEY=${SECRET_KEY}
      - DATABASE_URL=postgresql://user:pass@db:5432/mysite
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - db
      - redis

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: mysite
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - staticfiles:/app/staticfiles:ro
      - mediafiles:/app/media:ro
    depends_on:
      - web

volumes:
  pgdata:
  staticfiles:
  mediafiles:

18.13.4 性能优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
DATABASES["default"]["CONN_MAX_AGE"] = 60

CACHES = {
    "default": {
        "BACKEND": "django.core.cache.backends.redis.RedisCache",
        "LOCATION": os.environ.get("REDIS_URL"),
    }
}

MIDDLEWARE = [
    "django.middleware.cache.UpdateCacheMiddleware",
    # ... other middleware ...
    "django.middleware.cache.FetchFromCacheMiddleware",
]

STORAGES = {
    "default": {
        "BACKEND": "django.core.files.storage.FileSystemStorage",
    },
    "staticfiles": {
        "BACKEND": "django.contrib.staticfiles.storage.ManifestStaticFilesStorage",
    },
}

18.14 前沿技术动态

18.14.1 Django 5.0+新特性

  • 异步视图支持：原生支持async def视图与中间件
  • 数据库计算的默认值：db_default参数支持数据库层面的默认值
  • 生成的字段（GeneratedField）：支持数据库计算列
  • 改进的ORM：更高效的查询生成与执行
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from django.db import models
from django.db.models import GeneratedField, Value


class Product(models.Model):
    price = models.DecimalField(max_digits=10, decimal_places=2)
    tax_rate = models.DecimalField(max_digits=5, decimal_places=2)
    total_with_tax = GeneratedField(
        expression=models.F("price") * (1 + models.F("tax_rate") / 100),
        output_field=models.DecimalField(max_digits=12, decimal_places=2),
        db_persist=True,
    )


async def async_post_list(request):
    posts = await Post.objects.filter(status="PB").select_related("author").all()
    return render(request, "blog/post_list.html", {"posts": posts})

18.14.2 Django与ASGI

Django 3.0+支持ASGI，可实现WebSocket、长轮询等异步功能：

1
2
3
4
5
6
from django.urls import path
from . import consumers

websocket_urlpatterns = [
    path("ws/chat/<str:room_name>/", consumers.ChatConsumer.as_asgi()),
]

18.15 本章小结

本章系统阐述了Django Web开发的核心知识体系：

  1. MTV架构：理解Django的设计哲学与请求处理流程
  2. 模型与ORM：掌握模型定义、高级查询、迁移管理与性能优化
  3. 视图系统：函数视图与类视图的灵活运用，混入类设计模式
  4. URL配置：命名空间、反向解析与RESTful路由设计
  5. 模板系统：继承体系、自定义标签过滤器与组件化开发
  6. 表单处理：ModelForm验证、文件上传与安全处理
  7. Admin定制：深度定制管理后台，自定义动作与展示
  8. 认证授权：用户认证、自定义认证后端与权限管理
  9. 中间件机制：洋葱模型、自定义中间件实现
  10. 信号系统：内置信号与自定义信号的使用场景
  11. 缓存策略：Redis缓存配置、视图缓存与数据缓存
  12. 测试体系：模型测试、视图测试与API测试
  13. 生产部署：安全配置、Docker容器化与性能优化

18.16 习题与项目练习

基础题

  1. 使用Django创建一个博客项目，实现文章的CRUD操作，包含分类和标签功能。

  2. 实现用户注册、登录、退出功能，要求支持邮箱或用户名登录。

  3. 设计一个在线书店的数据模型，包含书籍、作者、出版社、分类和订单五个模型及其关联关系。

进阶题

  1. 实现一个完整的评论系统，支持嵌套回复、评论审核和评论通知功能。

  2. 使用Django中间件实现请求限流、访问日志记录和异常捕获三个功能。

  3. 设计并实现一个基于Django ORM的全文搜索功能，支持中文分词和搜索结果高亮。

综合项目

  1. 内容管理系统（CMS）：构建一个功能完整的CMS，包含：

    • 多用户角色管理（管理员、编辑、作者、读者）
    • 文章发布与审核流程
    • 自定义页面与导航管理
    • 媒体文件管理
    • 站点配置管理
    • RESTful API
    • 完整的Admin定制
    • 缓存与性能优化

  2. 电商后台管理系统：构建一个电商后台，包含：

    • 商品管理（SPU/SKU模型）
    • 订单管理与状态流转
    • 库存管理与预警
    • 数据统计与报表
    • 权限管理
    • 操作日志审计

思考题

  1. Django ORM的select_related和prefetch_related在底层实现上有何本质区别？在什么场景下应优先选择哪种方式？请从SQL生成、内存占用和查询效率三个维度分析。

  2. 在高并发场景下，Django如何实现请求的异步处理？ASGI与WSGI的核心差异是什么？请设计一个混合使用同步视图与异步视图的方案。

18.17 延伸阅读

18.17.1 Django官方资源

18.17.2 进阶书籍

  • 《Two Scoops of Django》 (Daniel Feldroy) — Django最佳实践
  • 《Django 5 By Example》 (Antonio Melé) — 实战项目教程
  • 《Speed Up Your Django Tests》 (Adam Johnson) — 测试优化

18.17.3 扩展与生态

18.17.4 部署与运维

下一章：第19章 API开发