# Vulnerable: N+1 query problem in Django
from django.shortcuts import render
from .models import Author, Book
# Model definitions
class Author(models.Model):
name = models.CharField(max_length=100)
email = models.EmailField()
class Book(models.Model):
title = models.CharField(max_length=200)
author = models.ForeignKey(Author, on_delete=models.CASCADE)
publication_date = models.DateField()
# Problematic view - N+1 queries
def book_list_view(request):
books = Book.objects.all() # 1 query
book_data = []
for book in books: # N additional queries (one per book)
book_data.append({
'title': book.title,
'author_name': book.author.name, # Triggers additional query
'author_email': book.author.email, # Same author, but still hits DB
'publication_date': book.publication_date
})
return render(request, 'books/list.html', {'books': book_data})
# Another problematic pattern
def author_books_view(request):
authors = Author.objects.all() # 1 query
author_data = []
for author in authors:
# Each access to author.book_set triggers a new query
book_count = author.book_set.count() # N queries
recent_books = author.book_set.order_by('-publication_date')[:3] # N more queries
author_data.append({
'name': author.name,
'book_count': book_count,
'recent_books': list(recent_books)
})
return author_data
# Template that causes N+1 queries
# books/list.html
# {% for book in books %}
#
#
{{ book.title }}
#
Author: {{ book.author.name }}
#
Email: {{ book.author.email }}
#
# {% endfor %}
# API endpoint with N+1 problem
from rest_framework.views import APIView
from rest_framework.response import Response
class BookAPIView(APIView):
def get(self, request):
books = Book.objects.all()
book_list = []
for book in books:
# Each iteration causes additional database queries
book_list.append({
'id': book.id,
'title': book.title,
'author': {
'id': book.author.id, # DB query
'name': book.author.name, # Same query repeated
'email': book.author.email
},
'publication_date': book.publication_date
})
return Response(book_list)
# Secure: Optimized Django queries to avoid N+1 problems
from django.shortcuts import render
from django.db import models
from .models import Author, Book
# Optimized view using select_related
def book_list_view_optimized(request):
# Single query with JOIN to fetch related author data
books = Book.objects.select_related('author').all()
book_data = []
for book in books:
book_data.append({
'title': book.title,
'author_name': book.author.name, # No additional query
'author_email': book.author.email, # No additional query
'publication_date': book.publication_date
})
return render(request, 'books/list.html', {'books': book_data})
# Optimized reverse foreign key access
def author_books_view_optimized(request):
# Use prefetch_related for reverse foreign key relationships
authors = Author.objects.prefetch_related(
'book_set' # Prefetch all related books
).all()
author_data = []
for author in authors:
# These operations use the prefetched data (no additional queries)
books = author.book_set.all()
book_count = len(books)
recent_books = sorted(books, key=lambda x: x.publication_date, reverse=True)[:3]
author_data.append({
'name': author.name,
'book_count': book_count,
'recent_books': [
{
'title': book.title,
'publication_date': book.publication_date
} for book in recent_books
]
})
return author_data
# Advanced optimization with custom prefetch
from django.db.models import Prefetch
def author_recent_books_view(request):
# Custom prefetch with filtering and ordering
recent_books_prefetch = Prefetch(
'book_set',
queryset=Book.objects.filter(
publication_date__year__gte=2020
).order_by('-publication_date'),
to_attr='recent_books'
)
authors = Author.objects.prefetch_related(recent_books_prefetch).all()
author_data = []
for author in authors:
# Uses the custom prefetched data
author_data.append({
'name': author.name,
'recent_books': [
{
'title': book.title,
'publication_date': book.publication_date
} for book in author.recent_books # Custom prefetched attribute
]
})
return author_data
# Optimized API endpoint
from rest_framework.views import APIView
from rest_framework.response import Response
from django.db.models import Count
class OptimizedBookAPIView(APIView):
def get(self, request):
# Single query with JOIN and aggregation
books = Book.objects.select_related('author').annotate(
author_book_count=Count('author__book')
).all()
book_list = []
for book in books:
# All data available from the single optimized query
book_list.append({
'id': book.id,
'title': book.title,
'author': {
'id': book.author.id,
'name': book.author.name,
'email': book.author.email,
'total_books': book.author_book_count
},
'publication_date': book.publication_date
})
return Response(book_list)
# Query optimization with only() and defer()
def book_list_minimal_view(request):
# Only fetch necessary fields
books = Book.objects.select_related('author').only(
'title',
'publication_date',
'author__name'
).all()
book_data = []
for book in books:
book_data.append({
'title': book.title,
'author_name': book.author.name,
'publication_date': book.publication_date
})
return render(request, 'books/minimal_list.html', {'books': book_data})
# Performance monitoring decorator
from django.db import connection
from functools import wraps
import time
def monitor_queries(func):
@wraps(func)
def wrapper(*args, **kwargs):
# Reset query log
connection.queries_log.clear()
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
query_count = len(connection.queries)
execution_time = end_time - start_time
# Log performance metrics
print(f"Function: {func.__name__}")
print(f"Queries executed: {query_count}")
print(f"Execution time: {execution_time:.4f} seconds")
# Alert if too many queries
if query_count > 10:
print(f"WARNING: High query count ({query_count}) detected!")
return result
return wrapper
# Usage with monitoring
@monitor_queries
def monitored_book_view(request):
return book_list_view_optimized(request)
# Caching for frequently accessed data
from django.core.cache import cache
from django.conf import settings
def cached_author_list_view(request):
cache_key = 'author_list_with_books'
cached_data = cache.get(cache_key)
if cached_data is None:
# Build optimized query only when cache miss
authors = Author.objects.prefetch_related('book_set').all()
author_data = []
for author in authors:
author_data.append({
'name': author.name,
'book_count': author.book_set.count(),
'books': [
{
'title': book.title,
'publication_date': book.publication_date.isoformat()
} for book in author.book_set.all()
]
})
# Cache for 15 minutes
cache.set(cache_key, author_data, 900)
cached_data = author_data
return render(request, 'authors/cached_list.html', {'authors': cached_data})