from django.db import models
classProductQuerySet(models.QuerySet):
"""Product 모델을 위한 커스텀 QuerySet."""defactive(self):
"""활성화된 제품만 반환."""returnself.filter(is_active=True)
defwith_category(self):
"""N+1 쿼리 방지를 위해 관련 카테고리 선택(select_related)."""returnself.select_related('category')
defwith_tags(self):
"""다대다 관계를 위한 태그 미리 가져오기(prefetch_related)."""returnself.prefetch_related('tags')
defin_stock(self):
"""재고가 0보다 큰 제품만 반환."""returnself.filter(stock__gt=0)
defsearch(self, query):
"""이름 또는 설명으로 제품 검색."""returnself.filter(
models.Q(name__icontains=query) |
models.Q(description__icontains=query)
)
classProduct(models.Model):
# ... 필드 정의 ...
objects = ProductQuerySet.as_manager() # 커스텀 QuerySet 사용# 사용 예시
Product.objects.active().with_category().in_stock()
매니저(Manager) 메서드
classProductManager(models.Manager):
"""복잡한 쿼리를 위한 커스텀 매니저."""defget_or_none(self, **kwargs):
"""DoesNotExist 대신 객체 또는 None을 반환."""try:
returnself.get(**kwargs)
exceptself.model.DoesNotExist:
returnNonedefcreate_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
defbulk_update_stock(self, product_ids, quantity):
"""여러 제품의 재고를 한꺼번에 업데이트."""returnself.filter(id__in=product_ids).update(stock=quantity)
# 모델에서 설정classProduct(models.Model):
# ... 필드 정의 ...
custom = ProductManager()
Django REST Framework 패턴
시리얼라이저(Serializer) 패턴
from rest_framework import serializers
from django.contrib.auth.password_validation import validate_password
from .models import Product, User
classProductSerializer(serializers.ModelSerializer):
"""Product 모델을 위한 시리얼라이저."""
category_name = serializers.CharField(source='category.name', read_only=True)
average_rating = serializers.FloatField(read_only=True)
discount_price = serializers.SerializerMethodField()
classMeta:
model = Product
fields = [
'id', 'name', 'slug', 'description', 'price',
'discount_price', 'stock', 'category_name',
'average_rating', 'created_at'
]
read_only_fields = ['id', 'slug', 'created_at']
defget_discount_price(self, obj):
"""할인 가격 계산(해당하는 경우)."""ifhasattr(obj, 'discount') and obj.discount:
return obj.price * (1 - obj.discount.percent / 100)
return obj.price
defvalidate_price(self, value):
"""가격이 음수가 아닌지 확인."""if value < 0:
raise serializers.ValidationError("가격은 음수일 수 없습니다.")
return value
classProductCreateSerializer(serializers.ModelSerializer):
"""제품 생성을 위한 시리얼라이저."""classMeta:
model = Product
fields = ['name', 'description', 'price', 'stock', 'category']
defvalidate(self, data):
"""여러 필드에 대한 커스텀 유효성 검사."""if data['price'] > 10000and data['stock'] > 100:
raise serializers.ValidationError(
"고가 제품은 대량 재고를 보유할 수 없습니다."
)
return data
classUserRegistrationSerializer(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'})
classMeta:
model = User
fields = ['email', 'username', 'password', 'password_confirm']
defvalidate(self, data):
"""비밀번호 일치 여부 확인."""if data['password'] != data['password_confirm']:
raise serializers.ValidationError({
"password_confirm": "비밀번호가 일치하지 않습니다."
})
return data
defcreate(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
classProductViewSet(viewsets.ModelViewSet):
"""Product 모델을 위한 뷰셋."""
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']
defget_serializer_class(self):
"""액션에 따라 적절한 시리얼라이저 반환."""ifself.action == 'create':
return ProductCreateSerializer
return ProductSerializer
defperform_create(self, serializer):
"""사용자 컨텍스트와 함께 저장."""
serializer.save(created_by=self.request.user)
@action(detail=False, methods=['get'])deffeatured(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'])defpurchase(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])defmy_products(self, request):
"""현재 사용자가 생성한 제품 반환."""
products = self.queryset.filter(created_by=request.user)
page = self.paginate_queryset(products)
serializer = self.get_serializer(page, many=True)
returnself.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])defadd_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)
서비스 레이어 패턴 (Service Layer Pattern)
# apps/orders/services.pyfrom typing importOptionalfrom django.db import transaction
from .models import Order, OrderItem
classOrderService:
"""주문 관련 비즈니스 로직을 위한 서비스 레이어.""" @staticmethod @transaction.atomicdefcreate_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
@staticmethoddefprocess_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)
returnTruereturnFalse @staticmethoddefsend_confirmation_email(order: Order):
"""주문 확인 이메일 발송."""# 이메일 발송 로직pass
캐싱 전략
뷰 레벨 캐싱 (View-Level Caching)
from django.views.decorators.cache import cache_page
from django.utils.decorators import method_decorator
@method_decorator(cache_page(60 * 15), name='dispatch') # 15분classProductListView(generic.ListView):
model = Product
template_name = 'products/list.html'
context_object_name = 'products'
템플릿 프래그먼트 캐싱 (Template Fragment Caching)
{% load cache %}
{% cache 500 sidebar %}
... 비용이 많이 드는 사이드바 콘텐츠 ...
{% endcache %}
저수준 캐싱 (Low-Level Caching)
from django.core.cache import cache
defget_featured_products():
"""추천 제품 정보를 캐싱과 함께 가져오기."""
cache_key = 'featured_products'
products = cache.get(cache_key)
if products isNone:
products = list(Product.objects.filter(is_featured=True))
cache.set(cache_key, products, timeout=60 * 15) # 15분return products
# apps/users/signals.pyfrom django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth import get_user_model
from .models import Profile
User = get_user_model()
@receiver(post_save, sender=User)defcreate_user_profile(sender, instance, created, **kwargs):
"""사용자 생성 시 프로필 생성."""if created:
Profile.objects.create(user=instance)
@receiver(post_save, sender=User)defsave_user_profile(sender, instance, **kwargs):
"""사용자 저장 시 프로필 저장."""
instance.profile.save()
# apps/users/apps.pyfrom django.apps import AppConfig
classUsersConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'apps.users'defready(self):
"""앱이 준비되었을 때 시그널 임포트."""import apps.users.signals
미들웨어(Middleware)
커스텀 미들웨어
# middleware/active_user_middleware.pyimport time
from django.utils.deprecation import MiddlewareMixin
classActiveUserMiddleware(MiddlewareMixin):
"""활성 사용자를 추적하기 위한 미들웨어."""defprocess_request(self, request):
"""들어오는 요청 처리."""if request.user.is_authenticated:
# 마지막 활성 시간 업데이트
request.user.last_active = timezone.now()
request.user.save(update_fields=['last_active'])
classRequestLoggingMiddleware(MiddlewareMixin):
"""요청 로깅을 위한 미들웨어."""defprocess_request(self, request):
"""요청 시작 시간 기록."""
request.start_time = time.time()
defprocess_response(self, request, response):
"""요청 소요 시간 기록."""ifhasattr(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()
for product in products:
print(product.category.name)
# 좋음 - 다대다 관계를 위한 prefetch_related
products = Product.objects.prefetch_related('tags').all()
for product in products:
for tag in product.tags.all():
print(tag.name)
# 대량 생성 (Bulk create)
Product.objects.bulk_create([
Product(name=f'Product {i}', price=10.00)
for i inrange(1000)
])
# 대량 업데이트 (Bulk update)
products = Product.objects.all()[:100]
for product in products:
product.is_active = True
Product.objects.bulk_update(products, ['is_active'])
# 대량 삭제 (Bulk delete)
Product.objects.filter(stock=0).delete()
빠른 참조
패턴
설명
설정 분리
개발/프로덕션/테스트 환경별 설정 관리
커스텀 QuerySet
재사용 가능한 쿼리 메서드 정의
서비스 레이어
비즈니스 로직의 분리 및 캡슐화
ViewSet
REST API 엔드포인트 구현
시리얼라이저 검증
요청/응답 데이터 변환 및 유효성 검사
select_related
외래 키 참조 최적화
prefetch_related
다대다 관계 참조 최적화
캐시 우선(Cache first)
비용이 많이 드는 작업의 결과 캐싱
시그널(Signals)
이벤트 중심의 액션 처리
미들웨어(Middleware)
요청/응답 전처리 및 후처리
기억하세요: Django는 많은 지름길을 제공하지만, 프로덕션 애플리케이션에서는 간결한 코드보다 구조와 조직화가 더 중요합니다. 유지 관리를 고려하여 설계하세요.