Python高级Web框架Django从入门到精通

┌─────────┐     ┌─────────┐     ┌───────────┐     ┌─────────┐
│  用户   │────>│  url    │────>│  view     │────>│ model   │
│ 浏览器  │<────│ router  │<────│ 业务逻辑  │<────│ 数据访问 │
└─────────┘     └─────────┘     └─────┬─────┘     └─────────┘
                                     │
                                     v
                              ┌───────────┐
                              │ template  │
                              │  html渲染  │
                              └───────────┘

# 创建虚拟环境
python -m venv myenv

# 激活（macos / linux）
source myenv/bin/activate

# 激活（windows）
myenv\scripts\activate

# 退出虚拟环境
deactivate

pip install django==5.1

# 验证安装
python -m django --version

myproject/                  # 项目容器
├── manage.py               # 命令行工具
├── myproject/              # 项目配置目录
│   ├── __init__.py
│   ├── settings.py         # 项目配置
│   ├── urls.py             # 根 url 配置
│   ├── asgi.py             # asgi 入口
│   └── wsgi.py             # wsgi 入口
├── app1/                   # 应用 1
│   ├── models.py
│   ├── views.py
│   ├── urls.py
│   └── ...
└── app2/                   # 应用 2
    └── ...

django-admin startproject myproject
cd myproject

python manage.py startapp blog

# myproject/settings.py

installed_apps = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'blog',  # 注册自定义应用
]

# 启动开发服务器
python manage.py runserver

# 指定端口
python manage.py runserver 0.0.0.0:8000

# 数据库迁移
python manage.py makemigrations
python manage.py migrate

# 创建超级管理员
python manage.py createsuperuser

# 启动 shell
python manage.py shell

# 收集静态文件
python manage.py collectstatic

# 查看已注册的 sql
python manage.py sqlmigrate blog 0001

# blog/urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('articles/', views.article_list, name='article_list'),
    path('articles/<int:id>/', views.article_detail, name='article_detail'),
    path('articles/<int:year>/<int:month>/', views.article_archive, name='article_archive'),
]

# myproject/urls.py
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('blog/', include('blog.urls')),
    path('api/', include('api.urls')),
]

urlpatterns = [
    path('articles/<int:pk>/', views.detail),           # 整数
    path('articles/<slug:title>/', views.detail),        # slug 字符串
    path('articles/<uuid:id>/', views.detail),           # uuid
    path('files/<path:filepath>/', views.download),     # 路径（含 /）
]

from django.urls import re_path

urlpatterns = [
    re_path(r'^articles/(?p<year>\d{4})/$', views.year_archive),
    re_path(r'^articles/(?p<year>\d{4})/(?p<month>\d{2})/$', views.month_archive),
]

<a href="{% url 'article_detail' id=article.id %}" rel="external nofollow" >查看详情</a>

from django.urls import reverse
from django.shortcuts import redirect

def create_article(request):
    article = article.objects.create(title="test")
    return redirect(reverse('article_detail', args=[article.id]))

# 或简写
    return redirect('article_detail', pk=article.id)

# myproject/urls.py
urlpatterns = [
    path('blog/', include('blog.urls', namespace='blog')),
]

# 模板中使用
<a href="{% url 'blog:article_detail' pk=1 %}" rel="external nofollow" >详情</a>

from django.shortcuts import render, get_object_or_404, redirect
from django.http import httpresponse, jsonresponse, http404
from .models import article

def article_list(request):
    articles = article.objects.all().order_by('-created_at')
    return render(request, 'blog/article_list.html', {'articles': articles})

def article_detail(request, pk):
    article = get_object_or_404(article, pk=pk)
    return render(request, 'blog/article_detail.html', {'article': article})

def create_article(request):
    if request.method == 'post':
        title = request.post.get('title')
        content = request.post.get('content')
        article.objects.create(title=title, content=content)
        return redirect('article_list')
    return render(request, 'blog/article_form.html')

def demo(request):
    # 请求方法
    request.method          # 'get', 'post', 'put', 'delete'

    # get 参数
    request.get.get('page', 1)

    # post 数据（表单）
    request.post.get('username')

    # json 请求体
    import json
    data = json.loads(request.body)

    # 文件上传
    file = request.files.get('avatar')

    # 请求头
    request.meta.get('http_user_agent')

    # cookies
    request.cookies.get('session_id')

    # 用户信息
    request.user            # 当前登录用户（需要中间件）

    # 完整 url
    request.get_full_path()

# html 响应
from django.shortcuts import render
return render(request, 'template.html', context={'key': 'value'})

# json 响应
from django.http import jsonresponse
return jsonresponse({'status': 'ok', 'data': list})

