Django Patterns

affaan-m/ECC/docs/zh-CN/skills/django-patterns

作者 affaan-mef648e01899ba3e8dc6371642deaaf64b4477775无许可证275K 个星标收录于 2026年10月9日更新于 2026年10月9日仓库4天前更新

Django架构模式,使用DRF设计REST API,ORM最佳实践,缓存,信号,中间件,以及生产级Django应用程序。

AI 生成的概览

生产级 Django 架构、DRF API、ORM、缓存、信号与中间件的参考模式。

功能
该技能提供一套面向可扩展、可维护应用的生产级 Django 架构模式参考。内容涵盖项目结构与拆分设置、模型与 QuerySet 设计、Django REST Framework 序列化器与 ViewSet、服务层、缓存策略、信号、自定义中间件,以及 select_related、prefetch_related、索引和批量操作等性能优化方法。它仅为指导性内容,不生成文件或脚本。
适用场景
适用于构建 Django Web 应用、设计 Django REST Framework API、使用 Django ORM 与模型、搭建 Django 项目结构,或实现缓存、信号与中间件时。
运行要求
无需任何工具、软件包或凭据;仅为说明性内容,不附带脚本。文中模式涉及 Django、Django REST Framework 及相关软件包,但阅读这些指导无需安装任何内容。

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/        └── ...

拆分设置模式

python
# config/settings/base.pyfrom pathlib import Path
BASE_DIR = Path(__file__).resolve().parent.parent.parent
SECRET_KEY = env('DJANGO_SECRET_KEY')DEBUG = FalseALLOWED_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',    # Local apps    '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.pyfrom .base import *
DEBUG = TrueALLOWED_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.pyfrom .base import *
DEBUG = FalseALLOWED_HOSTS = env.list('ALLOWED_HOSTS')SECURE_SSL_REDIRECT = TrueSESSION_COOKIE_SECURE = TrueCSRF_COOKIE_SECURE = TrueSECURE_HSTS_SECONDS = 31536000SECURE_HSTS_INCLUDE_SUBDOMAINS = TrueSECURE_HSTS_PRELOAD = True
# LoggingLOGGING = {    '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,        },    },}

模型设计模式

模型最佳实践

