Django 模式

中级 Intermediate 参考型 Reference ⚡ Claude Code 专属 ⚡ Claude Code Optimized
6 min read · 322 lines

Django 开发模式:ORM、DRF API、信号、中间件与缓存策略

Django 模式

原始文件:skills/django-patterns/SKILL.md

项目结构

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

模型设计模式

模型最佳实践

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'
    )
    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'
            )
        ]

自定义 QuerySet

class ProductQuerySet(models.QuerySet):
    def active(self):
        """只返回活跃产品。"""
        return self.filter(is_active=True)

    def with_category(self):
        """关联查询分类,避免 N+1 查询。"""
        return self.select_related('category')

    def with_tags(self):
        """预获取多对多关系的标签。"""
        return self.prefetch_related('tags')

    def in_stock(self):
        return self.filter(stock__gt=0)

    def search(self, query):
        return self.filter(
            models.Q(name__icontains=query) |
            models.Q(description__icontains=query)
        )

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

Django REST Framework 模式

序列化器(Serializer)

class ProductSerializer(serializers.ModelSerializer):
    category_name = serializers.CharField(source='category.name', read_only=True)
    discount_price = serializers.SerializerMethodField()

    class Meta:
        model = Product
        fields = ['id', 'name', 'slug', 'description', 'price',
                  'discount_price', 'stock', 'category_name', '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

ViewSet 模式

class ProductViewSet(viewsets.ModelViewSet):
    queryset = Product.objects.select_related('category').prefetch_related('tags')
    permission_classes = [IsAuthenticated, IsOwnerOrReadOnly]
    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
    search_fields = ['name', 'description']
    ordering_fields = ['price', 'created_at', 'name']

    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)

服务层模式(Service Layer)

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

    @staticmethod
    @transaction.atomic
    def create_order(user, cart: Cart) -> Order:
        """从购物车创建订单。"""
        order = Order.objects.create(
            user=user, total_price=cart.total_price
        )

        for item in cart.items.all():
            OrderItem.objects.create(
                order=order, product=item.product,
                quantity=item.quantity, price=item.product.price
            )

        cart.items.all().delete()  # 清空购物车
        return order

缓存策略

视图级缓存

@method_decorator(cache_page(60 * 15), name='dispatch')  # 15 分钟
class ProductListView(generic.ListView):
    model = Product

底层缓存

from django.core.cache import cache

def get_featured_products():
    """获取带缓存的推荐产品。"""
    cache_key = 'featured_products'
    products = cache.get(cache_key)

    if products is None:
        products = list(Product.objects.filter(is_featured=True))
        cache.set(cache_key, products, timeout=60 * 15)

    return products

信号(Signals)

from django.db.models.signals import post_save
from django.dispatch import receiver

@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    """用户创建时创建配置文件。"""
    if created:
        Profile.objects.create(user=instance)

中间件(Middleware)

class RequestLoggingMiddleware(MiddlewareMixin):
    def process_request(self, request):
        request.start_time = time.time()

    def process_response(self, request, response):
        if hasattr(request, 'start_time'):
            duration = time.time() - request.start_time
            logger.info(f'{request.method} {request.path} - {response.status_code} - {duration:.3f}s')
        return response

性能优化

N+1 查询防范

# 错误 -- N+1 查询
products = Product.objects.all()
for product in products:
    print(product.category.name)  # 每个产品一条单独查询

# 正确 -- 使用 select_related 单次查询
products = Product.objects.select_related('category').all()

# 正确 -- 多对多使用 prefetch_related
products = Product.objects.prefetch_related('tags').all()

批量操作

# 批量创建
Product.objects.bulk_create([
    Product(name=f'产品 {i}', price=10.00)
    for i in range(1000)
])

# 批量更新
Product.objects.bulk_update(products, ['is_active'])

分离设置模式(Split Settings)

# 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

速查表

模式 描述
分离设置 分开 dev/prod/test 设置
自定义 QuerySet 可复用的查询方法
服务层 业务逻辑分离
ViewSet REST API 端点
select_related 外键优化
prefetch_related 多对多优化
缓存优先 缓存昂贵操作
信号 事件驱动操作
中间件 请求/响应处理

记住:Django 提供了很多快捷方式,但对于生产应用,结构和组织比简洁代码更重要。为可维护性而构建。

相关技能 Related Skills