# 重定向
from django.shortcuts import redirect
return redirect('/some/url/')
return redirect('named_url', arg1=value1)

# 文件下载
from django.http import fileresponse
return fileresponse(open('file.pdf', 'rb'), as_attachment=true)

# 自定义响应头
from django.http import httpresponse
response = httpresponse('content')
response['x-custom-header'] = 'value'
return response

from django.views.generic import listview, detailview, createview, updateview, deleteview
from django.urls import reverse_lazy
from .models import article
from .forms import articleform

class articlelistview(listview):
    model = article
    template_name = 'blog/article_list.html'
    context_object_name = 'articles'
    paginate_by = 10
    ordering = ['-created_at']

class articledetailview(detailview):
    model = article
    template_name = 'blog/article_detail.html'
    context_object_name = 'article'

class articlecreateview(createview):
    model = article
    form_class = articleform
    template_name = 'blog/article_form.html'
    success_url = reverse_lazy('article_list')

class articleupdateview(updateview):
    model = article
    form_class = articleform
    template_name = 'blog/article_form.html'

class articledeleteview(deleteview):
    model = article
    success_url = reverse_lazy('article_list')

urlpatterns = [
    path('articles/', articlelistview.as_view(), name='article_list'),
    path('articles/<int:pk>/', articledetailview.as_view(), name='article_detail'),
    path('articles/create/', articlecreateview.as_view(), name='article_create'),
    path('articles/<int:pk>/edit/', articleupdateview.as_view(), name='article_update'),
    path('articles/<int:pk>/delete/', articledeleteview.as_view(), name='article_delete'),
]

from django.contrib.auth.mixins import loginrequiredmixin, userpassestestmixin
from django.views.generic import updateview

class articleupdateview(loginrequiredmixin, userpassestestmixin, updateview):
    model = article
    form_class = articleform

    def test_func(self):
        """只有文章作者才能编辑"""
        return self.get_object().author == self.request.user

    def form_valid(self, form):
        """自动设置作者"""
        form.instance.author = self.request.user
        return super().form_valid(form)

from django.contrib.auth.decorators import login_required, permission_required
from django.views.decorators.http import require_http_methods, require_post
from django.views.decorators.csrf import csrf_exempt

@login_required(login_url='/login/')
def dashboard(request):
    ...

@permission_required('blog.change_article', raise_exception=true)
def edit_article(request, pk):
    ...

@require_post
def like_article(request, pk):
    ...

@csrf_exempt  # 豁免 csrf 验证（用于 api）
def api_endpoint(request):
    ...

# settings.py
templates = [
    {
        'backend': 'django.template.backends.django.djangotemplates',
        'dirs': [base_dir / 'templates'],     # 模板目录
        'app_dirs': true,                      # 自动搜索各 app 的 templates 目录
        'options': {
            'context_processors': [
                'django.template.context_processors.debug',
                'django.template.context_processors.request',
                'django.contrib.auth.context_processors.auth',
                'django.contrib.messages.context_processors.messages',
            ],
        },
    },
]

<!-- templates/base.html -->
<!doctype html>
<html lang="zh-cn">
<head>
    <meta charset="utf-8">
    <title>{% block title %}my site{% endblock %}</title>
    {% load static %}
    <link rel="stylesheet" href="{% static 'css/style.css' %}" rel="external nofollow"  rel="external nofollow" >
    {% block extra_css %}{% endblock %}
</head>
<body>
    <header>{% include 'header.html' %}</header>

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

    <footer>{% include 'footer.html' %}</footer>

    {% block extra_js %}{% endblock %}
</body>
</html>

<!-- blog/templates/blog/article_list.html -->
{% extends 'base.html' %}

{% block title %}文章列表 - my blog{% endblock %}

{% block content %}
<h1>文章列表</h1>
<ul>
    {% for article in articles %}
    <li>
        <a href="{{ article.get_absolute_url }}" rel="external nofollow" >{{ article.title }}</a>
        <span>{{ article.created_at|date:"y-m-d" }}</span>
    </li>
    {% empty %}
    <li>暂无文章</li>
    {% endfor %}
</ul>
{% endblock %}

<!-- 变量输出 -->
{{ article.title }}

<!-- 过滤器 -->
{{ content|truncatewords:30 }}          <!-- 截断为 30 个单词 -->
{{ price|floatformat:2 }}              <!-- 保留两位小数 -->
{{ date|date:"y年m月d日" }}            <!-- 日期格式化 -->
{{ name|default:"匿名" }}              <!-- 默认值 -->
{{ content|safe }}                      <!-- 渲染 html（不转义） -->
{{ items|length }}                      <!-- 长度 -->
{{ list|join:", " }}                   <!-- 用逗号连接 -->
{{ value|upper }}                       <!-- 转大写 -->
{{ content|linebreaks }}                <!-- 换行转 
 -->
{{ value|filesizeformat }}              <!-- 文件大小格式化 -->

