django-production-engineering

# Django Production Engineering Complete methodology for building, scaling, and operating production Django applications. From project structure to deployment, security to performance — every decision framework a Django team needs. ## Quick Health Check Run this 8-signal triage on any Django project: | # | Signal | Check | Healthy | |---|--------|-------|---------| | 1 | Settings split | `settings/base.py`, `local.py`, `production.py` exist | ✅ Split by env | | 2 | Secret management | `SECRET_KEY` not in code, `DEBUG=False` in prod | ✅ Env vars / vault | | 3 | Database | Using connection pooling (pgbouncer / django-db-conn-pool) | ✅ Pool configured | | 4 | Migrations | `python manage.py showmigrations` — no unapplied | ✅ All applied | | 5 | Static files | `collectstatic` + CDN/whitenoise configured | ✅ Served properly | | 6 | Async tasks | Celery/django-q/Huey for background work | ✅ Not blocking views | | 7 | Caching | Cache backend configured (Redis/Memcached) | ✅ Not DummyCache | | 8 | Security | `python manage.py check --deploy` passes | ✅ All checks pass | **Score: count ✅ / 8** — Below 6 = stop and fix foundations first. --- ## Phase 1: Project Architecture ### Recommended Structure ``` myproject/ ├── config/ # Project config (was myproject/) │ ├── __init__.py │ ├── settings/ │ │ ├── __init__.py │ │ ├── base.py # Shared settings │ │ ├── local.py # Development │ │ ├── staging.py # Staging │ │ └── production.py # Production │ ├── urls.py # Root URL conf │ ├── wsgi.py │ ├── asgi.py │ └── celery.py # Celery app ├── apps/ │ ├── users/ # Custom user model (ALWAYS) │ │ ├── models.py │ │ ├── managers.py │ │ ├── admin.py │ │ ├── serializers.py │ │ ├── views.py │ │ ├── urls.py │ │ ├── services.py # Business logic │ │ ├── selectors.py # Complex queries │ │ ├── tests/ │ │ │ ├── test_models.py │ │ │ ├── test_views.py │ │ │ └── test_services.py │ │ └── migrations/ │ ├── core/ # Shared utilities │ │ ├── models.py # Abstract base models │ │ ├── permissions.py │ │ ├── pagination.py │ │ ├── exceptions.py │ │ └── middleware.py │ └── <domain>/ # Feature apps ├── templates/ ├── static/ ├── media/ ├── requirements/ │ ├── base.txt │ ├── local.txt │ └── production.txt ├── docker/ ├── scripts/ ├── manage.py ├── pyproject.toml ├── Makefile └── .env.example ``` ### 7 Architecture Rules 1. **Custom user model from Day 1** — `AUTH_USER_MODEL = 'users.User'`. Changing later is extremely painful. 2. **Fat services, thin views** — Views handle HTTP; `services.py` handles business logic; `selectors.py` handles complex queries. 3. **One app per domain** — Not per model. Group related models in one app. 4. **Settings split by environment** — Never use `if DEBUG` conditionally in a single file. 5. **Abstract base models in core** — `TimeStampedModel`, `UUIDModel` shared across apps. 6. **Separate requirements files** — `base.txt` (shared), `local.txt` (dev tools), `production.txt` (gunicorn, sentry). 7. **Keep apps decoupled** — Apps communicate through services, not by importing each other's models directly. Use signals sparingly. ### Abstract Base Models ```python # apps/core/models.py import uuid from django.db import models class TimeStampedModel(models.Model): created_at = models.DateTimeField(auto_now_add=True, db_index=True) updated_at = models.DateTimeField(auto_now=True) class Meta: abstract = True class UUIDModel(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) class Meta: abstract = True class SoftDeleteModel(models.Model): is_deleted = models.BooleanField(default=False, db_index=True) deleted_at = models.DateTimeField(null=True, blank=True) class Meta: abstract = True def soft_delete(self): from django.utils import timezone self.is_deleted = True self.deleted_at = timezone.now() self.save(update_fields=["is_deleted", "deleted_at"]) ``` --- ## Phase 2: ORM Mastery & Query Optimization ### N+1 Query Prevention ```python # ❌ N+1 — fires 1 + N queries for order in Order.objects.all(): print(order.customer.name) # Each access = new query for item in order.items.all(): # Each access = new query print(item.product.name) # Each access = new query # ✅ Optimized — fires 3 queries total orders = ( Order.objects .select_related("customer") # FK/OneToOne — JOIN .prefetch_related( Prefetch( "items", queryset=OrderItem.objects .select_related("product") # Nested FK .only("id", "quantity", "product__name") # Only needed fields ) ) ) ``` ### select_related vs prefetch_related Decision | Relationship | Use | Why | |---|---|---| | ForeignKey (forward) | `select_related` | SQL JOIN, single query | | OneToOneField | `select_related` | SQL JOIN, single query | | ManyToMany | `prefetch_related` | Separate query, Python join | | Reverse FK (set) | `prefetch_related` | Separate query, Python join | | Filtered prefetch | `Prefetch()` object | Custom queryset | ### QuerySet Evaluation Rules ```python # QuerySets are LAZY — no database hit until evaluated qs = Order.objects.filter(status="pending") # No query yet # These EVALUATE the queryset (trigger SQL): list(qs) # Iteration len(qs) # Use qs.count() instead bool(qs) # Use qs.exists() instead qs[0] # Indexing repr(qs) # In shell/debugger for obj in qs: # Iteration if qs: # Use qs.exists() ``` ### Bulk Operations ```python # ❌ N queries for item in items: Product.objects.create(name=item["name"], price=item["price"]) # ✅ 1 query Product.objects.bulk_create( [Product(name=i["name"], price=i["price"]) for i in items], batch_size=1000, ignore_conflicts=True, # Skip duplicates ) # ✅ Bulk update Product.objects.filter(category="sale").update( price=F("price") * 0.9 # 10% discount — single query, no race conditions ) # ✅ Bulk update with different values products = Product.objects.filter(id__in=ids) for p in products: p.price = new_prices[p.id] Product.objects.bulk_update(products, ["price"], batch_size=1000) ``` ### Database Functions & Expressions ```python from django.db.models import F, Q, Value, Count, Avg, Sum, Case, When from django.db.models.functions import Coalesce, Lower, TruncMonth # Conditional aggregation Order.objects.aggregate( total_revenue=Sum("amount"), paid_revenue=Sum("amount", filter=Q(status="paid")), refund_count=Count("id", filter=Q(status="refunded")), avg_order_value=Avg("amount", filter=Q(status="paid")), ) # Annotate with computed fields customers = ( Customer.objects .annotate( order_count=Count("orders"), total_spent=Coalesce(Sum("orders__amount"), Value(0)), last_order=Max("orders__created_at"), ) .filter(order_count__gte=5) .order_by("-total_spent") ) # Monthly revenue report monthly = ( Order.objects .filter(status="paid") .annotate(month=TruncMonth("created_at")) .values("month") .annotate( revenue=Sum("amount"), count=Count("id"), avg=Avg("amount"), ) .order_by("month") ) # Case/When for computed status users = User.objects.annotate( tier=Case( When(total_spent__gte=10000, then=Value("platinum")), When(total_spent__gte=5000, then=Value("gold")), When(total_spent__gte=1000, then=Value("silver")), default=Value("bronze"), ) ) ``` ### Index Strategy ```python class Order(TimeStampedModel): customer = models.ForeignKey(Customer, on_delete=models.CASCADE) status = models.CharField(max_length=20, db_index=True) # Single column amount = models.DecimalField(max_digits=10, decimal_places=2) class Meta: indexes = [ # Composite index — for queries filtering both models.Index(fields=["status", "created_at"], name="idx_order_status_created"), # Partial index — only index what you query models.Index( fields=["customer"], condition=Q(status="pending"), name="idx_order_pending_customer", ), # Covering index (Postgres) — avoid table lookup models.Index( fields=["status"], include=["amount", "created_at"], name="idx_order_status_covering", ), ] # Default ordering impacts ALL queries — be intentional ordering = ["-created_at"] ``` ### 8 ORM Rules 1. **Always use `select_related`/`prefetch_related`** — install django-debug-toolbar and watch query count. 2. **Never use `.count()` + `.all()` together** — use `.exists()` for boolean checks. 3. **Use `F()` expressions** for atomic updates — avoids race conditions. 4. **Use `.only()` / `.defer()`** for large text/JSON fields you don't need. 5. **Use `.values()` / `.values_list()`** when you don't need model instances. 6. **Use `iterator(chunk_size=2000)`** for large result sets — reduces memory. 7. **Profile with `django-silk` or `django-debug-toolbar`** — never guess at performance. 8. **Use `Exists()` subqueries** instead of `__in` with large lists. --- ## Phase 3: Django REST Framework (DRF) ### Serializer Patterns ```python # apps/orders/serializers.py from rest_framework import serializers class OrderListSerializer(serializers.ModelSerializer): """Lightweight for list views — minimal fields.""" customer_name = serializers.CharField(source="customer.name", read_only=True) class Meta: model = Order fields = ["id", "customer_name", "status", "amount", "created_at"] read_only_fields = ["id", "created_at"] class OrderDetailSerializer(serializers.ModelSerializer): """Full detail with nested items.""" items = OrderItemSerializer(many=True, read_only=True) customer = CustomerSerializer(read_only=True) class Meta: model = Order fields = "__all__" class OrderCreateSerializer(serializers.Serializer): """Explicit create — don't use ModelSerializer for writes.""" customer_id = serializers.UUIDField() items = OrderItemInputSerializer(many=True) notes = serializers.CharField(required=False, allow_blank=True) def validate_items(self, value): if not value: raise serializers.ValidationError("At least one item required.") return value def create(self, validated_data): # Delegate to service layer from apps.orders.services import create_order return create_order(**validated_data) ``` ### Service Layer Pattern ```python # apps/orders/services.py from django.db import transaction from django.core.exceptions import ValidationError def create_order(*, customer_id: str, items: list[dict], notes: str = "") -> Order: """ Create order with items atomically. Raises: ValidationError: If customer not found or insufficient stock. """ customer = Customer.objects.filter(id=customer_id).first() if not customer: raise ValidationError("Customer not found.") with transaction.atomic(): order = Order.objects.create(customer=customer, notes=notes) order_items = [] for item_data in items: product = Product.objects.select_for_update().get(id=item_data["product_id"]) if product.stock < item_data["quantity"]: raise ValidationError(f"Insufficient stock for {product.name}") product.stock -= item_data["quantity"] product.save(update_fields=["stock"]) order_items.append( OrderItem(order=order, product=product, quantity=item_data["quantity"], unit_price=product.price) ) OrderItem.objects.bulk_create(order_items) order.amount = sum(i.unit_price * i.quantity for i in order_items) order.save(update_fields=["amount"]) # Side effects OUTSIDE transaction send_order_confirmation.delay(order.id) return order ``` ### ViewSet Best Practices ```python # apps/orders/views.py from rest_framework import viewsets, status from rest_framework.decorators import action from rest_framework.response import Response class OrderViewSet(viewsets.ModelViewSet): permission_classes = [IsAuthenticated]