django-patterns

Django 开发模式

适用于可扩展、可维护应用的生产级 Django 架构模式。

何时激活

• 构建 Django Web 应用
• 设计 Django REST Framework API
• 使用 Django ORM 和模型
• 设置 Django 项目结构
• 实现缓存、信号、中间件

项目结构

推荐布局

myproject/
├── config/
│   ├── __init__.py
│   ├── settings/
│   │   ├── __init__.py
│   │   ├── base.py          # 基础设置
│   │   ├── development.py   # 开发设置
│   │   ├── production.py    # 生产设置
│   │   └── test.py          # 测试设置
│   ├── urls.py
│   ├── wsgi.py
│   └── asgi.py
├── manage.py
└── apps/
    ├── __init__.py
    ├── users/
    │   ├── __init__.py
    │   ├── models.py
    │   ├── views.py
    │   ├── serializers.py
    │   ├── urls.py
    │   ├── permissions.py
    │   ├── filters.py
    │   ├── services.py
    │   └── tests/
    └── products/
        └── ...

拆分设置模式

# config/settings/base.py
from pathlib import Path

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

SECRET_KEY = env('DJANGO_SECRET_KEY')
DEBUG = False
ALLOWED_HOSTS = []

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'rest_framework',
    'rest_framework.authtoken',
    'corsheaders',
    # 本地应用
    'apps.users',
    'apps.products',
]

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'whitenoise.middleware.WhiteNoiseMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'corsheaders.middleware.CorsMiddleware',
    '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': 'django.db.backends.postgresql',
        'NAME': env('DB_NAME'),
        'USER': env('DB_USER'),
        'PASSWORD': env('DB_PASSWORD'),
        'HOST': env('DB_HOST'),
        'PORT': env('DB_PORT', default='5432'),
    }
}

# config/settings/development.py
from .base import *

DEBUG = True
ALLOWED_HOSTS = ['localhost', '127.0.0.1']

DATABASES['default']['NAME'] = 'myproject_dev'

INSTALLED_APPS += ['debug_toolbar']

MIDDLEWARE += ['debug_toolbar.middleware.DebugToolbarMiddleware']

EMAIL_BACKEND = 'django.core.mail.backends.console.EmailBackend'

# config/settings/production.py
from .base import *

DEBUG = False
ALLOWED_HOSTS = env.list('ALLOWED_HOSTS')
SECURE_SSL_REDIRECT = True
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True
SECURE_HSTS_SECONDS = 31536000
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True

# 日志
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'handlers': {
        'file': {
            'level': 'WARNING',
            'class': 'logging.FileHandler',
            'filename': '/var/log/django/django.log',
        },
    },
    'loggers': {
        'django': {
            'handlers': ['file'],
            'level': 'WARNING',
            'propagate': True,
        },
    },
}

模型设计模式

模型最佳实践

from django.db import models
from django.contrib.auth.models import AbstractUser
from django.core.validators import MinValueValidator, MaxValueValidator

class User(AbstractUser):
    """扩展 AbstractUser 的自定义用户模型。"""
    email = models.EmailField(unique=True)
    phone = models.CharField(max_length=20, blank=True)
    birth_date = models.DateField(null=True, blank=True)

    USERNAME_FIELD = 'email'
    REQUIRED_FIELDS = ['username']

    class Meta:
        db_table = 'users'
        verbose_name = 'user'
        verbose_name_plural = 'users'
        ordering = ['-date_joined']

    def __str__(self):
        return self.email

    def get_full_name(self):
        return f"{self.first_name} {self.last_name}".strip()