<!-- 逻辑标签 -->
{% if user.is_authenticated %}
    <p>欢迎, {{ user.username }}</p>
{% else %}
    <p>请 <a href="{% url 'login' %}" rel="external nofollow" >登录</a></p>
{% endif %}

{% for item in items %}
    {{ forloop.counter }}. {{ item.name }}
{% empty %}
    暂无数据
{% endfor %}

{% for i in "xxxx" %}
    {{ i }}
{% endfor %}

<!-- url 反向解析 -->
{% url 'article_detail' pk=article.id %}

<!-- 静态文件 -->
{% load static %}
<link href="{% static 'css/style.css' %}" rel="external nofollow"  rel="external nofollow"  rel="stylesheet">
<img src="{% static 'images/logo.png' %}">

<!-- 注释 -->
{# 单行注释 #}
{% comment %}
多行注释
{% endcomment %}

<!-- 自定义标签/过滤器需要先 load -->
{% load my_tags %}
{{ value|my_filter }}
{% my_tag arg1 arg2 %}

# blog/templatetags/__init__.py        # 空文件
# blog/templatetags/blog_extras.py

from django import template

register = template.library()

@register.filter
def truncate_chars(value, max_length):
    if len(value) > max_length:
        return value[:max_length] + '...'
    return value

@register.filter(name='add_class')
def add_css_class(value, css_class):
    return value.as_widget(attrs={'class': css_class})

@register.simple_tag
def get_recent_articles(count=5):
    from blog.models import article
    return article.objects.order_by('-created_at')[:count]

{% load blog_extras %}

{{ article.content|truncate_chars:100 }}
{{ form.username|add_class:"form-control" }}
{% get_recent_articles 10 as recent %}

from django.db import models
from django.contrib.auth.models import user

class category(models.model):
    name = models.charfield('分类名', max_length=100, unique=true)
    description = models.textfield('描述', blank=true, default='')
    created_at = models.datetimefield('创建时间', auto_now_add=true)

    class meta:
        verbose_name = '分类'
        verbose_name_plural = '分类'
        ordering = ['name']

    def __str__(self):
        return self.name


class article(models.model):
    status_choices = [
        ('draft', '草稿'),
        ('published', '已发布'),
        ('archived', '已归档'),
    ]

    title = models.charfield('标题', max_length=200)
    content = models.textfield('内容')
    summary = models.charfield('摘要', max_length=500, blank=true)
    author = models.foreignkey(user, on_delete=models.cascade, verbose_name='作者')
    category = models.foreignkey(category, on_delete=models.set_null, null=true, blank=true)
    tags = models.manytomanyfield('tag', blank=true, verbose_name='标签')
    status = models.charfield('状态', max_length=20, choices=status_choices, default='draft')
    view_count = models.positiveintegerfield('浏览量', default=0)
    created_at = models.datetimefield('创建时间', auto_now_add=true)
    updated_at = models.datetimefield('更新时间', auto_now=true)
    is_deleted = models.booleanfield('已删除', default=false)

    class meta:
        verbose_name = '文章'
        verbose_name_plural = '文章'
        ordering = ['-created_at']
        indexes = [
            models.index(fields=['-created_at']),
            models.index(fields=['author', 'status']),
        ]

    def __str__(self):
        return self.title

    def get_absolute_url(self):
        from django.urls import reverse
        return reverse('article_detail', args=[self.id])

# 字符串
charfield(max_length=n)           # 短字符串
textfield()                       # 长文本

# 数字
integerfield()                     # 整数
bigintegerfield()                  # 大整数
floatfield()                      # 浮点数
decimalfield(max_digits=n, decimal_places=m)  # 精确小数
positiveintegerfield()            # 正整数
positivesmallintegerfield()       # 小正整数

# 布尔
booleanfield()                    # 布尔值
nullbooleanfield()                 # 可为空的布尔值（django 4+ 已废弃，用 booleanfield(null=true)）

# 日期时间
datefield()                       # 日期
datetimefield()                   # 日期时间
timefield()                       # 时间
durationfield()                   # 时间间隔

# 文件
filefield(upload_to='uploads/')   # 文件
imagefield(upload_to='images/')   # 图片（需要 pillow）

# 关系字段
foreignkey(model, on_delete=...)   # 多对一
onetoonefield(model, on_delete=...)  # 一对一
manytomanyfield(model)             # 多对多

# 特殊
uuidfield()                       # uuid
slugfield()                       # url 友好字符串
genericipaddressfield()            # ip 地址
jsonfield()                       # json 数据
binaryfield()                     # 二进制数据

# 常用参数
null=true                         # 数据库允许 null
blank=true                        # 表单验证允许为空
default=n                         # 默认值
unique=true                       # 唯一约束
db_index=true                     # 数据库索引
choices=[...]                     # 选项
verbose_name='中文'                # 字段人类可读名称
help_text='帮助文本'               # 表单帮助文本
auto_now=true                     # 每次保存自动更新
auto_now_add=true                 # 创建时自动设置

# 一对多：foreignkey
class article(models.model):
    author = models.foreignkey(user, on_delete=models.cascade)
    category = models.foreignkey(category, on_delete=models.set_null, null=true)

# on_delete 选项
cascade     # 级联删除
set_null    # 设为 null（需要 null=true）
set_default # 设为默认值（需要 default）
protect     # 阻止删除，抛异常
do_nothing  # 不做任何操作

# 反向查询
user.articles.all()             # 默认: model_name_set
user.article_set.all()

# 自定义反向查询名称
author = models.foreignkey(user, on_delete=models.cascade, related_name='articles')

# 多对多：manytomanyfield
class article(models.model):
    tags = models.manytomanyfield('tag', blank=true)

article.tags.add(tag)
article.tags.remove(tag)
article.tags.clear()
article.tags.set([tag1, tag2])

# 多对多带额外字段
class membership(models.model):
    person = models.foreignkey(person, on_delete=models.cascade)
    group = models.foreignkey(group, on_delete=models.cascade)
    date_joined = models.datefield()

class person(models.model):
    groups = models.manytomanyfield(group, through='membership')

# 基础查询
article.objects.all()                         # 所有记录
article.objects.filter(status='published')     # 过滤
article.objects.exclude(status='draft')        # 排除
article.objects.get(pk=1)                      # 获取单个（不存在抛异常）
article.objects.get_or_create(title='test')    # 获取或创建

# 链式查询
article.objects.filter(status='published').filter(author__username='admin')

# 字段查找
article.objects.filter(title__contains='django')
article.objects.filter(title__icontains='django')      # 不区分大小写
article.objects.filter(title__startswith='django')
article.objects.filter(title__endswith='教程')
article.objects.filter(view_count__gte=100)            # 大于等于
article.objects.filter(view_count__lt=100)             # 小于
article.objects.filter(created_at__year=2024)
article.objects.filter(created_at__date__gte='2024-01-01')
article.objects.filter(title__in=['django', 'flask'])
article.objects.filter(title__isnull=true)

# q 对象（复杂查询）
from django.db.models import q

article.objects.filter(q(title__contains='django') | q(title__contains='flask'))
article.objects.filter(q(status='published') & ~q(author__username='spam'))

# f 对象（字段间比较）
from django.db.models import f

article.objects.filter(view_count__gt=f('comment_count') * 10)
article.objects.update(view_count=f('view_count') + 1)

# 排序
article.objects.order_by('-created_at')
article.objects.order_by('created_at', '-title')

# 切片（limit / offset）
article.objects.all()[:10]           # 前 10 条
article.objects.all()[10:20]         # 11-20 条

# 聚合
from django.db.models import count, sum, avg, max, min

article.objects.aggregate(avg_views=avg('view_count'))
article.objects.annotate(comment_count=count('comment'))

# 分组统计
from django.db.models import count

category.objects.annotate(article_count=count('article'))

# 去重
article.objects.filter(tags__name='python').distinct()

# 日期查询
from django.utils import timezone
article.objects.filter(created_at__gte=timezone.now() - timedelta(days=7))

# 选择字段
article.objects.values('id', 'title')
article.objects.values_list('id', 'title')
article.objects.values_list('id', flat=true)   # 只返回单值列表

# exists
article.objects.filter(title='test').exists()  # 是否存在

# 批量操作
article.objects.filter(status='draft').delete()
article.objects.filter(status='draft').update(status='archived')
article.objects.bulk_create([...])              # 批量创建

# manager.raw() —— 返回 model 实例
for article in article.objects.raw('select * from blog_article where status = %s', ['published']):
    print(article.title)

# connection.cursor() —— 执行任意 sql
from django.db import connection

with connection.cursor() as cursor:
    cursor.execute("select count(*) from blog_article where status = %s", ['published'])
    count = cursor.fetchone()[0]

# 生成迁移文件
python manage.py makemigrations

# 执行迁移
python manage.py migrate

# 查看迁移状态
python manage.py showmigrations

# 回滚到指定迁移
python manage.py migrate blog 0002

# 查看迁移 sql
python manage.py sqlmigrate blog 0003

# migrations/0004_add_initial_data.py
from django.db import migrations

def create_initial_categories(apps, schema_editor):
    category = apps.get_model('blog', 'category')
    category.objects.bulk_create([
        category(name='python'),
        category(name='django'),
        category(name='数据库'),
    ])

class migration(migrations.migration):
    dependencies = [
        ('blog', '0003_auto_20240101_0000'),
    ]
    operations = [
        migrations.runpython(create_initial_categories),
    ]

# blog/admin.py
from django.contrib import admin
from .models import article, category, tag

admin.site.register(category)
admin.site.register(tag)

@admin.register(article)
class articleadmin(admin.modeladmin):
    list_display = ['id', 'title', 'author', 'status', 'view_count', 'created_at']
    list_display_links = ['id', 'title']
    list_filter = ['status', 'category', 'created_at']
    search_fields = ['title', 'content']
    list_editable = ['status']
    list_per_page = 20
    ordering = ['-created_at']
    date_hierarchy = 'created_at'
    actions = ['make_published', 'make_draft']
    filter_horizontal = ['tags']

    fieldsets = (
        ('基本信息', {
            'fields': ('title', 'author', 'category', 'tags')
        }),
        ('内容', {
            'fields': ('content', 'summary'),
            'classes': ('wide',)
        }),
        ('设置', {
            'fields': ('status',)
        }),
    )

    readonly_fields = ['view_count', 'created_at', 'updated_at']

    @admin.display(description='是否热门')
    def is_hot(self, obj):
        return obj.view_count > 1000
    is_hot.boolean = true

    @admin.action(description='发布选中文章')
    def make_published(self, request, queryset):
        updated = queryset.update(status='published')
        self.message_user(request, f'{updated} 篇文章已发布')

    @admin.action(description='转为草稿')
    def make_draft(self, request, queryset):
        updated = queryset.update(status='draft')
        self.message_user(request, f'{updated} 篇文章已转为草稿')

# blog/forms.py
from django import forms
from .models import article, comment

class articleform(forms.modelform):
    class meta:
        model = article
        fields = ['title', 'category', 'tags', 'content', 'summary']
        widgets = {
            'content': forms.textarea(attrs={'class': 'editor', 'rows': 10}),
            'tags': forms.checkboxselectmultiple(),
        }

    def clean_title(self):
        title = self.cleaned_data['title']
        if len(title) < 5:
            raise forms.validationerror('标题至少 5 个字符')
        return title

from django.shortcuts import render, redirect
from .forms import articleform

def create_article(request):
    if request.method == 'post':
        form = articleform(request.post)
        if form.is_valid():
            article = form.save(commit=false)
            article.author = request.user
            article.save()
            form.save_m2m()  # 保存多对多关系
            return redirect(article)
    else:
        form = articleform()
    return render(request, 'blog/article_form.html', {'form': form})

<form method="post">
    {% csrf_token %}
    {{ form.as_p }}

    {# 手动渲染每个字段 #}
    <div>
        {{ form.title.label_tag }}
        {{ form.title }}
        {% if form.title.errors %}
            <span class="error">{{ form.title.errors }}</span>
        {% endif %}
    </div>

    <button type="submit">保存</button>
</form>

class contactform(forms.form):
    name = forms.charfield(max_length=100, label='姓名')
    email = forms.emailfield(label='邮箱')
    message = forms.charfield(widget=forms.textarea, label='留言')

    def clean_email(self):
        email = self.cleaned_data['email']
        if not email.endswith('@company.com'):
            raise forms.validationerror('请使用公司邮箱')
        return email

# middleware.py
import time
import logging

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
        logger.info(f'{request.method} {request.path} - {duration:.3f}s')
        return response


class apikeymiddleware:
    """api key 验证"""

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

    def __call__(self, request):
        if request.path.startswith('/api/'):
            api_key = request.meta.get('http_x_api_key')
            if not api_key or api_key != 'expected-key':
                from django.http import jsonresponse
                return jsonresponse({'error': 'invalid api key'}, status=401)
        return self.get_response(request)

# 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',
    'middleware.requesttimingmiddleware',
    'middleware.apikeymiddleware',
]

pip install djangorestframework

# blog/serializers.py
from rest_framework import serializers
from .models import article, category

class categoryserializer(serializers.modelserializer):
    article_count = serializers.integerfield(read_only=true)

    class meta:
        model = category
        fields = ['id', 'name', 'description', 'article_count']


class articlelistserializer(serializers.modelserializer):
    category_name = serializers.charfield(source='category.name', read_only=true)
    author_name = serializers.charfield(source='author.username', read_only=true)

    class meta:
        model = article
        fields = ['id', 'title', 'author_name', 'category_name', 'status',
                  'view_count', 'created_at']


class articledetailserializer(serializers.modelserializer):
    category = categoryserializer(read_only=true)
    tags = serializers.stringrelatedfield(many=true, read_only=true)
    author_name = serializers.charfield(source='author.username', read_only=true)

    class meta:
        model = article
        fields = '__all__'

# blog/api_views.py
from rest_framework import viewsets, status, permissions
from rest_framework.decorators import action
from rest_framework.response import response
from rest_framework.permissions import isauthenticatedorreadonly
from .models import article
from .serializers import articlelistserializer, articledetailserializer

class articleviewset(viewsets.modelviewset):
    queryset = article.objects.select_related('author', 'category').prefetch_related('tags')
    permission_classes = [isauthenticatedorreadonly]

    def get_serializer_class(self):
        if self.action == 'list':
            return articlelistserializer
        return articledetailserializer

    def perform_create(self, serializer):
        serializer.save(author=self.request.user)

    @action(detail=false, methods=['get'])
    def published(self, request):
        articles = self.queryset.filter(status='published')
        serializer = self.get_serializer(articles, many=true)
        return response(serializer.data)

    @action(detail=true, methods=['post'])
    def publish(self, request, pk=none):
        article = self.get_object()
        article.status = 'published'
        article.save()
        return response({'status': 'published'})

# myproject/urls.py
from rest_framework.routers import defaultrouter
from blog.api_views import articleviewset

router = defaultrouter()
router.register(r'articles', articleviewset, basename='article')

urlpatterns = [
    path('api/', include(router.urls)),
    path('api-auth/', include('rest_framework.urls')),
]

# settings.py
rest_framework = {
    'default_pagination_class': 'rest_framework.pagination.pagenumberpagination',
    'page_size': 10,
    'default_filter_backends': [
        'rest_framework.filters.searchfilter',
        'rest_framework.filters.orderingfilter',
    ],
    'default_authentication_classes': [
        'rest_framework.authentication.tokenauthentication',
        'rest_framework.authentication.sessionauthentication',
    ],
}

# 视图中启用搜索和排序
class articleviewset(viewsets.modelviewset):
    search_fields = ['title', 'content']
    ordering_fields = ['created_at', 'view_count']
    ordering = ['-created_at']

from django.contrib.auth import authenticate, login, logout
from django.contrib.auth.decorators import login_required

def login_view(request):
    if request.method == 'post':
        username = request.post.get('username')
        password = request.post.get('password')
        user = authenticate(request, username=username, password=password)
        if user is not none:
            if user.is_active:
                login(request, user)
                return redirect('dashboard')
        return render(request, 'login.html', {'error': '用户名或密码错误'})
    return render(request, 'login.html')

@login_required
def dashboard(request):
    return render(request, 'dashboard.html')

def logout_view(request):
    logout(request)
    return redirect('login')

# auth/backends.py
from django.contrib.auth.backends import modelbackend
from django.db.models import q

class emailorusernamebackend(modelbackend):
    """支持邮箱或用户名登录"""

    def authenticate(self, request, username=none, password=none, **kwargs):
        if username is none:
            username = kwargs.get('username')
        if password is none:
            password = kwargs.get('password')

        usermodel = get_user_model()
        try:
            user = usermodel.objects.get(q(username=username) | q(email=username))
            if user.check_password(password) and self.user_can_authenticate(user):
                return user
        except usermodel.doesnotexist:
            return none

# settings.py
authentication_backends = [
    'auth.backends.emailorusernamebackend',
    'django.contrib.auth.backends.modelbackend',
]

pip install rest_framework.authtoken

# 自动创建 token
from django.db.models.signals import post_save
from django.dispatch import receiver
from rest_framework.authtoken.models import token

@receiver(post_save, sender=user)
def create_auth_token(sender, instance=none, created=false, **kwargs):
    if created:
        token.objects.create(user=instance)

# 客户端请求时携带 header: authorization: token xxxxxxxx

# settings.py

# 方式 1：本地内存缓存（开发用）
caches = {
    'default': {
        'backend': 'django.core.cache.backends.locmem.locmemcache',
        'location': 'unique-snowflake',
    }
}

# 方式 2：redis 缓存（生产用）
caches = {
    'default': {
        'backend': 'django.core.cache.backends.redis.rediscache',
        'location': 'redis://127.0.0.1:6379/1',
    }
}

from django.core.cache import cache

# 基础用法
cache.set('key', 'value', timeout=60)
data = cache.get('key')
cache.delete('key')
cache.set_many({'key1': 'val1', 'key2': 'val2'})
data = cache.get_many(['key1', 'key2'])

# 视图缓存装饰器
from django.views.decorators.cache import cache_page

@cache_page(60 * 15)  # 缓存 15 分钟
def article_list(request):
    ...

# 模板片段缓存
{% load cache %}
{% cache 500 sidebar %}
    {% get_recent_articles as recent %}
    {% for article in recent %}
        <li>{{ article.title }}</li>
    {% endfor %}
{% endcache %}

from django.db.models.signals import post_save, pre_delete
from django.dispatch import receiver
from .models import article

@receiver(post_save, sender=article)
def on_article_created(sender, instance, created, **kwargs):
    if created:
        from .tasks import send_notification
        send_notification.delay(instance.id)
        # 清除相关缓存
        from django.core.cache import cache
        cache.delete('article_list')

@receiver(pre_delete, sender=article)
def on_article_delete(sender, instance, **kwargs):
    # 删除前清理关联数据
    instance.tags.clear()

# signals.py
from django.dispatch import signal

article_viewed = signal()  # 文章被浏览

# 在视图或模型中发送信号
article_viewed.send(sender=article, article=article, user=request.user)

# 接收信号
@receiver(article_viewed, sender=article)
def track_article_view(sender, article, user, **kwargs):
    article.objects.filter(pk=article.pk).update(view_count=f('view_count') + 1)

pip install celery redis

# myproject/celery.py
import os
from celery import celery

os.environ.setdefault('django_settings_module', 'myproject.settings')

app = celery('myproject')
app.config_from_object('django.conf:settings', namespace='celery')
app.autodiscover_tasks()

# settings.py
celery_broker_url = 'redis://localhost:6379/0'
celery_result_backend = 'redis://localhost:6379/0'
celery_accept_content = ['json']
celery_task_serializer = 'json'
celery_result_serializer = 'json'
celery_timezone = 'asia/shanghai'

# blog/tasks.py
from celery import shared_task
from django.core.mail import send_mail

@shared_task
def send_welcome_email(user_id):
    from django.contrib.auth.models import user
    user = user.objects.get(id=user_id)
    send_mail(
        '欢迎加入',
        f'你好 {user.username}，感谢注册！',
        'noreply@example.com',
        [user.email],
    )

@shared_task
def generate_article_summary(article_id):
    from .models import article
    article = article.objects.get(id=article_id)
    summary = article.content[:200] + '...'
    article.summary = summary
    article.save()

# 调用任务
send_welcome_email.delay(user.id)
generate_article_summary.apply_async(args=[article.id], countdown=60)  # 60秒后执行

# 启动 worker
celery -a myproject worker -l info

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

# myproject/celery.py
from celery.schedules import crontab

app.conf.beat_schedule = {
    'cleanup-expired-sessions': {
        'task': 'myproject.tasks.cleanup_expired_sessions',
        'schedule': crontab(hour=3, minute=0),  # 每天凌晨 3 点
    },
    'send-daily-report': {
        'task': 'myproject.tasks.send_daily_report',
        'schedule': crontab(hour=8, minute=0, day_of_week='mon-fri'),
    },
}

# settings.py
debug = false

allowed_hosts = ['yourdomain.com', 'www.yourdomain.com']

secure_ssl_redirect = true
session_cookie_secure = true
csrf_cookie_secure = true

# settings.py
static_url = '/static/'
static_root = base_dir / 'staticfiles'

staticfiles_dirs = [
    base_dir / 'static',
]

python manage.py collectstatic

pip install gunicorn

gunicorn myproject.wsgi:application \
    --bind 0.0.0.0:8000 \
    --workers 4 \
    --worker-class gthread \
    --threads 2 \
    --timeout 120 \
    --access-logfile - \
    --error-logfile -

pip install uvicorn

uvicorn myproject.asgi:application --host 0.0.0.0 --port 8000 --workers 4

server {
    listen 80;
    server_name yourdomain.com;
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl;
    server_name yourdomain.com;

    ssl_certificate /path/to/cert.pem;
    ssl_certificate_key /path/to/key.pem;

    location /static/ {
        alias /path/to/staticfiles/;
        expires 30d;
    }

    location /media/ {
        alias /path/to/media/;
    }

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header host $host;
        proxy_set_header x-real-ip $remote_addr;
        proxy_set_header x-forwarded-for $proxy_add_x_forwarded_for;
        proxy_set_header x-forwarded-proto $scheme;
    }
}

