Django 进阶使用指南

1
posts = Post.objects.select_related("author").prefetch_related("tags", "comments")

1
# 只需要标题和发布时间，不拉正文
2
Post.objects.only("title", "pub_date")
3

4
# 只需要概要，跳过内容大字段
5
Post.objects.defer("body")

1
from django.db.models import F, Q, Count, Sum, Window
2
from django.db.models.functions import Rank
3

4
# F() 表达式：数据库端原子更新
5
Product.objects.filter(stock__gt=0).update(stock=F("stock") - 1)
6

7
# Q() 对象：复杂条件组合
8
Post.objects.filter(
9
    Q(status="published") & (Q(category="tech") | Q(tags__name="python"))
10
)
11

12
# annotate + 聚合：在数据库端完成统计
13
from django.db.models import Count
14

15
categories = Category.objects.annotate(post_count=Count("post")).filter(
16
    post_count__gt=0
17
)

1
from django.db.models import Window
2
from django.db.models.functions import Rank
3

4
posts_with_rank = Post.objects.annotate(
5
    rank=Window(
6
        expression=Rank(),
7
        order_by=F("views").desc(),
8
    )
9
)

1
MIDDLEWARE = [
2
    "django.middleware.security.SecurityMiddleware",  # 1. 安全头：HSTS、XSS 等
3
    "whitenoise.middleware.WhiteNoiseMiddleware",     # 2. 静态文件（尽量靠前）
4
    "django.contrib.sessions.middleware.SessionMiddleware",  # 3. 会话
5
    "corsheaders.middleware.CorsMiddleware",           # 4. CORS（必须在 Common 之前）
6
    "django.middleware.common.CommonMiddleware",       # 5. URL 规范化
7
    "django.middleware.csrf.CsrfViewMiddleware",       # 6. CSRF 保护
8
    "django.contrib.auth.middleware.AuthenticationMiddleware",  # 7. 用户认证
9
    "django.contrib.messages.middleware.MessageMiddleware",      # 8. 消息
10
    "django.middleware.clickjacking.XFrameOptionsMiddleware",    # 9. 点击劫持防护
11
]

1
import uuid
2

3

4
class RequestIDMiddleware:
5
    def __init__(self, get_response):
6
        self.get_response = get_response
7

8
    def __call__(self, request):
9
        request.id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
10
        response = self.get_response(request)
11
        response["X-Request-ID"] = request.id
12
        return response

1
# 信号方式：逻辑分散，隐式触发，难以追踪
2
from django.db.models.signals import post_save
3

4

5
@receiver(post_save, sender=Order)
6
def on_order_created(sender, instance, created, **kwargs):
7
    if created:
8
        send_confirmation_email(instance)
9
        update_inventory(instance)
10
        update_search_index(instance)

1
# Service 层：逻辑集中，显式调用，易于测试和调试
2
class OrderService:
3
    def place_order(self, user, items):
4
        order = Order.objects.create(user=user)
5
        for item in items:
6
            OrderItem.objects.create(order=order, product=item.product, quantity=item.qty)
7
        self._update_inventory(order)
8
        self._send_confirmation_email(order)
9
        self._update_search_index(order)
10
        return order

1
# 异步视图 + 异步 ORM
2
from django.http import JsonResponse
3
from .models import Post
4

5

6
async def post_detail(request, post_id: int):
7
    post = await Post.objects.filter(id=post_id).afirst()
8
    if not post:
9
        return JsonResponse({"error": "文章不存在"}, status=404)
10
    return JsonResponse({"title": post.title, "body": post.body})

1
import asyncio
2

3

4
async def dashboard(request):
5
    users, orders, revenue = await asyncio.gather(
6
        User.objects.acount(),
7
        Order.objects.filter(status="pending").acount(),
8
        Order.objects.filter(status="completed").aaggregate(total=Sum("amount")),
9
    )
10
    # 三个查询并发执行，总耗时 ≈ 最慢的一个，而非三者之和
11
    return JsonResponse({"users": users, "pending_orders": orders, "revenue": revenue})

1
from django.tasks import task
2

3

4
@task
5
def send_welcome_email(user_id: int):
6
    # 在后台执行，不阻塞请求
7
    ...

1
from rest_framework.viewsets import ModelViewSet
2
from rest_framework.decorators import action
3
from rest_framework.response import Response
4

5

6
class PostViewSet(ModelViewSet):
7
    queryset = Post.objects.select_related("author").prefetch_related("tags")