class Product(models.Model):
    """具有适当字段配置的产品模型。"""
    name = models.CharField(max_length=200)
    slug = models.SlugField(unique=True, max_length=250)
    description = models.TextField(blank=True)
    price = models.DecimalField(
        max_digits=10,
        decimal_places=2,
        validators=[MinValueValidator(0)]
    )
    stock = models.PositiveIntegerField(default=0)
    is_active = models.BooleanField(default=True)
    category = models.ForeignKey(
        'Category',
        on_delete=models.CASCADE,
        related_name='products'
    )
    tags = models.ManyToManyField('Tag', blank=True, related_name='products')
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        db_table = 'products'
        ordering = ['-created_at']
        indexes = [
            models.Index(fields=['slug']),
            models.Index(fields=['-created_at']),
            models.Index(fields=['category', 'is_active']),
        ]
        constraints = [
            models.CheckConstraint(
                check=models.Q(price__gte=0),
                name='price_non_negative'
            )
        ]

    def __str__(self):
        return self.name

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

QuerySet 最佳实践

from django.db import models

class ProductQuerySet(models.QuerySet):
    """Product 模型的自定义 QuerySet。"""

    def active(self):
        """仅返回活跃产品。"""
        return self.filter(is_active=True)

    def with_category(self):
        """预加载关联的 category 以避免 N+1 查询。"""
        return self.select_related('category')

    def with_tags(self):
        """为多对多关系预加载 tags。"""
        return self.prefetch_related('tags')

    def in_stock(self):
        """返回库存大于 0 的产品。"""
        return self.filter(stock__gt=0)

    def search(self, query):
        """按名称或描述搜索产品。"""
        return self.filter(
            models.Q(name__icontains=query) |
            models.Q(description__icontains=query)
        )

class Product(models.Model):
    # ... 字段 ...

    objects = ProductQuerySet.as_manager()  # 使用自定义 QuerySet

# 用法
Product.objects.active().with_category().in_stock()

管理器方法

class ProductManager(models.Manager):
    """用于复杂查询的自定义管理器。"""

    def get_or_none(self, **kwargs):
        """返回对象或 None，而不是引发 DoesNotExist。"""
        try:
            return self.get(**kwargs)
        except self.model.DoesNotExist:
            return None

    def create_with_tags(self, name, price, tag_names):
        """创建产品并关联标签。"""
        product = self.create(name=name, price=price)
        tags = [Tag.objects.get_or_create(name=name)[0] for name in tag_names]
        product.tags.set(tags)
        return product

    def bulk_update_stock(self, product_ids, quantity):
        """批量更新多个产品的库存。"""
        return self.filter(id__in=product_ids).update(stock=quantity)

# 在模型中
class Product(models.Model):
    # ... 字段 ...
    custom = ProductManager()

Django REST Framework 模式

序列化器模式

from rest_framework import serializers
from django.contrib.auth.password_validation import validate_password
from .models import Product, User

class ProductSerializer(serializers.ModelSerializer):
    """Product 模型的序列化器。"""

    category_name = serializers.CharField(source='category.name', read_only=True)
    average_rating = serializers.FloatField(read_only=True)
    discount_price = serializers.SerializerMethodField()

    class Meta:
        model = Product
        fields = [
            'id', 'name', 'slug', 'description', 'price',
            'discount_price', 'stock', 'category_name',
            'average_rating', 'created_at'
        ]
        read_only_fields = ['id', 'slug', 'created_at']

    def get_discount_price(self, obj):
        """如果适用，计算折扣价格。"""
        if hasattr(obj, 'discount') and obj.discount:
            return obj.price * (1 - obj.discount.percent / 100)
        return obj.price

    def validate_price(self, value):
        """确保价格非负。"""
        if value < 0:
            raise serializers.ValidationError("价格不能为负。")
        return value

class ProductCreateSerializer(serializers.ModelSerializer):
    """用于创建产品的序列化器。"""

    class Meta:
        model = Product
        fields = ['name', 'description', 'price', 'stock', 'category']

    def validate(self, data):
        """多个字段的自定义验证。"""
        if data['price'] > 10000 and data['stock'] > 100:
            raise serializers.ValidationError(
                "高价值产品不能有大库存。"
            )
        return data