python
from django.db import modelsfrom django.contrib.auth.models import AbstractUserfrom django.core.validators import MinValueValidator, MaxValueValidator
class User(AbstractUser):    """Custom user model extending 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):    """Product model with proper field configuration."""    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 最佳实践

python
from django.db import models
class ProductQuerySet(models.QuerySet):    """Custom QuerySet for Product model."""
    def active(self):        """Return only active products."""        return self.filter(is_active=True)
    def with_category(self):        """Select related category to avoid N+1 queries."""        return self.select_related('category')
    def with_tags(self):        """Prefetch tags for many-to-many relationship."""        return self.prefetch_related('tags')
    def in_stock(self):        """Return products with stock > 0."""        return self.filter(stock__gt=0)
    def search(self, query):        """Search products by name or description."""        return self.filter(            models.Q(name__icontains=query) |            models.Q(description__icontains=query)        )
class Product(models.Model):    # ... fields ...
    objects = ProductQuerySet.as_manager()  # Use custom QuerySet
# UsageProduct.objects.active().with_category().in_stock()

管理器方法

python
class ProductManager(models.Manager):    """Custom manager for complex queries."""
    def get_or_none(self, **kwargs):        """Return object or None instead of DoesNotExist."""        try:            return self.get(**kwargs)        except self.model.DoesNotExist:            return None
    def create_with_tags(self, name, price, tag_names):        """Create product with associated tags."""        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):        """Bulk update stock for multiple products."""        return self.filter(id__in=product_ids).update(stock=quantity)
# In modelclass Product(models.Model):    # ... fields ...    custom = ProductManager()

Django REST Framework 模式

序列化器模式

python
from rest_framework import serializersfrom django.contrib.auth.password_validation import validate_passwordfrom .models import Product, User
class ProductSerializer(serializers.ModelSerializer):    """Serializer for Product model."""
    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):        """Calculate discount price if applicable."""        if hasattr(obj, 'discount') and obj.discount:            return obj.price * (1 - obj.discount.percent / 100)        return obj.price
    def validate_price(self, value):        """Ensure price is non-negative."""        if value < 0:            raise serializers.ValidationError("Price cannot be negative.")        return value
class ProductCreateSerializer(serializers.ModelSerializer):    """Serializer for creating products."""
    class Meta:        model = Product        fields = ['name', 'description', 'price', 'stock', 'category']
    def validate(self, data):        """Custom validation for multiple fields."""        if data['price'] > 10000 and data['stock'] > 100:            raise serializers.ValidationError(                "Cannot have high-value products with large stock."            )        return data
class UserRegistrationSerializer(serializers.ModelSerializer):    """Serializer for user registration."""
    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):        """Validate passwords match."""        if data['password'] != data['password_confirm']:            raise serializers.ValidationError({                "password_confirm": "Password fields didn't match."            })        return data
    def create(self, validated_data):        """Create user with hashed password."""        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 模式

python
from rest_framework import viewsets, status, filtersfrom rest_framework.decorators import actionfrom rest_framework.response import Responsefrom rest_framework.permissions import IsAuthenticated, IsAdminUserfrom django_filters.rest_framework import DjangoFilterBackendfrom .models import Productfrom .serializers import ProductSerializer, ProductCreateSerializerfrom .permissions import IsOwnerOrReadOnlyfrom .filters import ProductFilterfrom .services import ProductService
class ProductViewSet(viewsets.ModelViewSet):    """ViewSet for Product model."""
    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):        """Return appropriate serializer based on action."""        if self.action == 'create':            return ProductCreateSerializer        return ProductSerializer
    def perform_create(self, serializer):        """Save with user context."""        serializer.save(created_by=self.request.user)
    @action(detail=False, methods=['get'])    def featured(self, request):        """Return featured products."""        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):        """Purchase a product."""        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):        """Return products created by current user."""        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)

自定义操作

python
from rest_framework.decorators import api_view, permission_classesfrom rest_framework.permissions import IsAuthenticatedfrom rest_framework.response import Response
@api_view(['POST'])@permission_classes([IsAuthenticated])def add_to_cart(request):    """Add product to user cart."""    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': 'Product not found'},            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': 'Added to cart'}, status=status.HTTP_201_CREATED)

服务层模式

python
# apps/orders/services.pyfrom typing import Optionalfrom django.db import transactionfrom .models import Order, OrderItem
class OrderService:    """Service layer for order-related business logic."""
    @staticmethod    @transaction.atomic    def create_order(user, cart: Cart) -> Order:        """Create order from cart."""        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            )
        # Clear cart        cart.items.all().delete()
        return order
    @staticmethod    def process_payment(order: Order, payment_data: dict) -> bool:        """Process payment for order."""        # Integration with payment gateway        payment = PaymentGateway.charge(            amount=order.total_price,            token=payment_data['token']        )
        if payment.success:            order.status = Order.Status.PAID            order.save()            # Send confirmation email            OrderService.send_confirmation_email(order)            return True
        return False
    @staticmethod    def send_confirmation_email(order: Order):        """Send order confirmation email."""        # Email sending logic        pass

缓存策略

视图级缓存

python
from django.views.decorators.cache import cache_pagefrom django.utils.decorators import method_decorator
@method_decorator(cache_page(60 * 15), name='dispatch')  # 15 minutesclass ProductListView(generic.ListView):    model = Product    template_name = 'products/list.html'    context_object_name = 'products'

模板片段缓存

django
{% load cache %}{% cache 500 sidebar %}    ... expensive sidebar content ...{% endcache %}

低级缓存

python
from django.core.cache import cache
def get_featured_products():    """Get featured products with caching."""    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)  # 15 minutes
    return products

QuerySet 缓存

python
from django.core.cache import cache
def get_popular_categories():    cache_key = 'popular_categories'    categories = cache.get(cache_key)
    if categories is None:        categories = list(Category.objects.annotate(            product_count=Count('products')        ).filter(product_count__gt=10).order_by('-product_count')[:20])        cache.set(cache_key, categories, timeout=60 * 60)  # 1 hour
    return categories

信号

信号模式

python
# apps/users/signals.pyfrom django.db.models.signals import post_savefrom django.dispatch import receiverfrom django.contrib.auth import get_user_modelfrom .models import Profile
User = get_user_model()
@receiver(post_save, sender=User)def create_user_profile(sender, instance, created, **kwargs):    """Create profile when user is created."""    if created:        Profile.objects.create(user=instance)
@receiver(post_save, sender=User)def save_user_profile(sender, instance, **kwargs):    """Save profile when user is saved."""    instance.profile.save()
# apps/users/apps.pyfrom django.apps import AppConfig
class UsersConfig(AppConfig):    default_auto_field = 'django.db.models.BigAutoField'    name = 'apps.users'
    def ready(self):        """Import signals when app is ready."""        import apps.users.signals

中间件

自定义中间件

python
# middleware/active_user_middleware.pyimport timefrom django.utils.deprecation import MiddlewareMixin
class ActiveUserMiddleware(MiddlewareMixin):    """Middleware to track active users."""
    def process_request(self, request):        """Process incoming request."""        if request.user.is_authenticated:            # Update last active time            request.user.last_active = timezone.now()            request.user.save(update_fields=['last_active'])
class RequestLoggingMiddleware(MiddlewareMixin):    """Middleware for logging requests."""
    def process_request(self, request):        """Log request start time."""        request.start_time = time.time()
    def process_response(self, request, response):        """Log request duration."""        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 查询预防

python
# Bad - N+1 queriesproducts = Product.objects.all()for product in products:    print(product.category.name)  # Separate query for each product
# Good - Single query with select_relatedproducts = Product.objects.select_related('category').all()for product in products:    print(product.category.name)
# Good - Prefetch for many-to-manyproducts = Product.objects.prefetch_related('tags').all()for product in products:    for tag in product.tags.all():        print(tag.name)

数据库索引

python
class Product(models.Model):    name = models.CharField(max_length=200, db_index=True)    slug = models.SlugField(unique=True)    category = models.ForeignKey('Category', on_delete=models.CASCADE)    created_at = models.DateTimeField(auto_now_add=True)
    class Meta:        indexes = [            models.Index(fields=['name']),            models.Index(fields=['-created_at']),            models.Index(fields=['category', 'created_at']),        ]

批量操作

python
# Bulk createProduct.objects.bulk_create([    Product(name=f'Product {i}', price=10.00)    for i in range(1000)])
# Bulk updateproducts = Product.objects.all()[:100]for product in products:    product.is_active = TrueProduct.objects.bulk_update(products, ['is_active'])
# Bulk deleteProduct.objects.filter(stock=0).delete()

快速参考

模式描述
拆分设置分离开发/生产/测试设置
自定义 QuerySet可重用的查询方法
服务层业务逻辑分离
ViewSetREST API 端点
序列化器验证请求/响应转换
select_related外键优化
prefetch_related多对多优化
缓存优先缓存昂贵操作
信号事件驱动操作
中间件请求/响应处理

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

来源与署名

来源:affaan-m/ECC位于docs/zh-CN/skills/django-patterns提交ef648e0

许可证: 无许可证

内容归原作者所有。SourceWeft 从公开仓库中收录这些内容。

举报或申请下架