8
    serializer_class = PostSerializer
9
    permission_classes = [IsAuthenticatedOrReadOnly]
10

11
    @action(detail=True, methods=["post"])
12
    def publish(self, request, pk=None):
13
        post = self.get_object()
14
        post.status = "published"
15
        post.save()
16
        return Response({"status": "published"})
17

18
    @action(detail=False, methods=["get"])
19
    def popular(self, request):
20
        posts = self.get_queryset().filter(views__gt=1000)[:20]
21
        return Response(self.get_serializer(posts, many=True).data)

1
class PostViewSet(ModelViewSet):
2
    # 这个 queryset 会被 ViewSet 的所有 action 复用
3
    queryset = Post.objects.select_related("author").prefetch_related("tags")

1
from rest_framework.views import exception_handler
2

3

4
def custom_exception_handler(exc, context):
5
    response = exception_handler(exc, context)
6
    if response is not None:
7
        response.data = {
8
            "error": {
9
                "code": getattr(exc, "default_code", "error"),
10
                "message": response.data.get("detail", str(exc)),
11
            }
12
        }
13
    return response

1
REST_FRAMEWORK = {
2
    "EXCEPTION_HANDLER": "myapp.exceptions.custom_exception_handler",
3
    "DEFAULT_PAGINATION_CLASS": "rest_framework.pagination.CursorPagination",
4
    "PAGE_SIZE": 20,
5
    "DEFAULT_THROTTLE_CLASSES": [
6
        "rest_framework.throttling.AnonRateThrottle",
7
        "rest_framework.throttling.UserRateThrottle",
8
    ],
9
    "DEFAULT_THROTTLE_RATES": {"anon": "100/hour", "user": "1000/hour"},
10
}

1
from django.views.decorators.cache import cache_page
2

3

4
@cache_page(60 * 15)  # 缓存 15 分钟
5
def homepage(request):
6
    ...

1
{% load cache %}
2
{% cache 300 sidebar request.user.id %}
3
  <!-- 复杂的侧栏逻辑 -->
4
{% endcache %}

1
from django.core.cache import cache
2

3

4
def get_popular_posts():
5
    key = "popular_posts"
6
    posts = cache.get(key)
7
    if posts is None:
8
        posts = list(Post.objects.filter(views__gt=1000).order_by("-views")[:20])
9
        cache.set(key, posts, 300)  # 5 分钟
10
    return posts

1
from django.contrib import admin
2

3

4
@admin.register(Post)
5
class PostAdmin(admin.ModelAdmin):
6
    list_display = ["title", "author", "status", "pub_date", "views"]
7
    list_filter = ["status", "pub_date"]
8
    search_fields = ["title", "body"]
9
    readonly_fields = ["created_at", "updated_at", "views"]
10
    ordering = ["-pub_date"]
11
    date_hierarchy = "pub_date"
12
    actions = ["make_published", "make_draft"]
13

14
    @admin.action(description="发布选中文章")
15
    def make_published(self, request, queryset):
16
        queryset.update(status="published")

1
# settings.py — production.py
2
DEBUG = False
3
ALLOWED_HOSTS = ["yourdomain.com"]
4
SECURE_SSL_REDIRECT = True
5
SESSION_COOKIE_SECURE = True
6
CSRF_COOKIE_SECURE = True
7
SECURE_HSTS_SECONDS = 31536000
8
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
9
SECURE_HSTS_PRELOAD = True
10
SECURE_BROWSER_XSS_FILTER = True
11
SECURE_CONTENT_TYPE_NOSNIFF = True
12
X_FRAME_OPTIONS = "DENY"
13
CORS_ALLOW_ALL_ORIGINS = False
14
CORS_ALLOWED_ORIGINS = ["https://your-frontend.com"]

1
                     CDN（Cloudflare / 阿里云 CDN）
2
                              │
3
                     Nginx / Caddy（TLS 终结 + 静态文件）
4
                              │
5
              ┌───────────────┼───────────────┐
6
              │               │               │
7
        Gunicorn        Uvicorn/Daphne    Celery Worker
8
        (WSGI 主力)     (ASGI，如用 Channels)   │
9
              │               │               │
10
              └───────────────┼───────────────┘
11
                              │
12
                        PgBouncer（连接池）
13
                              │
14
                   ┌──────────┴──────────┐
15
                   │                     │
16
               PostgreSQL              Redis
17
               (主数据库)           (缓存/队列/会话)