class UserRegistrationSerializer(serializers.ModelSerializer):
    """用于用户注册的序列化器。"""

    password = serializers.CharField(
        write_only=True,
        required=True,
        validators=[validate_password],
        style={'input_type': 'password'}
    )
    password_confirm = serializers.CharField(write_only=True, style={'input_type': 'password'})

    class Meta:
        model = User
        fields = ['email', 'username', 'password', 'password_confirm']

    def validate(self, data):
        """验证密码匹配。"""
        if data['password'] != data['password_confirm']:
            raise serializers.ValidationError({
                "password_confirm": "密码字段不匹配。"
            })
        return data

    def create(self, validated_data):
        """使用哈希密码创建用户。"""
        validated_data.pop('password_confirm')
        password = validated_data.pop('password')
        user = User.objects.create(**validated_data)
        user.set_password(password)
        user.save()
        return user

ViewSet 模式

from rest_framework import viewsets, status, filters
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated, IsAdminUser
from django_filters.rest_framework import DjangoFilterBackend
from .models import Product
from .serializers import ProductSerializer, ProductCreateSerializer
from .permissions import IsOwnerOrReadOnly
from .filters import ProductFilter
from .services import ProductService

class ProductViewSet(viewsets.ModelViewSet):
    """Product 模型的 ViewSet。"""

    queryset = Product.objects.select_related('category').prefetch_related('tags')
    permission_classes = [IsAuthenticated, IsOwnerOrReadOnly]
    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
    filterset_class = ProductFilter
    search_fields = ['name', 'description']
    ordering_fields = ['price', 'created_at', 'name']
    ordering = ['-created_at']

    def get_serializer_class(self):
        """根据操作返回适当的序列化器。"""
        if self.action == 'create':
            return ProductCreateSerializer
        return ProductSerializer

    def perform_create(self, serializer):
        """使用用户上下文保存。"""
        serializer.save(created_by=self.request.user)

    @action(detail=False, methods=['get'])
    def featured(self, request):
        """返回精选产品。"""
        featured = self.queryset.filter(is_featured=True)[:10]
        serializer = self.get_serializer(featured, many=True)
        return Response(serializer.data)

    @action(detail=True, methods=['post'])
    def purchase(self, request, pk=None):
        """购买产品。"""
        product = self.get_object()
        service = ProductService()
        result = service.purchase(product, request.user)
        return Response(result, status=status.HTTP_201_CREATED)

    @action(detail=False, methods=['get'], permission_classes=[IsAuthenticated])
    def my_products(self, request):
        """返回当前用户创建的产品。"""
        products = self.queryset.filter(created_by=request.user)
        page = self.paginate_queryset(products)
        serializer = self.get_serializer(page, many=True)
        return self.get_paginated_response(serializer.data)

自定义操作

from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response

@api_view(['POST'])
@permission_classes([IsAuthenticated])
def add_to_cart(request):
    """将产品添加到用户购物车。"""
    product_id = request.data.get('product_id')
    quantity = request.data.get('quantity', 1)

    try:
        product = Product.objects.get(id=product_id)
    except Product.DoesNotExist:
        return Response(
            {'error': '产品未找到'},
            status=status.HTTP_404_NOT_FOUND
        )

    cart, _ = Cart.objects.get_or_create(user=request.user)
    CartItem.objects.create(
        cart=cart,
        product=product,
        quantity=quantity
    )

    return Response({'message': '已添加到购物车'}, status=status.HTTP_201_CREATED)

服务层模式

# apps/orders/services.py
from typing import Optional
from django.db import transaction
from .models import Order, OrderItem