1. 数据库优化
   - 使用 select_related() 减少一对多查询的 sql 次数
   - 使用 prefetch_related() 减少多对多查询的 sql 次数
   - 添加合适的数据库索引
   - 使用 only() / defer() 限制查询字段
   - 使用 iterator() 处理大数据集，减少内存

2. 缓存策略
   - 视图缓存：@cache_page
   - 模板片段缓存：{% cache %}
   - 数据查询缓存：cache.set/get
   - 使用 redis 作为缓存后端

3. 查询优化
   - 避免在循环中执行查询（n+1 问题）
   - 使用 bulk_create 批量插入
   - 使用 update() 批量更新
   - 使用 count() 或 exists() 代替 len()

4. 模板优化
   - 使用 {% load cached %} 缓存模板加载
   - 使用 compressed 模板加载器
   - 避免在模板中执行复杂逻辑

5. 部署优化
   - 开启 gzip 压缩
   - 配置 cdn 托管静态文件
   - 使用多个 worker 进程
   - 设置合适的 worker 超时时间

myproject/
├── manage.py
├── myproject/
│   ├── settings/
│   │   ├── __init__.py
│   │   ├── base.py           # 公共配置
│   │   ├── development.py    # 开发环境
│   │   ├── testing.py        # 测试环境
│   │   └── production.py     # 生产环境
│   ├── urls.py
│   └── wsgi.py
├── apps/
│   ├── __init__.py
│   ├── blog/                 # 博客应用
│   ├── users/                # 用户应用
│   └── common/               # 公共工具
├── config/
│   ├── celery.py
│   └── middleware.py
├── templates/
│   ├── base.html
│   └── blog/
├── static/
│   ├── css/
│   ├── js/
│   └── images/
├── tests/
│   ├── test_blog/
│   └── test_users/
└── requirements/
    ├── base.txt
    ├── dev.txt
    └── prod.txt

