Django Deep Dive

  myproject/
├── config/
│   ├── settings/
│   │   ├── base.py
│   │   ├── development.py
│   │   └── production.py
│   ├── urls.py
│   └── wsgi.py
├── apps/
│   ├── users/
│   │   ├── models.py
│   │   ├── views.py
│   │   ├── serializers.py
│   │   ├── services.py
│   │   └── tests/
│   └── orders/
└── manage.py
  

  # apps/users/models.py
from django.db import models

class ActiveUserManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().filter(is_active=True)

class UserQuerySet(models.QuerySet):
    def with_orders(self):
        return self.prefetch_related('orders')

    def admins(self):
        return self.filter(role='admin')

class User(models.Model):
    email = models.EmailField(unique=True)
    name = models.CharField(max_length=100)
    role = models.CharField(max_length=20, default='user')
    is_active = models.BooleanField(default=True)

    objects = models.Manager()
    active = ActiveUserManager()
    queryset = UserQuerySet.as_manager()

# Usage
User.active.admins().with_orders()
  

  # ❌ N+1 — one query per order's user
orders = Order.objects.all()
for order in orders:
    print(order.user.email)  # hits DB each iteration

# ✅ Two queries total
orders = Order.objects.select_related('user').all()
for order in orders:
    print(order.user.email)

# Many-to-many / reverse FK
users = User.objects.prefetch_related('orders__items').all()
  

  # apps/users/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import User

@receiver(post_save, sender=User)
def send_welcome_email(sender, instance, created, **kwargs):
    if created:
        from .tasks import send_email
        send_email.delay(
            to=instance.email,
            subject='Welcome!',
            body=f'Hello {instance.name}',
        )
  

  class UsersConfig(AppConfig):
    def ready(self):
        import apps.users.signals  # noqa
  

  # config/middleware.py
import time
import logging

logger = logging.getLogger(__name__)

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

    def __call__(self, request):
        start = time.perf_counter()
        response = self.get_response(request)
        duration = time.perf_counter() - start
        logger.info(
            '%s %s %.3fs',
            request.method,
            request.path,
            duration,
        )
        response['X-Request-Duration'] = f'{duration:.3f}'
        return response
  

  MIDDLEWARE = [
    'config.middleware.RequestTimingMiddleware',
    # ...
]
  

  pip install djangorestframework djangorestframework-simplejwt
  

  # apps/users/serializers.py
from rest_framework import serializers
from .models import User

class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'email', 'name', 'role']
        read_only_fields = ['id']

# apps/users/views.py
from rest_framework import viewsets, permissions
from rest_framework.decorators import action
from rest_framework.response import Response
from .models import User
from .serializers import UserSerializer

class UserViewSet(viewsets.ModelViewSet):
    queryset = User.active.all()
    serializer_class = UserSerializer
    permission_classes = [permissions.IsAuthenticated]

    @action(detail=True, methods=['post'])
    def deactivate(self, request, pk=None):
        user = self.get_object()
        user.is_active = False
        user.save()
        return Response({'status': 'deactivated'})
  

  from rest_framework.routers import DefaultRouter
from apps.users.views import UserViewSet

router = DefaultRouter()
router.register('users', UserViewSet)
urlpatterns = router.urls
  

  # config/urls.py
from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView

urlpatterns = [
    path('api/token/', TokenObtainPairView.as_view()),
    path('api/token/refresh/', TokenRefreshView.as_view()),
    path('api/', include(router.urls)),
]
  

  pip install celery redis
  

  # config/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.production')
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# apps/orders/tasks.py
from celery import shared_task

@shared_task(bind=True, max_retries=3)
def process_order(self, order_id):
    from apps.orders.services import fulfill_order
    try:
        fulfill_order(order_id)
    except Exception as exc:
        self.retry(exc=exc, countdown=60)
  

  # settings
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
  

  from django.core.cache import cache
from django.views.decorators.cache import cache_page

@cache_page(60 * 15)  # 15 minutes
def product_list(request):
    products = cache.get('products:all')
    if products is None:
        products = list(Product.objects.all().values())
        cache.set('products:all', products, 900)
    return JsonResponse({'products': products})
  

  # apps/users/tests/test_views.py
import pytest
from django.urls import reverse
from rest_framework.test import APIClient
from apps.users.models import User

@pytest.fixture
def api_client():
    return APIClient()

@pytest.fixture
def auth_client(api_client):
    user = User.objects.create_user(email='[email protected]', password='pass')
    api_client.force_authenticate(user=user)
    return api_client

@pytest.mark.django_db
def test_list_users(auth_client):
    User.objects.create(email='[email protected]', name='Alice')
    response = auth_client.get('/api/users/')
    assert response.status_code == 200
    assert len(response.data) >= 1