Table of Contents
Quick Navigation
- Django Architecture Overview
- ORM Query Optimization
- Middleware Patterns
- Security Best Practices
- Caching Strategies
- Django REST Framework
- Async Views & ASGI
- Background Tasks with Celery
- Database Configuration
- Production Deployment
- Monitoring & Logging
- Testing Strategies
- Scaling Django Apps
- Performance Tuning
- Resources & Learning Path
Django Architecture Overview
Request/Response Flow
┌─────────────────────────────────────────────────────────┐
│ Client Request │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ WSGI/ASGI Server │
│ (Gunicorn, uWSGI, Uvicorn) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Middleware Stack (Request) │
│ ┌──────────────────────────────────────────────────┐ │
│ │ SecurityMiddleware │ │
│ │ SessionMiddleware │ │
│ │ CommonMiddleware │ │
│ │ CsrfViewMiddleware │ │
│ │ AuthenticationMiddleware │ │
│ │ MessageMiddleware │ │
│ │ Custom Middleware │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ URL Dispatcher │
│ (urls.py pattern matching) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ View │
│ ┌──────────────────────────────────────────────────┐ │
│ │ • Function-based View (FBV) │ │
│ │ • Class-based View (CBV) │ │
│ │ • Generic Views │ │
│ │ • ViewSets (DRF) │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ ORM Layer │
│ ┌──────────────────────────────────────────────────┐ │
│ │ Models → QuerySets → Database Queries │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Database │
│ (PostgreSQL, MySQL, etc.) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Template Rendering │
│ (if using templates) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Middleware Stack (Response) │
│ (processed in reverse order) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Client Response │
└─────────────────────────────────────────────────────────┘
Project Structure (Production)
myproject/ ├── config/ # Project configuration │ ├── __init__.py │ ├── settings/ │ │ ├── __init__.py │ │ ├── base.py # Shared settings │ │ ├── development.py # Dev-specific │ │ ├── staging.py # Staging-specific │ │ └── production.py # Production-specific │ ├── urls.py │ ├── asgi.py │ └── wsgi.py ├── apps/ # Application modules │ ├── accounts/ │ │ ├── models.py │ │ ├── views.py │ │ ├── serializers.py │ │ ├── services.py # Business logic │ │ ├── selectors.py # Query logic │ │ └── tests/ │ ├── core/ │ └── api/ ├── static/ ├── media/ ├── templates/ ├── requirements/ │ ├── base.txt │ ├── development.txt │ ├── production.txt │ └── testing.txt ├── docker/ │ ├── Dockerfile │ ├── docker-compose.yml │ └── nginx.conf ├── scripts/ ├── .env.example ├── manage.py ├── pyproject.toml └── pytest.ini Environment Variables (.env) DEBUG=False SECRET_KEY=your-secret-key DATABASE_URL=postgresql://user:pass@host:5432/db REDIS_URL=redis://localhost:6379/0 ALLOWED_HOSTS=yourdomain.com,www.yourdomain.com AWS_ACCESS_KEY_ID=your-key AWS_SECRET_ACCESS_KEY=your-secret SENTRY_DSN=your-sentry-dsn
Settings Configuration
# config/settings/base.py
import os
from pathlib import Path
import environ
# Initialize environment variables
env = environ.Env(
DEBUG=(bool, False),
ALLOWED_HOSTS=(list, []),
)
BASE_DIR = Path(__file__).resolve().parent.parent.parent
environ.Env.read_env(os.path.join(BASE_DIR, '.env'))
SECRET_KEY = env('SECRET_KEY')
DEBUG = env('DEBUG')
ALLOWED_HOSTS = env.list('ALLOWED_HOSTS')
# Application definition
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
# Third-party apps
'rest_framework',
'django_filters',
'corsheaders',
'drf_spectacular',
'celery',
'django_redis',
# Local apps
'apps.accounts',
'apps.core',
'apps.api',
]
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'whitenoise.middleware.WhiteNoiseMiddleware',
'corsheaders.middleware.CorsMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'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'
# config/settings/production.py
from .base import *
DEBUG = False
# Security settings
SECURE_SSL_REDIRECT = True
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True
SECURE_BROWSER_XSS_FILTER = True
SECURE_CONTENT_TYPE_NOSNIFF = True
SECURE_HSTS_SECONDS = 31536000
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True
X_FRAME_OPTIONS = 'DENY'
# Database
DATABASES = {
'default': env.db('DATABASE_URL'),
}
DATABASES['default']['CONN_MAX_AGE'] = 600
DATABASES['default']['OPTIONS'] = {
'connect_timeout': 10,
}
# Cache
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': env('REDIS_URL'),
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
'PARSER_CLASS': 'redis.connection.HiredisParser',
'CONNECTION_POOL_KWARGS': {
'max_connections': 50,
'retry_on_timeout': True,
},
'SOCKET_CONNECT_TIMEOUT': 5,
'SOCKET_TIMEOUT': 5,
},
'KEY_PREFIX': 'myproject',
'TIMEOUT': 300,
}
}
# Logging
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '{levelname} {asctime} {module} {process:d} {thread:d} {message}',
'style': '{',
},
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'verbose',
},
'file': {
'class': 'logging.handlers.RotatingFileHandler',
'filename': '/var/log/django/app.log',
'maxBytes': 1024 * 1024 * 10, # 10MB
'backupCount': 5,
'formatter': 'verbose',
},
},
'root': {
'handlers': ['console', 'file'],
'level': 'INFO',
},
'loggers': {
'django': {
'handlers': ['console', 'file'],
'level': 'INFO',
'propagate': False,
},
'django.db.backends': {
'handlers': ['file'],
'level': 'WARNING',
'propagate': False,
},
},
}
ORM Query Optimization
N+1 Query Problem Solutions
# ❌ BAD: N+1 queries
def bad_view(request):
authors = Author.objects.all()
for author in authors:
# This triggers a query for EACH author
books = author.books.all()
print(f"{author.name}: {books.count()} books")
# Query count: 1 + N (where N = number of authors)
# ✅ GOOD: Use select_related for ForeignKey/OneToOne
def good_view_fk(request):
# Performs JOIN at database level
books = Book.objects.select_related(
'author',
'publisher',
'category'
).all()
for book in books:
# No additional query needed
print(f"{book.title} by {book.author.name}")
# Query count: 1
# ✅ GOOD: Use prefetch_related for ManyToMany/reverse FK
def good_view_m2m(request):
# Performs separate query and joins in Python
authors = Author.objects.prefetch_related(
'books',
'books__publisher',
'awards'
).all()
for author in authors:
# No additional queries
books = author.books.all()
print(f"{author.name}: {books.count()} books")
# Query count: 3 (authors, books, awards)
# 🚀 ADVANCED: Prefetch with filtering
from django.db.models import Prefetch
def advanced_prefetch(request):
recent_books = Book.objects.filter(
published_date__year=2024
).select_related('publisher')
authors = Author.objects.prefetch_related(
Prefetch(
'books',
queryset=recent_books,
to_attr='recent_books'
)
).all()
for author in authors:
# Access prefetched filtered books
for book in author.recent_books:
print(f"{book.title} ({book.publisher.name})")
# 🚀 ADVANCED: Custom Prefetch with annotations
from django.db.models import Count, Avg
def prefetch_with_annotations(request):
authors = Author.objects.prefetch_related(
Prefetch(
'books',
queryset=Book.objects.annotate(
review_count=Count('reviews'),
avg_rating=Avg('reviews__rating')
).order_by('-avg_rating')
)
).all()
# 📊 Measure query count
from django.db import connection
from django.test.utils import override_settings
@override_settings(DEBUG=True)
def measure_queries():
from django.db import reset_queries
reset_queries()
# Your query code here
authors = Author.objects.prefetch_related('books').all()
for author in authors:
_ = list(author.books.all())
print(f"Query count: {len(connection.queries)}")
for query in connection.queries:
print(query['sql'])
Query Optimization Techniques
# only() - Fetch only specified fields
# Only retrieves id, name, email from database
users = User.objects.only('id', 'name', 'email')
# Accessing other fields triggers additional query
for user in users:
print(user.name) # ✅ No query
print(user.bio) # ❌ Additional query
# defer() - Exclude specified fields
# Retrieves all fields except bio and profile_image
users = User.objects.defer('bio', 'profile_image')
# Good for large text/binary fields you don't need
# values() - Returns dictionaries instead of model instances
# Much faster, less memory
users = User.objects.values('id', 'name', 'email')
# Returns: [{'id': 1, 'name': 'Alice', 'email': '[email protected]'}, ...]
# values_list() - Returns tuples
# Even more efficient
user_names = User.objects.values_list('name', flat=True)
# Returns: ['Alice', 'Bob', 'Charlie', ...]
user_data = User.objects.values_list('id', 'name', 'email')
# Returns: [(1, 'Alice', '[email protected]'), ...]
# iterator() - Stream results for large querysets
# Saves memory by not caching results
for user in User.objects.iterator(chunk_size=2000):
process_user(user)
# bulk_create() - Efficient bulk inserts
books = [
Book(title=f'Book {i}', author=author)
for i in range(1000)
]
Book.objects.bulk_create(books, batch_size=500)
# Inserts all in a few queries instead of 1000
# bulk_update() - Efficient bulk updates
books = Book.objects.all()
for book in books:
book.price *= 1.1 # 10% price increase
Book.objects.bulk_update(books, ['price'], batch_size=500)
# update() - Direct SQL UPDATE
# Much faster than save() in loop
Book.objects.filter(
published_date__year=2023
).update(
status='archived',
archived_at=timezone.now()
)
# F() expressions - Database-level operations
from django.db.models import F
# Increment without race conditions
Product.objects.filter(id=product_id).update(
stock=F('stock') - 1
)
# Comparing fields in same model
expensive_books = Book.objects.filter(
price__gt=F('original_price') * 1.5
)
# Q() objects - Complex queries
from django.db.models import Q
# OR condition
books = Book.objects.filter(
Q(title__icontains='python') | Q(title__icontains='django')
)
# Complex nested conditions
books = Book.objects.filter(
Q(published_date__year=2024) &
(Q(author__name='Alice') | Q(author__name='Bob')) &
~Q(status='draft')
)
# exists() - Check existence efficiently
# Instead of:
if Book.objects.filter(isbn=isbn).count() > 0: # ❌ Counts all
pass
# Use:
if Book.objects.filter(isbn=isbn).exists(): # ✅ Just checks
pass
# Aggregation and Annotation
from django.db.models import Count, Avg, Sum, Max, Min
# Aggregate over entire queryset
stats = Book.objects.aggregate(
total_books=Count('id'),
avg_price=Avg('price'),
total_revenue=Sum('price'),
max_price=Max('price')
)
# Returns: {'total_books': 150, 'avg_price': 29.99, ...}
# Annotate each object
authors = Author.objects.annotate(
book_count=Count('books'),
avg_rating=Avg('books__reviews__rating'),
total_sales=Sum('books__sales')
).filter(book_count__gt=5)
for author in authors:
print(f"{author.name}: {author.book_count} books, {author.avg_rating} rating")
Database Indexes
# Basic index on single field
class Book(models.Model):
title = models.CharField(max_length=200, db_index=True)
isbn = models.CharField(max_length=13, unique=True) # Implicit index
published_date = models.DateField(db_index=True)
# Composite indexes (multiple fields)
class Book(models.Model):
title = models.CharField(max_length=200)
author = models.ForeignKey(Author, on_delete=models.CASCADE)
status = models.CharField(max_length=20)
class Meta:
indexes = [
models.Index(fields=['author', 'status']),
models.Index(fields=['published_date', '-price']),
models.Index(fields=['status', 'created_at']),
]
# Named indexes
class Book(models.Model):
class Meta:
indexes = [
models.Index(
fields=['title'],
name='book_title_idx'
),
]
# Partial indexes (PostgreSQL)
from django.db.models import Q
class Book(models.Model):
class Meta:
indexes = [
models.Index(
fields=['published_date'],
name='published_books_idx',
condition=Q(status='published')
),
]
# Functional indexes (PostgreSQL 11+)
from django.db.models.functions import Lower
class Book(models.Model):
class Meta:
indexes = [
models.Index(
Lower('title'),
name='book_title_lower_idx'
),
]
# Query using functional index
books = Book.objects.filter(title__lower='python basics')
# Full-text search indexes (PostgreSQL)
from django.contrib.postgres.search import SearchVector
from django.contrib.postgres.indexes import GinIndex
class Article(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
class Meta:
indexes = [
GinIndex(
fields=['title', 'content'],
name='article_search_idx',
opclasses=['gin_trgm_ops', 'gin_trgm_ops']
),
]
# Check query plan
def check_query_plan():
qs = Book.objects.filter(
author__name='Alice',
status='published'
).select_related('author')
# PostgreSQL
print(qs.explain(analyze=True, verbose=True))
# Look for:
# - "Index Scan" (good) vs "Seq Scan" (bad for large tables)
# - Actual time vs estimated time
# - Rows processed
# Index usage patterns
# ✅ Index will be used
Book.objects.filter(author_id=1) # Direct ID lookup
Book.objects.filter(title='Python') # Exact match
Book.objects.filter(title__startswith='Python') # Prefix match
# ❌ Index won't be used efficiently
Book.objects.filter(title__icontains='python') # Case-insensitive contains
Book.objects.filter(title__endswith='Python') # Suffix match
Book.objects.exclude(status='draft') # Negation
# Monitor index usage (PostgreSQL)
# Add to management command
from django.db import connection
def check_index_usage():
with connection.cursor() as cursor:
cursor.execute("""
SELECT
schemaname,
tablename,
indexname,
idx_scan,
idx_tup_read,
idx_tup_fetch
FROM pg_stat_user_indexes
WHERE schemaname = 'public'
ORDER BY idx_scan ASC;
""")
for row in cursor.fetchall():
print(row)
Raw SQL & Custom Queries
# raw() - Execute custom SQL returning model instances
books = Book.objects.raw(
'SELECT * FROM books_book WHERE price > %s',
[20.00]
)
for book in books:
print(book.title) # Full model instance
# raw() with mapping
books = Book.objects.raw(
'''
SELECT
b.id,
b.title,
a.name as author_name,
COUNT(r.id) as review_count
FROM books_book b
JOIN authors_author a ON b.author_id = a.id
LEFT JOIN reviews_review r ON r.book_id = b.id
GROUP BY b.id, b.title, a.name
'''
)
# Direct cursor for complex queries
from django.db import connection
def get_sales_report():
with connection.cursor() as cursor:
cursor.execute("""
SELECT
DATE_TRUNC('month', order_date) as month,
SUM(total) as revenue,
COUNT(*) as order_count
FROM orders
WHERE order_date >= %s
GROUP BY DATE_TRUNC('month', order_date)
ORDER BY month DESC
""", [start_date])
columns = [col[0] for col in cursor.description]
results = [
dict(zip(columns, row))
for row in cursor.fetchall()
]
return results
# RawSQL for annotations
from django.db.models.expressions import RawSQL
books = Book.objects.annotate(
full_text_rank=RawSQL(
"ts_rank(to_tsvector('english', title || ' ' || description), plainto_tsquery(%s))",
('python',)
)
).filter(full_text_rank__gt=0).order_by('-full_text_rank')
# Execute stored procedures
def call_stored_procedure():
with connection.cursor() as cursor:
cursor.callproc('calculate_monthly_revenue', [2024, 1])
results = cursor.fetchall()
return results
# SQL injection prevention
# ❌ NEVER do this
def bad_query(user_input):
Book.objects.raw(
f'SELECT * FROM books WHERE title = "{user_input}"'
)
# ✅ Always use parameterized queries
def good_query(user_input):
Book.objects.raw(
'SELECT * FROM books WHERE title = %s',
[user_input]
)
# Extra() for complex WHERE clauses (legacy)
# Note: Use Q() objects or RawSQL instead when possible
books = Book.objects.extra(
select={'is_bestseller': "sales > 10000"},
where=["published_date > '2020-01-01'"],
order_by=['-sales']
)
Middleware Patterns
Custom Middleware Structure
# Middleware execution flow
┌─────────────────────────────────────────────────────────┐
│ Request Phase │
│ (top to bottom) │
│ ┌────────────────────────────────────────────────────┐│
│ │ Middleware 1 (process_request) ││
│ │ ↓ ││
│ │ Middleware 2 (process_request) ││
│ │ ↓ ││
│ │ Middleware 3 (process_request) ││
│ │ ↓ ││
│ │ View Function ││
│ │ ↓ ││
│ │ Middleware 3 (process_response) ││
│ │ ↓ ││
│ │ Middleware 2 (process_response) ││
│ │ ↓ ││
│ │ Middleware 1 (process_response) ││
│ └────────────────────────────────────────────────────┘│
│ Response Phase │
│ (bottom to top) │
└─────────────────────────────────────────────────────────┘
# Basic middleware class
class SimpleMiddleware:
def __init__(self, get_response):
self.get_response = get_response
# One-time configuration and initialization
def __call__(self, request):
# Code executed for each request before
# the view (and later middleware) are called
response = self.get_response(request)
# Code executed for each request/response after
# the view is called
return response
# Full middleware with all hooks
import logging
from django.utils.deprecation import MiddlewareMixin
logger = logging.getLogger(__name__)
class FullMiddleware(MiddlewareMixin):
def process_request(self, request):
"""
Called on each request, before Django decides
which view to execute
Return None to continue processing
Return HttpResponse to short-circuit
"""
logger.info(f"Request: {request.method} {request.path}")
request.start_time = time.time()
return None
def process_view(self, request, view_func, view_args, view_kwargs):
"""
Called just before Django calls the view
Args:
view_func: Python function Django is about to use
view_args: Positional arguments
view_kwargs: Keyword arguments
"""
logger.info(f"View: {view_func.__name__}")
return None
def process_exception(self, request, exception):
"""
Called when a view raises an exception
Return None to continue exception handling
Return HttpResponse to handle the exception
"""
logger.error(f"Exception: {exception}")
return None
def process_template_response(self, request, response):
"""
Called just after view finished executing
Only for responses with render() method
"""
return response
def process_response(self, request, response):
"""
Called on all responses before returning to browser
Must return HttpResponse object
"""
if hasattr(request, 'start_time'):
duration = time.time() - request.start_time
logger.info(f"Response time: {duration:.3f}s")
return response
Production Middleware Examples
# 1. Request ID Middleware
import uuid
from django.utils.deprecation import MiddlewareMixin
class RequestIDMiddleware(MiddlewareMixin):
"""Adds unique request ID for tracing"""
def process_request(self, request):
request.id = request.META.get(
'HTTP_X_REQUEST_ID',
str(uuid.uuid4())
)
request.META['HTTP_X_REQUEST_ID'] = request.id
def process_response(self, request, response):
if hasattr(request, 'id'):
response['X-Request-ID'] = request.id
return response
# 2. Performance Monitoring Middleware
import time
from django.conf import settings
from django.core.cache import cache
class PerformanceMiddleware:
def __init__(self, get_response):
self.get_response = get_response
self.slow_threshold = settings.SLOW_REQUEST_THRESHOLD
def __call__(self, request):
start_time = time.time()
response = self.get_response(request)
duration = time.time() - start_time
# Log slow requests
if duration > self.slow_threshold:
logger.warning(
f"Slow request: {request.method} {request.path} "
f"took {duration:.3f}s"
)
# Add timing header
response['X-Response-Time'] = f"{duration:.3f}s"
# Track endpoint performance
cache_key = f"perf:{request.path}"
cache.incr(cache_key, delta=int(duration * 1000))
return response
# 3. Rate Limiting Middleware
from django.core.cache import cache
from django.http import HttpResponse
class RateLimitMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
# Skip rate limiting for authenticated staff
if request.user.is_authenticated and request.user.is_staff:
return self.get_response(request)
# Get client IP
ip = self.get_client_ip(request)
# Rate limit key
cache_key = f"rate_limit:{ip}"
# Get request count
requests = cache.get(cache_key, 0)
if requests >= 100: # 100 requests per minute
return HttpResponse(
'Rate limit exceeded',
status=429
)
# Increment counter
cache.set(cache_key, requests + 1, 60)
response = self.get_response(request)
response['X-RateLimit-Limit'] = '100'
response['X-RateLimit-Remaining'] = str(100 - requests - 1)
return response
def get_client_ip(self, request):
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
if x_forwarded_for:
ip = x_forwarded_for.split(',')[0]
else:
ip = request.META.get('REMOTE_ADDR')
return ip
# 4. Database Query Counter Middleware
from django.db import connection
from django.conf import settings
class QueryCountMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
# Reset query log
from django.db import reset_queries
reset_queries()
response = self.get_response(request)
if settings.DEBUG:
query_count = len(connection.queries)
# Warn about excessive queries
if query_count > 50:
logger.warning(
f"High query count: {query_count} queries "
f"for {request.path}"
)
# Add header
response['X-DB-Query-Count'] = str(query_count)
return response
# 5. CORS Middleware (Custom)
class CORSMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
response = self.get_response(request)
# Add CORS headers
response['Access-Control-Allow-Origin'] = '*'
response['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS'
response['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
response['Access-Control-Max-Age'] = '3600'
return response
# 6. API Version Middleware
import re
class APIVersionMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
# Extract version from URL or header
version_pattern = r'/api/v(\d+)/'
match = re.search(version_pattern, request.path)
if match:
request.api_version = int(match.group(1))
else:
request.api_version = int(
request.META.get('HTTP_API_VERSION', 1)
)
return self.get_response(request)
# 7. Authentication Token Middleware
from rest_framework.authtoken.models import Token
class TokenAuthMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
auth_header = request.META.get('HTTP_AUTHORIZATION', '')
if auth_header.startswith('Token '):
token_key = auth_header.split(' ')[1]
try:
token = Token.objects.select_related('user').get(
key=token_key
)
request.user = token.user
except Token.DoesNotExist:
pass
return self.get_response(request)
# Registering middleware in settings
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'myapp.middleware.RequestIDMiddleware',
'whitenoise.middleware.WhiteNoiseMiddleware',
'corsheaders.middleware.CorsMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'myapp.middleware.PerformanceMiddleware',
'myapp.middleware.RateLimitMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
Security Best Practices
Security Settings Checklist
# Production security settings (settings/production.py)
# 1. Secret Key
# NEVER commit to version control
SECRET_KEY = env('SECRET_KEY') # From environment variable
# Generate with: python -c 'from django.core.management.utils import get_random_secret_key; print(get_random_secret_key())'
# 2. Debug Mode
DEBUG = False # MUST be False in production
ALLOWED_HOSTS = env.list('ALLOWED_HOSTS') # Specific domains only
# 3. HTTPS/SSL
SECURE_SSL_REDIRECT = True # Redirect all HTTP to HTTPS
SESSION_COOKIE_SECURE = True # Send session cookie only over HTTPS
CSRF_COOKIE_SECURE = True # Send CSRF cookie only over HTTPS
# 4. HSTS (HTTP Strict Transport Security)
SECURE_HSTS_SECONDS = 31536000 # 1 year
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True
# 5. Content Security
X_FRAME_OPTIONS = 'DENY' # Prevent clickjacking
SECURE_CONTENT_TYPE_NOSNIFF = True # Prevent MIME sniffing
SECURE_BROWSER_XSS_FILTER = True # Enable XSS filter
# 6. Cookie Security
SESSION_COOKIE_HTTPONLY = True # JavaScript can't access
SESSION_COOKIE_SAMESITE = 'Strict' # CSRF protection
SESSION_COOKIE_AGE = 3600 # 1 hour session timeout
CSRF_COOKIE_HTTPONLY = True
CSRF_COOKIE_SAMESITE = 'Strict'
# 7. Password Validation
AUTH_PASSWORD_VALIDATORS = [
{
'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
'OPTIONS': {'min_length': 12},
},
{
'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',
},
]
# 8. Admin Security
ADMIN_URL = env('ADMIN_URL', default='admin/') # Obscure admin URL
# 9. CORS (if using django-cors-headers)
CORS_ALLOWED_ORIGINS = [
'https://yourdomain.com',
'https://www.yourdomain.com',
]
CORS_ALLOW_CREDENTIALS = True
# 10. CSP (Content Security Policy)
CSP_DEFAULT_SRC = ("'self'",)
CSP_SCRIPT_SRC = ("'self'", 'https://cdn.example.com')
CSP_STYLE_SRC = ("'self'", "'unsafe-inline'")
CSP_IMG_SRC = ("'self'", 'data:', 'https:')
CSP_FONT_SRC = ("'self'", 'data:')
# 11. File Upload Security
FILE_UPLOAD_MAX_MEMORY_SIZE = 5242880 # 5MB
DATA_UPLOAD_MAX_MEMORY_SIZE = 5242880
FILE_UPLOAD_PERMISSIONS = 0o644
FILE_UPLOAD_DIRECTORY_PERMISSIONS = 0o755
# Validate file types
ALLOWED_UPLOAD_EXTENSIONS = ['.jpg', '.jpeg', '.png', '.gif', '.pdf']
# 12. Database Security
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'),
'CONN_MAX_AGE': 600,
'OPTIONS': {
'sslmode': 'require', # Require SSL connection
},
}
}
# 13. Logging Security
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '{levelname} {asctime} {module} {message}',
'style': '{',
},
},
'filters': {
'require_debug_false': {
'()': 'django.utils.log.RequireDebugFalse',
},
},
'handlers': {
'file': {
'level': 'WARNING',
'class': 'logging.handlers.RotatingFileHandler',
'filename': '/var/log/django/app.log',
'maxBytes': 1024 * 1024 * 10,
'backupCount': 5,
'formatter': 'verbose',
},
'mail_admins': {
'level': 'ERROR',
'class': 'django.utils.log.AdminEmailHandler',
'filters': ['require_debug_false'],
},
},
'loggers': {
'django.security': {
'handlers': ['file', 'mail_admins'],
'level': 'WARNING',
'propagate': False,
},
},
}
SQL Injection Prevention
# ✅ SAFE: ORM automatically escapes
def safe_query(user_input):
books = Book.objects.filter(title=user_input)
return books
# ✅ SAFE: Parameterized queries
def safe_raw_query(user_input):
books = Book.objects.raw(
'SELECT * FROM books WHERE title = %s',
[user_input]
)
return list(books)
# ✅ SAFE: Named parameters
def safe_cursor_query(user_input):
from django.db import connection
with connection.cursor() as cursor:
cursor.execute(
'SELECT * FROM books WHERE title = %(title)s',
{'title': user_input}
)
results = cursor.fetchall()
return results
# ❌ DANGEROUS: String formatting
def dangerous_query(user_input):
# Never do this!
books = Book.objects.raw(
f'SELECT * FROM books WHERE title = "{user_input}"'
)
# user_input could be: " OR 1=1 --"
# ❌ DANGEROUS: String concatenation
def dangerous_cursor(user_input):
with connection.cursor() as cursor:
cursor.execute(
'SELECT * FROM books WHERE title = "' + user_input + '"'
)
# ✅ SAFE: extra() with params
def safe_extra(category_id):
books = Book.objects.extra(
where=['category_id = %s'],
params=[category_id]
)
return books
# Validate input before queries
def validate_and_query(user_input):
# Whitelist validation
if user_input not in ['published', 'draft', 'archived']:
raise ValueError('Invalid status')
books = Book.objects.filter(status=user_input)
return books
# Use Q objects for complex conditions
from django.db.models import Q
def safe_complex_query(search_term):
# Q objects are automatically escaped
books = Book.objects.filter(
Q(title__icontains=search_term) |
Q(description__icontains=search_term)
)
return books
XSS Prevention
# Template auto-escaping (enabled by default){{ user.bio }}
{{ user.bio|safe }}
{{ user.bio|linebreaks }}
{{ user.bio|urlize }}
# Python code escaping from django.utils.html import escape, format_html def display_user_input(user_input): # ✅ SAFE: Escape before displaying safe_input = escape(user_input) return f'{safe_input}
' def display_with_link(name, url): # ✅ SAFE: format_html escapes parameters return format_html( '{}', url, name ) # JSON responses from django.http import JsonResponse def api_view(request): user_data = { 'name': request.user.name, # Will be JSON-escaped 'bio': request.user.bio, } # ✅ SAFE: JsonResponse handles escaping return JsonResponse(user_data) # Mark strings as safe (use carefully) from django.utils.safestring import mark_safe def render_html(): # Only mark trusted content as safe html = 'This is trusted content
' return mark_safe(html) # Sanitize HTML input import bleach ALLOWED_TAGS = ['p', 'br', 'strong', 'em', 'a'] ALLOWED_ATTRIBUTES = {'a': ['href', 'title']} def sanitize_html(html_content): # ✅ SAFE: Remove dangerous tags/attributes clean_html = bleach.clean( html_content, tags=ALLOWED_TAGS, attributes=ALLOWED_ATTRIBUTES, strip=True ) return clean_html # Content Security Policy # settings.py CSP_DEFAULT_SRC = ("'self'",) CSP_SCRIPT_SRC = ("'self'",) # No inline scripts CSP_STYLE_SRC = ("'self'", "'unsafe-inline'") # Allow inline styles CSP_IMG_SRC = ("'self'", 'data:', 'https:')
CSRF Protection
# CSRF protection is enabled by default
# Ensure middleware is active:
MIDDLEWARE = [
# ...
'django.middleware.csrf.CsrfViewMiddleware',
# ...
]
# Templates: Add CSRF token
# Exempt views from CSRF (use carefully)
from django.views.decorators.csrf import csrf_exempt
from django.utils.decorators import method_decorator
# Function-based view
@csrf_exempt
def webhook_view(request):
# For external webhooks
pass
# Class-based view
@method_decorator(csrf_exempt, name='dispatch')
class WebhookView(View):
def post(self, request):
pass
# Ensure CSRF for specific view
from django.views.decorators.csrf import csrf_protect
@csrf_protect
def sensitive_view(request):
pass
# Custom CSRF failure view
# settings.py
CSRF_FAILURE_VIEW = 'myapp.views.csrf_failure'
# views.py
def csrf_failure(request, reason=""):
return render(request, 'csrf_failure.html', {
'reason': reason
}, status=403)
# CSRF settings
# settings.py
CSRF_COOKIE_SECURE = True # HTTPS only
CSRF_COOKIE_HTTPONLY = True # No JavaScript access
CSRF_COOKIE_SAMESITE = 'Strict' # Strict same-site policy
CSRF_USE_SESSIONS = False # Store in cookie (default)
CSRF_COOKIE_AGE = 31449600 # 1 year
# Trusted origins for unsafe methods
CSRF_TRUSTED_ORIGINS = [
'https://yourdomain.com',
'https://www.yourdomain.com',
]
Authentication & Authorization
# Custom User Model (best practice)
# models.py
from django.contrib.auth.models import AbstractUser
class User(AbstractUser):
email = models.EmailField(unique=True)
phone = models.CharField(max_length=20, blank=True)
is_verified = models.BooleanField(default=False)
USERNAME_FIELD = 'email'
REQUIRED_FIELDS = ['username']
# settings.py
AUTH_USER_MODEL = 'accounts.User'
# Permission decorators
from django.contrib.auth.decorators import (
login_required,
permission_required,
user_passes_test
)
@login_required
def profile_view(request):
return render(request, 'profile.html')
@permission_required('books.add_book', raise_exception=True)
def create_book_view(request):
pass
def is_staff_user(user):
return user.is_staff
@user_passes_test(is_staff_user)
def admin_dashboard(request):
pass
# Class-based view permissions
from django.contrib.auth.mixins import (
LoginRequiredMixin,
PermissionRequiredMixin,
UserPassesTestMixin
)
class ProfileView(LoginRequiredMixin, View):
login_url = '/login/'
redirect_field_name = 'next'
def get(self, request):
return render(request, 'profile.html')
class CreateBookView(PermissionRequiredMixin, CreateView):
permission_required = 'books.add_book'
model = Book
fields = ['title', 'author']
class EditOwnBookView(UserPassesTestMixin, UpdateView):
model = Book
fields = ['title', 'description']
def test_func(self):
book = self.get_object()
return self.request.user == book.author
# Object-level permissions
def edit_book_view(request, book_id):
book = get_object_or_404(Book, id=book_id)
# Check if user can edit
if book.author != request.user and not request.user.is_staff:
raise PermissionDenied
# Process edit
pass
# Custom permission class
from django.contrib.auth.models import Permission
from django.contrib.contenttypes.models import ContentType
def create_custom_permission():
content_type = ContentType.objects.get_for_model(Book)
permission = Permission.objects.create(
codename='can_publish_book',
name='Can Publish Book',
content_type=content_type,
)
# Check permission
if request.user.has_perm('books.can_publish_book'):
book.status = 'published'
book.save()
# Token-based authentication
# settings.py
INSTALLED_APPS = [
# ...
'rest_framework',
'rest_framework.authtoken',
]
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.TokenAuthentication',
'rest_framework.authentication.SessionAuthentication',
],
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticated',
],
}
# Generate token
from rest_framework.authtoken.models import Token
token = Token.objects.create(user=user)
print(token.key)
# Use in API request
# Header: Authorization: Token 9944b09199c62bcf9418ad846dd0e4bbdfc6ee4b
# JWT Authentication (djangorestframework-simplejwt)
# settings.py
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework_simplejwt.authentication.JWTAuthentication',
],
}
from datetime import timedelta
SIMPLE_JWT = {
'ACCESS_TOKEN_LIFETIME': timedelta(minutes=15),
'REFRESH_TOKEN_LIFETIME': timedelta(days=7),
'ROTATE_REFRESH_TOKENS': True,
'BLACKLIST_AFTER_ROTATION': True,
}
# urls.py
from rest_framework_simplejwt.views import (
TokenObtainPairView,
TokenRefreshView,
)
urlpatterns = [
path('api/token/', TokenObtainPairView.as_view()),
path('api/token/refresh/', TokenRefreshView.as_view()),
]
# Rate limiting per user
from django.core.cache import cache
from django.http import HttpResponseForbidden
def rate_limit_user(view_func):
def wrapper(request, *args, **kwargs):
if not request.user.is_authenticated:
return view_func(request, *args, **kwargs)
cache_key = f'rate_limit_user:{request.user.id}'
requests = cache.get(cache_key, 0)
if requests >= 100:
return HttpResponseForbidden('Rate limit exceeded')
cache.set(cache_key, requests + 1, 3600)
return view_func(request, *args, **kwargs)
return wrapper
Caching Strategies
Cache Configuration
# Redis Cache (Production)
# settings.py
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1',
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
'PARSER_CLASS': 'redis.connection.HiredisParser',
'CONNECTION_POOL_KWARGS': {
'max_connections': 50,
'retry_on_timeout': True,
},
'SOCKET_CONNECT_TIMEOUT': 5,
'SOCKET_TIMEOUT': 5,
'COMPRESSOR': 'django_redis.compressors.zlib.ZlibCompressor',
'IGNORE_EXCEPTIONS': True, # Fail gracefully
},
'KEY_PREFIX': 'myapp',
'TIMEOUT': 300, # 5 minutes default
},
# Separate cache for sessions
'sessions': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/2',
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
},
'KEY_PREFIX': 'session',
},
}
# Use Redis for sessions
SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
SESSION_CACHE_ALIAS = 'sessions'
# Memcached (Alternative)
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.memcached.PyMemcacheCache',
'LOCATION': [
'127.0.0.1:11211',
'127.0.0.1:11212', # Multiple servers
],
'OPTIONS': {
'no_delay': True,
'ignore_exc': True,
'max_pool_size': 4,
'use_pooling': True,
},
'KEY_PREFIX': 'myapp',
'TIMEOUT': 300,
}
}
# Local Memory Cache (Development)
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
'LOCATION': 'unique-snowflake',
'OPTIONS': {
'MAX_ENTRIES': 1000,
},
}
}
# Database Cache
# python manage.py createcachetable
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.db.DatabaseCache',
'LOCATION': 'cache_table',
}
}
# File-based Cache
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.filebased.FileBasedCache',
'LOCATION': '/var/tmp/django_cache',
'OPTIONS': {
'MAX_ENTRIES': 1000,
},
}
}
Cache Usage Patterns
# Low-level cache API
from django.core.cache import cache
# Set cache
cache.set('my_key', 'my_value', timeout=300) # 5 minutes
# Get cache
value = cache.get('my_key')
if value is None:
value = expensive_computation()
cache.set('my_key', value, timeout=300)
# Get with default
value = cache.get('my_key', default='default_value')
# Get or set in one call
value = cache.get_or_set('my_key', expensive_computation, timeout=300)
# Delete cache
cache.delete('my_key')
# Clear all cache
cache.clear()
# Multiple operations
cache.set_many({'a': 1, 'b': 2, 'c': 3}, timeout=300)
values = cache.get_many(['a', 'b', 'c']) # {'a': 1, 'b': 2, 'c': 3}
cache.delete_many(['a', 'b', 'c'])
# Increment/Decrement (atomic)
cache.set('counter', 0)
cache.incr('counter') # Returns 1
cache.incr('counter', delta=10) # Returns 11
cache.decr('counter', delta=5) # Returns 6
# Cache decorators
from django.views.decorators.cache import cache_page
# Cache view for 15 minutes
@cache_page(60 * 15)
def my_view(request):
# Expensive operations
return render(request, 'template.html')
# Cache with key prefix
@cache_page(60 * 15, key_prefix='special')
def my_view(request):
pass
# Per-user caching
from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_page
def cache_per_user(timeout):
def decorator(view_func):
def wrapper(request, *args, **kwargs):
cache_key = f'view:{request.user.id}:{request.path}'
result = cache.get(cache_key)
if result is None:
result = view_func(request, *args, **kwargs)
cache.set(cache_key, result, timeout)
return result
return wrapper
return decorator
@cache_per_user(60 * 15)
def user_dashboard(request):
pass
# Template fragment caching
{% load cache %}
{% cache 300 sidebar %}
{% include "sidebar.html" %}
{% endcache %}
{% cache 300 sidebar request.user.id %}
{% endcache %}
# QuerySet caching pattern
class BookManager(models.Manager):
def published(self):
cache_key = 'books:published'
books = cache.get(cache_key)
if books is None:
books = list(
self.filter(status='published')
.select_related('author')
.prefetch_related('categories')
)
cache.set(cache_key, books, timeout=300)
return books
# Model
class Book(models.Model):
objects = BookManager()
# Usage
books = Book.objects.published()
# Invalidation patterns
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
@receiver([post_save, post_delete], sender=Book)
def invalidate_book_cache(sender, instance, **kwargs):
cache_keys = [
'books:published',
f'book:{instance.id}',
f'author:{instance.author_id}:books',
]
cache.delete_many(cache_keys)
# Cache warming
from django.core.management.base import BaseCommand
class Command(BaseCommand):
help = 'Warm up cache with frequently accessed data'
def handle(self, *args, **options):
# Published books
books = list(Book.objects.filter(status='published'))
cache.set('books:published', books, timeout=3600)
# Popular categories
categories = list(Category.objects.annotate(
book_count=Count('books')
).order_by('-book_count')[:10])
cache.set('categories:popular', categories, timeout=3600)
self.stdout.write(self.style.SUCCESS('Cache warmed successfully'))
# Cache with versioning
CACHE_VERSION = 1
cache.set('my_key', value, version=CACHE_VERSION)
value = cache.get('my_key', version=CACHE_VERSION)
# Invalidate all caches with version bump
CACHE_VERSION = 2 # All previous caches now invalid
Advanced Caching Strategies
# Cache-aside pattern (Lazy loading)
def get_user_stats(user_id):
cache_key = f'user_stats:{user_id}'
stats = cache.get(cache_key)
if stats is None:
# Calculate expensive stats
stats = {
'total_orders': Order.objects.filter(user_id=user_id).count(),
'total_spent': Order.objects.filter(
user_id=user_id
).aggregate(Sum('total'))['total__sum'] or 0,
'favorite_category': get_favorite_category(user_id),
}
cache.set(cache_key, stats, timeout=3600)
return stats
# Write-through cache
def create_order(user_id, items):
# Create order in database
order = Order.objects.create(user_id=user_id)
order.items.add(*items)
# Update cache immediately
cache_key = f'user_stats:{user_id}'
stats = cache.get(cache_key)
if stats:
stats['total_orders'] += 1
stats['total_spent'] += order.total
cache.set(cache_key, stats, timeout=3600)
return order
# Cache stampede prevention (Lock pattern)
import time
from django.core.cache import cache
def get_expensive_data(key):
lock_key = f'lock:{key}'
data_key = f'data:{key}'
# Try to get cached data
data = cache.get(data_key)
if data is not None:
return data
# Try to acquire lock
if cache.add(lock_key, 'locked', timeout=10):
try:
# We got the lock, compute the data
data = expensive_computation()
cache.set(data_key, data, timeout=300)
return data
finally:
cache.delete(lock_key)
else:
# Someone else is computing, wait and retry
time.sleep(0.1)
return get_expensive_data(key)
# Stale-while-revalidate pattern
def get_data_with_swr(key, compute_func, ttl=300):
cache_key = f'data:{key}'
stale_key = f'stale:{key}'
computing_key = f'computing:{key}'
# Try to get fresh data
data = cache.get(cache_key)
if data is not None:
return data
# Try to get stale data
stale_data = cache.get(stale_key)
# Start background refresh if not already computing
if not cache.get(computing_key):
cache.set(computing_key, True, timeout=60)
# Trigger async task to refresh
from .tasks import refresh_cache
refresh_cache.delay(key, compute_func, ttl)
# Return stale data if available
if stale_data is not None:
return stale_data
# Compute synchronously as last resort
data = compute_func()
cache.set(cache_key, data, timeout=ttl)
cache.set(stale_key, data, timeout=ttl * 2)
cache.delete(computing_key)
return data
# Multi-tier caching
class MultiTierCache:
def __init__(self):
self.local_cache = {} # In-process cache
self.distributed_cache = cache # Redis/Memcached
def get(self, key):
# Try local cache first (fastest)
if key in self.local_cache:
return self.local_cache[key]
# Try distributed cache
value = self.distributed_cache.get(key)
if value is not None:
# Populate local cache
self.local_cache[key] = value
return value
return None
def set(self, key, value, timeout=300):
# Set in both caches
self.local_cache[key] = value
self.distributed_cache.set(key, value, timeout)
def delete(self, key):
# Delete from both caches
self.local_cache.pop(key, None)
self.distributed_cache.delete(key)
# Cache key generation
import hashlib
import json
def generate_cache_key(prefix, **kwargs):
"""Generate consistent cache key from parameters"""
# Sort kwargs for consistency
params = json.dumps(kwargs, sort_keys=True)
hash_key = hashlib.md5(params.encode()).hexdigest()
return f'{prefix}:{hash_key}'
# Usage
cache_key = generate_cache_key(
'books',
status='published',
category_id=5,
sort='-published_date'
)
# Cache monitoring
from django.core.management.base import BaseCommand
class Command(BaseCommand):
help = 'Monitor cache hit/miss rates'
def handle(self, *args, **options):
# Get Redis info
from django_redis import get_redis_connection
redis_conn = get_redis_connection('default')
info = redis_conn.info('stats')
hits = info['keyspace_hits']
misses = info['keyspace_misses']
total = hits + misses
if total > 0:
hit_rate = (hits / total) * 100
self.stdout.write(f'Cache hit rate: {hit_rate:.2f}%')
# Check memory usage
memory_info = redis_conn.info('memory')
used_memory = memory_info['used_memory_human']
self.stdout.write(f'Memory used: {used_memory}')
Django REST Framework
DRF Setup & Configuration
# Installation
pip install djangorestframework
pip install django-filter
pip install drf-spectacular # OpenAPI schema
# Settings configuration
# settings.py
INSTALLED_APPS = [
# ...
'rest_framework',
'django_filters',
'drf_spectacular',
]
REST_FRAMEWORK = {
# Authentication
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.SessionAuthentication',
'rest_framework.authentication.TokenAuthentication',
'rest_framework_simplejwt.authentication.JWTAuthentication',
],
# Permissions
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticatedOrReadOnly',
],
# Pagination
'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
'PAGE_SIZE': 20,
# Filtering
'DEFAULT_FILTER_BACKENDS': [
'django_filters.rest_framework.DjangoFilterBackend',
'rest_framework.filters.SearchFilter',
'rest_framework.filters.OrderingFilter',
],
# Throttling
'DEFAULT_THROTTLE_CLASSES': [
'rest_framework.throttling.AnonRateThrottle',
'rest_framework.throttling.UserRateThrottle',
],
'DEFAULT_THROTTLE_RATES': {
'anon': '100/hour',
'user': '1000/hour',
},
# Rendering
'DEFAULT_RENDERER_CLASSES': [
'rest_framework.renderers.JSONRenderer',
],
# Versioning
'DEFAULT_VERSIONING_CLASS': 'rest_framework.versioning.NamespaceVersioning',
# Schema
'DEFAULT_SCHEMA_CLASS': 'drf_spectacular.openapi.AutoSchema',
# Exception handling
'EXCEPTION_HANDLER': 'myapp.utils.custom_exception_handler',
# Test settings
'TEST_REQUEST_DEFAULT_FORMAT': 'json',
}
# Spectacular (OpenAPI) settings
SPECTACULAR_SETTINGS = {
'TITLE': 'My API',
'DESCRIPTION': 'API documentation',
'VERSION': '1.0.0',
'SERVE_INCLUDE_SCHEMA': False,
'SWAGGER_UI_SETTINGS': {
'deepLinking': True,
'persistAuthorization': True,
'displayOperationId': True,
},
}
# URLs configuration
# urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from drf_spectacular.views import (
SpectacularAPIView,
SpectacularSwaggerView,
)
router = DefaultRouter()
router.register(r'books', BookViewSet)
router.register(r'authors', AuthorViewSet)
urlpatterns = [
path('api/', include(router.urls)),
path('api/schema/', SpectacularAPIView.as_view(), name='schema'),
path('api/docs/', SpectacularSwaggerView.as_view(url_name='schema')),
path('api-auth/', include('rest_framework.urls')),
]
Serializers
# Basic ModelSerializer
from rest_framework import serializers
class AuthorSerializer(serializers.ModelSerializer):
class Meta:
model = Author
fields = ['id', 'name', 'bio', 'created_at']
read_only_fields = ['created_at']
class BookSerializer(serializers.ModelSerializer):
author = AuthorSerializer(read_only=True)
author_id = serializers.IntegerField(write_only=True)
class Meta:
model = Book
fields = [
'id', 'title', 'description', 'price',
'author', 'author_id', 'published_date',
'created_at', 'updated_at'
]
read_only_fields = ['created_at', 'updated_at']
# Nested serializers
class ReviewSerializer(serializers.ModelSerializer):
user = serializers.StringRelatedField()
class Meta:
model = Review
fields = ['id', 'user', 'rating', 'comment', 'created_at']
class BookDetailSerializer(serializers.ModelSerializer):
author = AuthorSerializer()
reviews = ReviewSerializer(many=True, read_only=True)
average_rating = serializers.SerializerMethodField()
class Meta:
model = Book
fields = [
'id', 'title', 'description', 'price',
'author', 'reviews', 'average_rating'
]
def get_average_rating(self, obj):
return obj.reviews.aggregate(
Avg('rating')
)['rating__avg'] or 0
# Custom validation
class BookSerializer(serializers.ModelSerializer):
class Meta:
model = Book
fields = '__all__'
def validate_price(self, value):
"""Validate single field"""
if value < 0:
raise serializers.ValidationError(
'Price must be positive'
)
return value
def validate(self, data):
"""Validate multiple fields"""
if data.get('published_date'):
if data['published_date'] > timezone.now().date():
raise serializers.ValidationError(
'Cannot publish in the future'
)
return data
# Custom fields
class ISBNField(serializers.Field):
def to_representation(self, value):
# Convert internal → API format
return value.replace('-', '')
def to_internal_value(self, data):
# Convert API format → internal
if len(data) != 13:
raise serializers.ValidationError(
'ISBN must be 13 digits'
)
return f'{data[:3]}-{data[3:12]}-{data[12]}'
class BookSerializer(serializers.ModelSerializer):
isbn = ISBNField()
class Meta:
model = Book
fields = ['id', 'title', 'isbn']
# Dynamic fields
class DynamicFieldsSerializer(serializers.ModelSerializer):
def __init__(self, *args, **kwargs):
fields = kwargs.pop('fields', None)
super().__init__(*args, **kwargs)
if fields:
allowed = set(fields)
existing = set(self.fields)
for field_name in existing - allowed:
self.fields.pop(field_name)
class BookSerializer(DynamicFieldsSerializer):
class Meta:
model = Book
fields = '__all__'
# Usage: BookSerializer(books, many=True, fields=['id', 'title'])
# Write-only fields for create/update
class UserSerializer(serializers.ModelSerializer):
password = serializers.CharField(write_only=True)
password_confirm = serializers.CharField(write_only=True)
class Meta:
model = User
fields = ['id', 'email', 'password', 'password_confirm']
def validate(self, data):
if data['password'] != data['password_confirm']:
raise serializers.ValidationError(
'Passwords do not match'
)
return data
def create(self, validated_data):
validated_data.pop('password_confirm')
user = User.objects.create_user(**validated_data)
return user
# SerializerMethodField
class BookSerializer(serializers.ModelSerializer):
is_new = serializers.SerializerMethodField()
category_names = serializers.SerializerMethodField()
class Meta:
model = Book
fields = ['id', 'title', 'is_new', 'category_names']
def get_is_new(self, obj):
return obj.published_date > timezone.now().date() - timedelta(days=30)
def get_category_names(self, obj):
return [cat.name for cat in obj.categories.all()]
# Context in serializers
class BookSerializer(serializers.ModelSerializer):
is_favorited = serializers.SerializerMethodField()
class Meta:
model = Book
fields = ['id', 'title', 'is_favorited']
def get_is_favorited(self, obj):
request = self.context.get('request')
if request and request.user.is_authenticated:
return obj.favorites.filter(user=request.user).exists()
return False
# Usage in view:
serializer = BookSerializer(
book,
context={'request': request}
)
ViewSets & Views
# ModelViewSet (full CRUD)
from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.response import Response
class BookViewSet(viewsets.ModelViewSet):
"""
Provides list, create, retrieve, update, destroy
"""
queryset = Book.objects.all()
serializer_class = BookSerializer
permission_classes = [IsAuthenticatedOrReadOnly]
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
filterset_fields = ['status', 'author']
search_fields = ['title', 'description']
ordering_fields = ['published_date', 'price']
ordering = ['-published_date']
def get_queryset(self):
"""Optimize queries"""
queryset = super().get_queryset()
return queryset.select_related('author').prefetch_related('categories')
def get_serializer_class(self):
"""Different serializers for different actions"""
if self.action == 'retrieve':
return BookDetailSerializer
elif self.action == 'create':
return BookCreateSerializer
return BookSerializer
def get_permissions(self):
"""Different permissions for different actions"""
if self.action in ['create', 'update', 'partial_update', 'destroy']:
return [IsAuthenticated()]
return [AllowAny()]
def perform_create(self, serializer):
"""Hook before saving"""
serializer.save(created_by=self.request.user)
def perform_update(self, serializer):
serializer.save(updated_by=self.request.user)
@action(detail=True, methods=['post'])
def publish(self, request, pk=None):
"""Custom action: /api/books/{id}/publish/"""
book = self.get_object()
book.status = 'published'
book.published_date = timezone.now()
book.save()
return Response({'status': 'book published'})
@action(detail=False, methods=['get'])
def bestsellers(self, request):
"""Custom list action: /api/books/bestsellers/"""
books = self.get_queryset().filter(
sales__gt=10000
).order_by('-sales')[:10]
serializer = self.get_serializer(books, many=True)
return Response(serializer.data)
# ReadOnlyModelViewSet
class AuthorViewSet(viewsets.ReadOnlyModelViewSet):
"""
Provides only list and retrieve
"""
queryset = Author.objects.all()
serializer_class = AuthorSerializer
# Generic API Views
from rest_framework import generics
class BookListView(generics.ListAPIView):
queryset = Book.objects.all()
serializer_class = BookSerializer
class BookCreateView(generics.CreateAPIView):
queryset = Book.objects.all()
serializer_class = BookSerializer
class BookDetailView(generics.RetrieveUpdateDestroyAPIView):
queryset = Book.objects.all()
serializer_class = BookSerializer
# APIView (lowest level)
from rest_framework.views import APIView
from rest_framework import status
class BookListCreateView(APIView):
def get(self, request):
books = Book.objects.all()
serializer = BookSerializer(books, many=True)
return Response(serializer.data)
def post(self, request):
serializer = BookSerializer(data=request.data)
if serializer.is_valid():
serializer.save()
return Response(
serializer.data,
status=status.HTTP_201_CREATED
)
return Response(
serializer.errors,
status=status.HTTP_400_BAD_REQUEST
)
# Function-based views
from rest_framework.decorators import api_view, permission_classes
@api_view(['GET', 'POST'])
@permission_classes([IsAuthenticatedOrReadOnly])
def book_list(request):
if request.method == 'GET':
books = Book.objects.all()
serializer = BookSerializer(books, many=True)
return Response(serializer.data)
elif request.method == 'POST':
serializer = BookSerializer(data=request.data)
if serializer.is_valid():
serializer.save()
return Response(
serializer.data,
status=status.HTTP_201_CREATED
)
return Response(
serializer.errors,
status=status.HTTP_400_BAD_REQUEST
)
# Pagination
from rest_framework.pagination import PageNumberPagination
class StandardResultsSetPagination(PageNumberPagination):
page_size = 20
page_size_query_param = 'page_size'
max_page_size = 100
class BookViewSet(viewsets.ModelViewSet):
queryset = Book.objects.all()
serializer_class = BookSerializer
pagination_class = StandardResultsSetPagination
# Filtering
from django_filters import rest_framework as filters
class BookFilter(filters.FilterSet):
min_price = filters.NumberFilter(field_name='price', lookup_expr='gte')
max_price = filters.NumberFilter(field_name='price', lookup_expr='lte')
title = filters.CharFilter(lookup_expr='icontains')
published_after = filters.DateFilter(
field_name='published_date',
lookup_expr='gte'
)
class Meta:
model = Book
fields = ['status', 'author', 'min_price', 'max_price']
class BookViewSet(viewsets.ModelViewSet):
queryset = Book.objects.all()
serializer_class = BookSerializer
filterset_class = BookFilter
Permissions & Throttling
# Built-in permissions
from rest_framework.permissions import (
AllowAny,
IsAuthenticated,
IsAuthenticatedOrReadOnly,
IsAdminUser,
)
class BookViewSet(viewsets.ModelViewSet):
queryset = Book.objects.all()
serializer_class = BookSerializer
def get_permissions(self):
if self.action in ['list', 'retrieve']:
return [AllowAny()]
elif self.action == 'create':
return [IsAuthenticated()]
else: # update, destroy
return [IsAdminUser()]
return super().get_permissions()
# Custom permissions
from rest_framework import permissions
class IsAuthorOrReadOnly(permissions.BasePermission):
"""
Object-level permission to only allow authors to edit
"""
def has_permission(self, request, view):
# Allow read operations to anyone
if request.method in permissions.SAFE_METHODS:
return True
# Write permissions require authentication
return request.user and request.user.is_authenticated
def has_object_permission(self, request, view, obj):
# Read permissions to anyone
if request.method in permissions.SAFE_METHODS:
return True
# Write permissions only to author
return obj.author == request.user
class BookViewSet(viewsets.ModelViewSet):
queryset = Book.objects.all()
serializer_class = BookSerializer
permission_classes = [IsAuthorOrReadOnly]
# Multiple custom permissions
class IsOwnerOrAdmin(permissions.BasePermission):
def has_object_permission(self, request, view, obj):
return (
request.user and
(obj.owner == request.user or request.user.is_staff)
)
class IsVerifiedUser(permissions.BasePermission):
def has_permission(self, request, view):
return (
request.user and
request.user.is_authenticated and
request.user.is_verified
)
class BookViewSet(viewsets.ModelViewSet):
permission_classes = [IsVerifiedUser, IsOwnerOrAdmin]
# Throttling
from rest_framework.throttling import (
AnonRateThrottle,
UserRateThrottle,
ScopedRateThrottle,
)
# settings.py
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_CLASSES': [
'rest_framework.throttling.AnonRateThrottle',
'rest_framework.throttling.UserRateThrottle',
],
'DEFAULT_THROTTLE_RATES': {
'anon': '100/day',
'user': '1000/day',
'burst': '60/min',
'sustained': '1000/day',
},
}
# Custom throttle
from rest_framework.throttling import UserRateThrottle
class BurstRateThrottle(UserRateThrottle):
scope = 'burst'
class SustainedRateThrottle(UserRateThrottle):
scope = 'sustained'
class BookViewSet(viewsets.ModelViewSet):
throttle_classes = [BurstRateThrottle, SustainedRateThrottle]
# Per-view throttling
class BookViewSet(viewsets.ModelViewSet):
queryset = Book.objects.all()
serializer_class = BookSerializer
def get_throttles(self):
if self.action == 'create':
# More restrictive for creation
return [UserRateThrottle()]
return super().get_throttles()
@action(detail=False, methods=['post'], throttle_classes=[ScopedRateThrottle])
def bulk_create(self, request):
pass
Async Views & ASGI
Async Views (Django 4.1+)
# Async function-based view
import asyncio
from django.http import JsonResponse
async def async_view(request):
# Async database query
books = await Book.objects.filter(status='published').aall()
# Async external API call
async with aiohttp.ClientSession() as session:
async with session.get('https://api.example.com/data') as resp:
data = await resp.json()
return JsonResponse({
'books_count': len(books),
'external_data': data
})
# Async class-based view
from django.views import View
class AsyncBookListView(View):
async def get(self, request):
books = []
async for book in Book.objects.filter(status='published'):
books.append({
'id': book.id,
'title': book.title
})
return JsonResponse({'books': books})
# Async ORM operations
# QuerySet async methods
books = await Book.objects.filter(status='published').aall()
book = await Book.objects.aget(id=1)
exists = await Book.objects.filter(id=1).aexists()
count = await Book.objects.filter(status='published').acount()
# Create
book = await Book.objects.acreate(
title='New Book',
author_id=1
)
# Update
book.title = 'Updated Title'
await book.asave()
# Delete
await book.adelete()
# Aggregate
from django.db.models import Avg
avg_price = await Book.objects.aaggregate(Avg('price'))
# Async iteration
async for book in Book.objects.filter(status='published'):
print(book.title)
# Async transactions
from django.db import transaction
async def create_book_with_reviews(book_data, review_data):
async with transaction.atomic():
book = await Book.objects.acreate(**book_data)
for review in review_data:
await Review.objects.acreate(
book=book,
**review
)
return book
# Parallel async operations
async def get_dashboard_data(user_id):
# Run multiple queries in parallel
books, reviews, orders = await asyncio.gather(
Book.objects.filter(author_id=user_id).acount(),
Review.objects.filter(user_id=user_id).acount(),
Order.objects.filter(user_id=user_id).acount()
)
return {
'books': books,
'reviews': reviews,
'orders': orders
}
# Async middleware
class AsyncMiddleware:
async_capable = True
sync_capable = False
def __init__(self, get_response):
self.get_response = get_response
async def __call__(self, request):
# Async preprocessing
await some_async_operation()
response = await self.get_response(request)
# Async postprocessing
await another_async_operation()
return response
# Mix async and sync code
from asgiref.sync import sync_to_async, async_to_sync
# Call sync function from async context
@sync_to_async
def sync_function():
return Book.objects.all()
async def async_view(request):
books = await sync_function()
return JsonResponse({'count': len(books)})
# Call async function from sync context
async def async_function():
return await Book.objects.aall()
def sync_view(request):
books = async_to_sync(async_function)()
return JsonResponse({'count': len(books)})
# Async with caching
from django.core.cache import cache
from asgiref.sync import sync_to_async
async def get_cached_books():
# Cache operations are sync, wrap them
cache_key = 'books:published'
@sync_to_async
def get_cache():
return cache.get(cache_key)
@sync_to_async
def set_cache(data):
cache.set(cache_key, data, timeout=300)
books = await get_cache()
if books is None:
books = await Book.objects.filter(
status='published'
).aall()
await set_cache(list(books))
return books
ASGI Configuration
# asgi.py
import os
from django.core.asgi import get_asgi_application
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.production')
application = get_asgi_application()
# Running with Uvicorn
# Install
pip install uvicorn[standard]
# Development
uvicorn config.asgi:application --reload
# Production
uvicorn config.asgi:application \
--host 0.0.0.0 \
--port 8000 \
--workers 4 \
--limit-concurrency 1000 \
--timeout-keep-alive 5
# Running with Daphne
# Install
pip install daphne
# Development
daphne -b 0.0.0.0 -p 8000 config.asgi:application
# Production
daphne -b 0.0.0.0 -p 8000 \
--access-log /var/log/daphne/access.log \
config.asgi:application
# Systemd service file
# /etc/systemd/system/myapp.service
[Unit]
Description=My Django ASGI Application
After=network.target
[Service]
Type=notify
User=www-data
Group=www-data
WorkingDirectory=/var/www/myapp
Environment="PATH=/var/www/myapp/venv/bin"
ExecStart=/var/www/myapp/venv/bin/uvicorn \
config.asgi:application \
--host 0.0.0.0 \
--port 8000 \
--workers 4
[Install]
WantedBy=multi-user.target
# Nginx configuration for ASGI
upstream django_asgi {
server 127.0.0.1:8000;
}
server {
listen 80;
server_name yourdomain.com;
location / {
proxy_pass http://django_asgi;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_read_timeout 86400;
}
location /static/ {
alias /var/www/myapp/static/;
}
location /media/ {
alias /var/www/myapp/media/;
}
}
# Docker configuration
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "config.asgi:application", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
version: '3.8'
services:
web:
build: .
command: uvicorn config.asgi:application --host 0.0.0.0 --port 8000 --workers 4
volumes:
- .:/app
ports:
- "8000:8000"
environment:
- DEBUG=False
- DATABASE_URL=postgresql://user:pass@db:5432/mydb
- REDIS_URL=redis://redis:6379/0
depends_on:
- db
- redis
db:
image: postgres:15
volumes:
- postgres_data:/var/lib/postgresql/data
environment:
- POSTGRES_DB=mydb
- POSTGRES_USER=user
- POSTGRES_PASSWORD=pass
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
volumes:
postgres_data:
redis_data:
Background Tasks with Celery
⚠️ Content Note
This section covers Celery configuration, task definition, periodic tasks, task monitoring, and error handling. For detailed Celery patterns, see the dedicated "Celery & Task Queues" comprehensive sheet. Quick Celery Setup: 1. Install: pip install celery redis 2. Configure broker: CELERY_BROKER_URL = 'redis://localhost:6379/0' 3. Define tasks with @shared_task decorator 4. Run worker: celery -A config worker -l info 5. Run beat for periodic tasks: celery -A config beat -l info Related Sheets: → Celery & Task Queues (comprehensive patterns) → Redis (broker configuration) → Python Advanced (async patterns)
Resources & Learning Path
Learning Progression
Phase 1: Django Fundamentals (2-4 weeks) □ Django tutorial (polls app) □ Models and QuerySets □ Views and URL routing □ Templates and forms □ Admin interface □ Authentication basics Phase 2: Intermediate Django (4-6 weeks) □ Class-based views □ Model relationships and queries □ Django REST Framework basics □ File uploads and media □ Middleware basics □ Testing with pytest-django Phase 3: Production Django (6-8 weeks) □ ORM optimization techniques □ Caching strategies □ Security best practices □ Celery for background tasks □ Database configuration and tuning □ Monitoring and logging □ Deployment strategies Phase 4: Advanced Topics (Ongoing) □ Custom management commands □ Database migrations at scale □ Multi-tenancy patterns □ GraphQL with Graphene □ Async views and ASGI □ Microservices architecture □ Performance optimization □ Scalability patterns
Essential Resources
Official Documentation • Django Docs: https://docs.djangoproject.com/ • DRF Docs: https://www.django-rest-framework.org/ • Django Deployment Checklist • Django Security Overview Books • Two Scoops of Django (best practices) • High Performance Django • Django for Professionals • Django for APIs Courses • Django for Everybody (Coursera) • Ultimate Django Series (Code with Mosh) • Django REST Framework (TestDriven.io) Community • Django Forum: https://forum.djangoproject.com/ • r/django subreddit • Django Discord server • DjangoGirls community Tools • Django Debug Toolbar • django-extensions • django-silk (profiling) • django-pytest • pre-commit hooks
Related Comprehensive Sheets
Python Ecosystem → Python Advanced (async, decorators, performance) → FastAPI Advanced (alternative framework) → Flask Production (microservices approach) → Celery & Task Queues (background jobs) → PyTorch & Deep Learning (ML integration) Database & Caching → PostgreSQL Production → Redis Advanced → Database Design Patterns → SQL Query Optimization DevOps & Deployment → Docker & Kubernetes → CI/CD Pipelines → AWS Services → Monitoring & Observability → Nginx & Load Balancing Frontend Integration → React Advanced → Vue.js 3 → REST API Design → GraphQL Patterns → WebSockets Testing & Quality → Python Testing (pytest, mocking) → Load Testing → Security Testing → Test Automation
Pro Tips Summary
Performance ✓ Always use select_related() and prefetch_related() ✓ Add database indexes to frequently queried fields ✓ Use only() and defer() to reduce data transfer ✓ Implement caching at multiple levels ✓ Use iterator() for large querysets ✓ Monitor query counts in development ✓ Use bulk operations for batch updates Security ✓ Never set DEBUG=True in production ✓ Use environment variables for secrets ✓ Enable all security middleware ✓ Implement rate limiting ✓ Validate all user input ✓ Use parameterized queries ✓ Keep Django and dependencies updated Code Organization ✓ Separate business logic into services ✓ Use managers for common queries ✓ Keep views thin, models fat ✓ Create reusable mixins ✓ Organize settings by environment ✓ Document complex logic ✓ Follow Django conventions Testing ✓ Write tests before deploying ✓ Test edge cases and error conditions ✓ Use factories for test data ✓ Mock external services ✓ Test permissions and authorization ✓ Measure code coverage ✓ Run tests in CI/CD pipeline Deployment ✓ Use ASGI for async support ✓ Run migrations before deployment ✓ Collect static files properly ✓ Set up proper logging ✓ Use a process manager (systemd) ✓ Implement health checks ✓ Monitor application metrics ✓ Have a rollback plan