Django 5.0新特性全面解析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# views.py - 异步视图
from django.http import JsonResponse
import aiohttp

async def async_weather_view(request):
    """异步获取天气数据"""
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.weather.com/current') as response:
            weather_data = await response.json()

    return JsonResponse(weather_data)

# 异步中间件
class AsyncTimingMiddleware:
    async def __call__(self, request):
        start_time = time.time()

        response = await self.get_response(request)

        duration = time.time() - start_time
        response['X-Response-Time'] = str(duration)

        return response

 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

# models.py
from django.db import models
from asgiref.sync import sync_to_async

class Post(models.Model):
    title = models.CharField(max_length=200)
    content = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)

# 异步查询
async def get_recent_posts(limit=10):
    """异步获取最新文章"""
    posts = await sync_to_async(list)(
        Post.objects.order_by('-created_at')[:limit]
    )
    return posts

# 异步ORM操作
async def create_post_async(title, content):
    """异步创建文章"""
    post = await sync_to_async(Post.objects.create)(
        title=title,
        content=content
    )
    return post

 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

# forms.py
from django import forms

class AdvancedForm(forms.Form):
    # 新增字段类型
    color = forms.ColorField(
        help_text="选择颜色"
    )

    date_range = forms.SplitDateTimeField(
        widget=forms.SplitDateTimeWidget(
            date_attrs={'type': 'date'},
            time_attrs={'type': 'time'}
        )
    )

    decimal_range = forms.DecimalField(
        min_value=0,
        max_value=1000,
        decimal_places=2
    )

# 自定义字段渲染
class CustomModelForm(forms.ModelForm):
    class Meta:
        model = MyModel
        fields = '__all__'
        widgets = {
            'description': forms.Textarea(attrs={'rows': 4}),
            'email': forms.EmailInput(attrs={'autocomplete': 'email'}),
        }

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

# forms.py - 复杂验证逻辑
from django.core.exceptions import ValidationError

class RegistrationForm(forms.Form):
    username = forms.CharField(max_length=150)
    email = forms.EmailField()
    password = forms.CharField(widget=forms.PasswordInput)
    confirm_password = forms.CharField(widget=forms.PasswordInput)

    def clean_username(self):
        username = self.cleaned_data['username']
        if User.objects.filter(username__iexact=username).exists():
            raise ValidationError("用户名已存在")
        return username

    def clean(self):
        cleaned_data = super().clean()
        password = cleaned_data.get('password')
        confirm_password = cleaned_data.get('confirm_password')

        if password and confirm_password and password != confirm_password:
            raise ValidationError("两次输入的密码不一致")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# views.py - 优化的查询
from django.db.models import Prefetch, Q, F
from django.db import transaction

def optimized_product_list(request):
    # 使用select_related和prefetch_related
    products = Product.objects.select_related('category').prefetch_related(
        Prefetch('reviews', queryset=Review.objects.filter(is_approved=True))
    ).filter(
        Q(is_active=True) & Q(stock__gt=0)
    ).annotate(
        discount_price=F('price') * F('discount_rate')
    ).order_by('-created_at')

    return render(request, 'products/list.html', {'products': products})

# 批量操作
@transaction.atomic
def bulk_update_products():
    products = Product.objects.filter(category_id=1)
    for product in products:
        product.price *= 1.1  # 提价10%

    Product.objects.bulk_update(products, ['price'])

 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

# models.py - 使用数据库函数
from django.db.models import Count, Avg, Sum, StdDev, Variance
from django.db.models.functions import (
    Coalesce, Greatest, Least, Length, Lower, Upper,
    Substr, Concat, Now, Extract
)

class ProductSales(models.Model):
    product = models.ForeignKey(Product, on_delete=models.CASCADE)
    quantity = models.IntegerField()
    price = models.DecimalField(max_digits=10, decimal_places=2)
    sale_date = models.DateField()

# 复杂聚合查询
def sales_statistics():
    stats = ProductSales.objects.aggregate(
        total_sales=Sum('quantity'),
        avg_price=Avg('price'),
        price_stddev=StdDev('price'),
        unique_products=Count('product', distinct=True)
    )

    # 时间序列数据
    monthly_sales = ProductSales.objects.annotate(
        month=Extract('sale_date', 'month'),
        year=Extract('sale_date', 'year')
    ).values('year', 'month').annotate(
        total_quantity=Sum('quantity'),
        total_revenue=Sum('quantity' * F('price'))
    ).order_by('year', 'month')

    return stats, monthly_sales

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