class OrderService:
    """订单相关业务逻辑的服务层。"""

    @staticmethod
    @transaction.atomic
    def create_order(user, items):
        """创建订单及明细。"""
        order = Order.objects.create(user=user, status='pending')
        total_price = 0

        for item in items:
            product = item['product']
            quantity = item['quantity']
            if product.stock < quantity:
                raise ValueError(f"{product.name} 库存不足。")  # 添加错误处理示例
            OrderItem.objects.create(
                order=order,
                product=product,
                quantity=quantity,
                price=product.price
            )
            total_price += product.price * quantity
            product.stock -= quantity
            product.save()

        order.total_price = total_price
        order.save()
        return order

缓存模式

Redis 集成

# 使用 django-redis
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': env('REDIS_URL', default='redis://localhost:6379/0'),
        'OPTIONS': {
            'CLIENT_CLASS': 'django_redis.client.DefaultClient',
            'SOCKET_CONNECT_TIMEOUT': 5,
            'SOCKET_TIMEOUT': 5,
        },
        'KEY_PREFIX': 'myproject'
    }
}

# 缓存示例用法
from django.core.cache import cache

class ProductService:
    @staticmethod
    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.select_related('category').get(id=product_id)
            cache.set(cache_key, product, 3600)  # 缓存 1 小时
        return product

信号模式

from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from django.core.cache import cache
from .models import Product

@receiver(post_save, sender=Product)
def clear_product_cache(sender, instance, **kwargs):
    """产品保存后清除缓存。"""
    cache_key = f'product_{instance.id}'
    cache.delete(cache_key)

@receiver(post_delete, sender=Product)
def update_stock_on_delete(sender, instance, **kwargs):
    """产品删除后更新库存。"""
    # 示例：记录删除操作
    print(f"产品 {instance.name} 已删除。")

中间件模式

# apps/core/middleware.py
from django.utils import timezone
from django.core.cache import cache

class RateLimitMiddleware:
    """IP 的速率限制中间件。"""

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

    def __call__(self, request):
        ip = request.META.get('REMOTE_ADDR')
        cache_key = f'rate_limit_{ip}'
        count = cache.get(cache_key, 0)

        if count > 100:  # 每 IP 限制 100 个请求
            from django.http import HttpResponse
            return HttpResponse("速率限制超出", status=429)

        cache.set(cache_key, count + 1, 60)
        response = self.get_response(request)
        return response

class RequestLoggingMiddleware:
    """用于调试的请求日志中间件。"""

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

    def __call__(self, request):
        start_time = timezone.now()
        response = self.get_response(request)
        duration = timezone.now() - start_time
        print(f"{request.method} {request.path} - {duration.total_seconds():.2f}s")
        return response

测试模式

from django.test import TestCase, Client
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from .models import Product, Category
from .serializers import ProductSerializer

class ProductModelTest(TestCase):
    """Product 模型的测试。"""

    def setUp(self):
        self.category = Category.objects.create(name="电子设备")
        self.product = Product.objects.create(
            name="智能手机",
            price=699.99,
            category=self.category
        )

    def test_product_creation(self):
        """测试产品创建。"""
        self.assertEqual(self.product.name, "智能手机")
        self.assertTrue(self.product.slug)
        self.assertEqual(self.product.price, 699.99)

    def test_product_str_method(self):
        """测试字符串表示。"""
        self.assertEqual(str(self.product), "智能手机")

class ProductAPITest(APITestCase):
    """Product API 的测试。"""

    def setUp(self):
        self.client = Client()
        self.category = Category.objects.create(name="电子设备")
        self.product_data = {
            "name": "笔记本电脑",
            "price": 999.99,
            "category": self.category.id
        }

    def test_create_product(self):
        """测试创建产品 API。"""
        response = self.client.post(
            reverse('product-list'),
            self.product_data,
            format='json'
        )
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        self.assertEqual(response.data['name'], "笔记本电脑")

    def test_list_products(self):
        """测试列出产品 API。"""
        Product.objects.create(name="平板", price=499.99, category=self.category)
        response = self.client.get(reverse('product-list'))
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data['results']), 1)