# settings/base.py
import os
base_dir = path(__file__).resolve().parent.parent
secret_key = os.environ.get('django_secret_key')
installed_apps = [...]
databases = {...}

# settings/development.py
from .base import *
debug = true
allowed_hosts = ['*']

# settings/production.py
from .base import *
debug = false
allowed_hosts = os.environ.get('allowed_hosts', '').split(',')

# 启动时指定
django_settings_module=myproject.settings.production python manage.py runserver

# 1. n+1 查询问题
# 错误：循环中执行查询
for article in article.objects.all():
    print(article.category.name)  # 每次循环都会查询数据库

# 正确：使用 select_related
for article in article.objects.select_related('category').all():
    print(article.category.name)  # 只执行 1 次 sql

# 2. 修改查询集不会立即执行
qs = article.objects.all()
qs = qs.filter(status='published')  # 不会查数据库
list(qs)  # 此时才执行查询

# 3. 时区问题
from django.utils import timezone
now = timezone.now()  # 始终使用 aware datetime

# 4. 不要在视图外直接调用 orm
# models.py 中不要导入 views.py 中的内容，避免循环引用

# 5. 表单中 csrf 保护
# post 请求模板中必须包含 {% csrf_token %}

# 6. 静态文件在 debug=false 时需要 collectstatic