<!-- templates/base.html -->
{% load custom_tags %}

<!-- 新的模板上下文处理器 -->
{% block content %}
    <!-- 条件渲染改进 -->
    {% if user.is_authenticated|default:False %}
        <p>欢迎, {{ user.username }}!</p>
    {% endif %}

    <!-- 异步模板渲染 -->
    {% async_template "partials/async_content.html" %}

    <!-- 模板组件 -->
    {% component "card" title="标题" %}
        <p>卡片内容</p>
    {% endcomponent %}
{% endblock %}

 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

# templatetags/cache_tags.py
from django import template
from django.core.cache import cache

register = template.Library()

@register.inclusion_tag('includes/product_card.html')
def product_card(product):
    """产品卡片模板标签"""
    return {'product': product}

@register.simple_tag
def cached_product_count(category_slug):
    """缓存的产品数量"""
    cache_key = f'product_count_{category_slug}'
    count = cache.get(cache_key)

    if count is None:
        count = Product.objects.filter(
            category__slug=category_slug,
            is_active=True
        ).count()
        cache.set(cache_key, count, 300)  # 缓存5分钟

    return count

 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

# api/views.py
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
from asgiref.sync import sync_to_async

@require_http_methods(["GET"])
async def async_api_list(request):
    """异步API列表"""
    try:
        # 异步数据库查询
        items = await sync_to_async(list)(
            Item.objects.all().values('id', 'name', 'created_at')
        )

        # 并发HTTP请求
        async with aiohttp.ClientSession() as session:
            tasks = []
            for item in items:
                task = fetch_additional_data(session, item['id'])
                tasks.append(task)

            additional_data = await asyncio.gather(*tasks)

        # 合并数据
        for item, data in zip(items, additional_data):
            item['additional'] = data

        return JsonResponse({'items': items})

    except Exception as e:
        return JsonResponse(
            {'error': str(e)},
            status=500
        )

 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

# serializers.py - 异步序列化器
from rest_framework import serializers
from .models import Post

class AsyncPostSerializer(serializers.ModelSerializer):
    author_name = serializers.CharField(source='author.username', read_only=True)

    class Meta:
        model = Post
        fields = ['id', 'title', 'content', 'author_name', 'created_at']

# views.py - 异步ViewSet
from rest_framework import viewsets
from rest_framework.decorators import action
from asgiref.sync import sync_to_async

class AsyncPostViewSet(viewsets.ModelViewSet):
    serializer_class = AsyncPostSerializer
    queryset = Post.objects.all()

    async def list(self, request, *args, **kwargs):
        """异步列表"""
        queryset = await sync_to_async(list)(self.get_queryset())
        serializer = self.get_serializer(queryset, many=True)
        return Response(serializer.data)

    @action(detail=False, methods=['GET'])
    async def statistics(self, request):
        """异步统计"""
        stats = await sync_to_async(self.get_statistics)()
        return Response(stats)

 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

# admin.py - 增强的管理后台
from django.contrib import admin
from django.db.models import Count
from django.utils.html import format_html

