Django 開発パターン
スケーラブルで保守可能なアプリケーションのための本番グレードのDjangoアーキテクチャパターン。
いつ有効化するか
- Djangoウェブアプリケーションを構築するとき
- 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
# ロギング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, }, },}モデル設計パターン
モデルのベストプラクティス
python
from django.db import modelsfrom django.contrib.auth.models import AbstractUserfrom 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): """適切なフィールド設定を持つProductモデル。""" 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): """Productモデルのカスタム 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): """在庫が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()マネージャーメソッド
python
class ProductManager(models.Manager): """複雑なクエリ用のカスタムマネージャー。"""
def get_or_none(self, **kwargs): """DoesNotExistの代わりにオブジェクトまたはNoneを返す。""" 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パターン
シリアライザーパターン
python
from rest_framework import serializersfrom django.contrib.auth.password_validation import validate_passwordfrom .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("Price cannot be negative.") 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( "Cannot have high-value products with large stock." ) 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": "Password fields didn't match." }) 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 userViewSetパターン
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): """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)カスタムアクション
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): """製品をユーザーのカートに追加。""" 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: """注文関連のビジネスロジック用のサービスレイヤー。"""
@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
@staticmethod def process_payment(order: Order, payment_data: dict) -> bool: """注文の支払いを処理。""" # 決済ゲートウェイとの統合 payment = PaymentGateway.charge( amount=order.total_price, token=payment_data['token'] )
if payment.success: order.status = Order.Status.PAID order.save() # 確認メールを送信 OrderService.send_confirmation_email(order) return True
return False
@staticmethod def send_confirmation_email(order: Order): """注文確認メールを送信。""" # メール送信ロジック 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分class ProductListView(generic.ListView): model = Product template_name = 'products/list.html' context_object_name = 'products'テンプレートフラグメントのキャッシング
django
{% load cache %}{% cache 500 sidebar %} ... 高コストなサイドバーコンテンツ ...{% endcache %}低レベルキャッシング
python
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) # 15分
return productsQuerySetのキャッシング
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時間
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): """ユーザーが作成されたときにプロファイルを作成。""" if created: Profile.objects.create(user=instance)
@receiver(post_save, sender=User)def save_user_profile(sender, instance, **kwargs): """ユーザーが保存されたときにプロファイルを保存。""" 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 apps.users.signalsミドルウェア
カスタムミドルウェア
python
# middleware/active_user_middleware.pyimport timefrom django.utils.deprecation import MiddlewareMixin
class ActiveUserMiddleware(MiddlewareMixin): """アクティブユーザーを追跡するミドルウェア。"""
def process_request(self, request): """受信リクエストを処理。""" if request.user.is_authenticated: # 最終アクティブ時刻を更新 request.user.last_active = timezone.now() request.user.save(update_fields=['last_active'])
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クエリの防止
python
# Bad - N+1クエリproducts = Product.objects.all()for product in products: print(product.category.name) # 各製品に対して個別のクエリ
# Good - select_relatedで単一クエリproducts = Product.objects.select_related('category').all()for product in products: print(product.category.name)
# Good - 多対多のためのprefetchproducts = 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
# 一括作成Product.objects.bulk_create([ Product(name=f'Product {i}', price=10.00) for i in range(1000)])
# 一括更新products = Product.objects.all()[:100]for product in products: product.is_active = TrueProduct.objects.bulk_update(products, ['is_active'])
# 一括削除Product.objects.filter(stock=0).delete()クイックリファレンス
| パターン | 説明 |
|---|---|
| 分割設定 | dev/prod/test設定の分離 |
| カスタムQuerySet | 再利用可能なクエリメソッド |
| サービスレイヤー | ビジネスロジックの分離 |
| ViewSet | REST APIエンドポイント |
| シリアライザー検証 | リクエスト/レスポンス変換 |
| select_related | 外部キー最適化 |
| prefetch_related | 多対多最適化 |
| キャッシュファースト | 高コスト操作のキャッシング |
| シグナル | イベント駆動アクション |
| ミドルウェア | リクエスト/レスポンス処理 |
覚えておいてください: Djangoは多くのショートカットを提供しますが、本番アプリケーションでは、構造と組織が簡潔なコードよりも重要です。保守性を重視して構築してください。