# 7. migration 冲突
# 多人开发时，生成 migration 后及时提交
# 如果冲突，删除多余文件，重新生成

# tests/test_blog.py
from django.test import testcase, client
from django.urls import reverse
from .models import article

class articlemodeltest(testcase):
    def setup(self):
        self.article = article.objects.create(
            title='测试文章',
            content='测试内容',
        )

    def test_article_creation(self):
        self.assertequal(self.article.title, '测试文章')
        self.assertisnotnone(self.article.created_at)

    def test_article_str(self):
        self.assertequal(str(self.article), '测试文章')


class articleviewtest(testcase):
    def setup(self):
        self.client = client()
        article.objects.create(title='文章1', status='published')
        article.objects.create(title='文章2', status='draft')

    def test_article_list(self):
        response = self.client.get('/blog/articles/')
        self.assertequal(response.status_code, 200)
        self.assertcontains(response, '文章1')

    def test_article_detail(self):
        article = article.objects.first()
        response = self.client.get(f'/blog/articles/{article.id}/')
        self.assertequal(response.status_code, 200)

    def test_create_article_requires_login(self):
        response = self.client.get('/blog/articles/create/')
        self.assertequal(response.status_code, 302)  # 重定向到登录页


# 使用 pytest-django
# pip install pytest-django
# conftest.py
import pytest
from django.contrib.auth.models import user

@pytest.fixture
def user(db):
    return user.objects.create_user(username='test', password='pass123')

@pytest.fixture
def auth_client(client, user):
    client.force_login(user)
    return client