@admin.register(Product)
class ProductAdmin(admin.ModelAdmin):
    list_display = ['name', 'category', 'price', 'stock_status', 'image_preview']
    list_filter = ['category', 'is_active', 'created_at']
    search_fields = ['name', 'description']
    ordering = ['-created_at']

    # 自定义字段
    def stock_status(self, obj):
        if obj.stock > 50:
            color = 'green'
        elif obj.stock > 10:
            color = 'orange'
        else:
            color = 'red'
        return format_html(
            '<span style="color: {};">{}</span>',
            color,
            obj.stock
        )

    def image_preview(self, obj):
        if obj.image:
            return format_html(
                '<img src="{}" width="50" height="50" />',
                obj.image.url
            )
        return "无图片"

    # 批量操作
    actions = ['make_active', 'make_inactive']

    def make_active(self, request, queryset):
        queryset.update(is_active=True)
    make_active.short_description = "设为激活"

    # 自定义表单
    fieldsets = (
        ('基本信息', {
            'fields': ('name', 'category', 'description')
        }),
        ('价格和库存', {
            'fields': ('price', 'discount_rate', 'stock')
        }),
        ('媒体', {
            'fields': ('image', 'gallery')
        }),
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# settings.py - CSRF配置
CSRF_TRUSTED_ORIGINS = [
    'https://yourdomain.com',
    'https://api.yourdomain.com'
]

# 使用SameSite cookie
CSRF_COOKIE_SAMESITE = 'Lax'
CSRF_COOKIE_SECURE = True
CSRF_COOKIE_HTTPONLY = True

# 自定义CSRF保护
class CustomCSRFMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        if request.path.startswith('/api/'):
            # API请求使用不同的CSRF处理
            pass

        response = self.get_response(request)
        return response

 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

# authentication.py - 自定义认证
from django.contrib.auth.backends import BaseBackend
from django.contrib.auth.models import User

class APIKeyBackend(BaseBackend):
    """API密钥认证"""

    def authenticate(self, request, api_key=None):
        if not api_key:
            return None

        try:
            user = User.objects.get(api_key__key=api_key, is_active=True)
            return user
        except User.DoesNotExist:
            return None

    def get_user(self, user_id):
        try:
            return User.objects.get(pk=user_id, is_active=True)
        except User.DoesNotExist:
            return None

# permissions.py - 权限控制
from rest_framework import permissions

class IsOwnerOrReadOnly(permissions.BasePermission):
    """只有所有者可以编辑"""

    def has_object_permission(self, request, view, obj):
        if request.method in permissions.SAFE_METHODS:
            return True

        return obj.owner == request.user

 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

# middleware.py - 性能监控
import time
import logging
from django.conf import settings

logger = logging.getLogger(__name__)

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

    def __call__(self, request):
        if not settings.DEBUG:
            return self.get_response(request)

        start_time = time.time()

        # 记录查询数
        from django.conf import settings
        from django.db import connection
        initial_queries = len(connection.queries)

        response = self.get_response(request)

        duration = time.time() - start_time
        query_count = len(connection.queries) - initial_queries

        logger.info(
            f"Path: {request.path} | "
            f"Duration: {duration:.3f}s | "
            f"Queries: {query_count}"
        )

        response['X-Debug-Duration'] = str(duration)
        response['X-Debug-Queries'] = str(query_count)

        return response

 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

# views.py - 多级缓存
from django.views.decorators.cache import cache_page, never_cache
from django.core.cache import cache
from django.utils.decorators import method_decorator
from django.views.decorators.vary import vary_on_headers

@cache_page(60 * 15)  # 缓存15分钟
@vary_on_headers('Cookie', 'Accept-Language')
def product_list(request):
    """产品列表页"""
    pass

# 模板片段缓存
{% load cache %}
{% cache 500 sidebar request.user.username %}
    <!-- 侧边栏内容 -->
{% endcache %}

# 对象级缓存
def get_product_with_cache(product_id):
    cache_key = f'product_{product_id}'
    product = cache.get(cache_key)

    if product is None:
        product = Product.objects.get(id=product_id)
        cache.set(cache_key, product, 60 * 30)  # 30分钟

    return product

 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

# tasks.py - 异步任务
from celery import shared_task
from django.core.mail import send_mail
import asyncio

@shared_task
def send_welcome_email(user_id):
    """发送欢迎邮件"""
    from django.contrib.auth import get_user_model
    User = get_user_model()

    user = User.objects.get(id=user_id)
    send_mail(
        '欢迎加入我们',
        f'欢迎 {user.username}！',
        'from@example.com',
        [user.email],
        fail_silently=False,
    )

@shared_task
def process_uploaded_file(file_path):
    """处理上传的文件"""
    # 异步处理大文件
    pass

 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

# consumers.py - WebSocket消费者
import json
from channels.generic.websocket import AsyncWebsocketConsumer

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = f'chat_{self.room_name}'

        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )

        await self.accept()

    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']

        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message
            }
        )

    async def chat_message(self, event):
        message = event['message']

        await self.send(text_data=json.dumps({
            'message': message
        }))

1
2
3
4
5
6
7
8

# 检查兼容性
pip install django==5.0

# 运行测试
python manage.py test

# 检查弃用警告
python -Wd manage.py runserver

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# settings.py - 更新配置
INSTALLED_APPS = [
    # 确保所有应用兼容Django 5.0
]

# 中间件更新
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',
]