- Django ORM : maîtriser les requêtes optimisées

QuerySet : les bases et la paresse

Les QuerySets Django sont lazily evaluated (évaluation paresseuse) : ils ne génèrent pas de requête SQL tant que vous ne consommez pas leur résultat. C'est un principe fondamental à comprendre pour optimiser votre code.

Évaluation paresseuse vs eager

# Pas de requête SQL — construction du QuerySet seulement
users = User.objects.filter(is_active=True)
users = users.filter(age__gte=18)
users = users.order_by('-created_at')

# La requête SQL est exécutée ICI (évaluation)
for user in users:           # itération → SQL
    print(user.name)

# Ou lors d'un slice
first_10 = users[:10]        # LIMIT 10 → SQL
count = users.count()        # SELECT COUNT(*) → SQL
exists = users.exists()      # SELECT 1 WHERE ... LIMIT 1 → SQL
user = users.first()         # ORDER BY ... LIMIT 1 → SQL
user_list = list(users)      # force évaluation → SQL

Lookups de filtres courants

# Comparaisons
User.objects.filter(age=25)           # exact
User.objects.filter(age__gte=18)      # >=
User.objects.filter(age__lte=65)      # <=
User.objects.filter(age__gt=18)       # >
User.objects.filter(age__lt=65)       # <
User.objects.filter(age__range=(18, 65))  # BETWEEN

# Texte
User.objects.filter(name__exact='Alice')     # case-sensitive
User.objects.filter(name__iexact='alice')    # case-insensitive
User.objects.filter(name__contains='ali')   # LIKE %ali%
User.objects.filter(name__icontains='ali')  # LIKE %ali% (insensible)
User.objects.filter(name__startswith='Al')  # LIKE Al%
User.objects.filter(name__endswith='ce')    # LIKE %ce

# Collections
User.objects.filter(status__in=['active', 'pending'])   # IN (...)
User.objects.exclude(status__in=['banned', 'deleted'])  # NOT IN (...)

# Null
User.objects.filter(deleted_at__isnull=True)   # IS NULL
User.objects.filter(deleted_at__isnull=False)  # IS NOT NULL

# Dates
Article.objects.filter(created_at__date=date.today())
Article.objects.filter(created_at__year=2024)
Article.objects.filter(created_at__month=4)
Article.objects.filter(created_at__week_day=2)  # 1=Sunday, 2=Monday

Relations entre modèles

Django ORM gère trois types de relations : ForeignKey (1:N), ManyToManyField (N:N), et OneToOneField (1:1). La traversée des relations s'effectue avec la notation double underscore __.

# models.py
from django.db import models

class Author(models.Model):
    name = models.CharField(max_length=100)
    email = models.EmailField(unique=True)
    bio = models.TextField(blank=True)

    class Meta:
        ordering = ['name']

class Category(models.Model):
    name = models.CharField(max_length=50, unique=True)
    slug = models.SlugField(unique=True)

class Book(models.Model):
    title = models.CharField(max_length=200)
    author = models.ForeignKey(Author, on_delete=models.CASCADE, related_name='books')
    categories = models.ManyToManyField(Category, related_name='books', blank=True)
    publication_date = models.DateField()
    is_published = models.BooleanField(default=False)
    price = models.DecimalField(max_digits=8, decimal_places=2, default=0)

class Review(models.Model):
    book = models.ForeignKey(Book, on_delete=models.CASCADE, related_name='reviews')
    rating = models.PositiveSmallIntegerField()  # 1-5
    comment = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)

Traversée de relations dans les filtres

# Filtrer les livres d'un auteur par son email
books = Book.objects.filter(author__email='alice@example.com')

# Livres publiés d'une catégorie spécifique
books = Book.objects.filter(
    categories__slug='python',
    is_published=True
)

# Auteurs dont un livre a reçu une note >= 4
authors = Author.objects.filter(books__reviews__rating__gte=4).distinct()

# Livres sans aucune review (relation inverse vide)
books_no_review = Book.objects.filter(reviews__isnull=True)

# Livres avec au moins 5 reviews
from django.db.models import Count
books = Book.objects.annotate(
    review_count=Count('reviews')
).filter(review_count__gte=5)

Éviter le problème N+1

Le problème N+1 est la source de lenteur la plus fréquente avec les ORM. Il se produit quand une boucle déclenche une requête supplémentaire pour chaque enregistrement. Django fournit select_related() et prefetch_related() pour le résoudre.

select_related() pour les ForeignKey (JOIN SQL)

# ❌ MAUVAIS : 1 requête principale + 1 par livre = N+1
books = Book.objects.all()[:20]
for book in books:
    print(book.author.name)  # Requête SQL à chaque itération!

# ✅ BON : 1 seule requête avec JOIN
books = Book.objects.select_related('author').all()[:20]
for book in books:
    print(book.author.name)  # Aucune requête supplémentaire

# Chaîner les relations
books = Book.objects.select_related('author', 'author__profile')

# Toutes les FK et OneToOne (attention aux performances si beaucoup de relations)
books = Book.objects.select_related()

prefetch_related() pour les ManyToMany

# ❌ MAUVAIS : 1 requête + N requêtes pour les catégories
books = Book.objects.all()
for book in books:
    cats = ', '.join(c.name for c in book.categories.all())  # N requêtes!

# ✅ BON : 2 requêtes (une pour les livres, une pour toutes les catégories)
books = Book.objects.prefetch_related('categories')
for book in books:
    cats = ', '.join(c.name for c in book.categories.all())  # Cache utilisé

# Combiner select_related et prefetch_related
books = Book.objects.select_related('author') \
                    .prefetch_related('categories', 'reviews')

# Prefetch avec queryset personnalisé (Prefetch object)
from django.db.models import Prefetch

recent_reviews = Review.objects.filter(rating__gte=4).order_by('-created_at')[:3]
books = Book.objects.prefetch_related(
    Prefetch('reviews', queryset=recent_reviews, to_attr='top_reviews')
)
for book in books:
    # book.top_reviews est une liste Python (pas un QuerySet)
    for review in book.top_reviews:
        print(review.comment)

Q() objects : requêtes complexes

Les objets Q() permettent de combiner des conditions avec des opérateurs logiques | (OR), & (AND) et ~ (NOT), ce qu'il est impossible de faire avec les seuls arguments de filter().

from django.db.models import Q

# OR : livres de Python OU publiés en 2024
books = Book.objects.filter(
    Q(categories__slug='python') | Q(publication_date__year=2024)
)

# AND combiné avec OR
books = Book.objects.filter(
    Q(is_published=True) &
    (Q(price__lt=20) | Q(categories__slug='free'))
)

# NOT : livres pas de la catégorie 'deprecated'
books = Book.objects.filter(~Q(categories__slug='deprecated'))

# Combinaison complexe : recherche full-text sur titre OU auteur
def search_books(query: str):
    return Book.objects.filter(
        Q(title__icontains=query) |
        Q(author__name__icontains=query) |
        Q(categories__name__icontains=query)
    ).distinct()

# Q() dans un exclude()
active_books = Book.objects.exclude(
    Q(is_published=False) | Q(price__lte=0)
)

Conditions dynamiques avec Q()

def filter_books(title=None, author=None, min_price=None, category=None):
    """Filtre dynamique : n'applique que les conditions non nulles."""
    q = Q()  # Q() vide = aucun filtre

    if title:
        q &= Q(title__icontains=title)
    if author:
        q &= Q(author__name__icontains=author)
    if min_price is not None:
        q &= Q(price__gte=min_price)
    if category:
        q &= Q(categories__slug=category)

    return Book.objects.filter(q).distinct()

# Usage
results = filter_books(title='django', category='python')

F() expressions : opérations en base

Les expressions F() référencent un champ de la base de données et permettent d'effectuer des opérations directement en SQL, sans charger les objets en mémoire Python.

from django.db.models import F

# ❌ MAUVAIS : charge tous les objets en mémoire
products = Product.objects.filter(stock__gt=0)
for product in products:
    product.views_count += 1  # Problème race condition!
    product.save()

# ✅ BON : une seule requête UPDATE, sans race condition
Product.objects.filter(stock__gt=0).update(views_count=F('views_count') + 1)

# Comparer deux champs d'un même modèle
# Livres dont le prix barré est supérieur au prix normal
discounted = Book.objects.filter(original_price__gt=F('price'))

# Tri basé sur une expression
from django.db.models import ExpressionWrapper, FloatField
books = Book.objects.annotate(
    discount_pct=ExpressionWrapper(
        (F('original_price') - F('price')) / F('original_price') * 100,
        output_field=FloatField()
    )
).order_by('-discount_pct')

# Incrémenter une date
from datetime import timedelta
from django.db.models import DurationValue
Article.objects.update(
    expires_at=F('created_at') + timedelta(days=30)
)

# Référence à une relation ForeignKey
# Filtrer les articles dont le prix est inférieur à la moyenne de leur catégorie
from django.db.models import Avg, OuterRef, Subquery
avg_price_subquery = Book.objects.filter(
    categories=OuterRef('categories')
).values('categories').annotate(avg=Avg('price')).values('avg')

cheap_books = Book.objects.annotate(
    category_avg=Subquery(avg_price_subquery)
).filter(price__lt=F('category_avg'))

Agrégation et annotation

aggregate() retourne un dictionnaire avec des statistiques globales. annotate() ajoute une colonne calculée à chaque enregistrement du QuerySet.

from django.db.models import Count, Sum, Avg, Max, Min, StdDev, Variance

# aggregate() : statistiques globales (retourne un dict)
stats = Book.objects.aggregate(
    total=Count('id'),
    avg_price=Avg('price'),
    max_price=Max('price'),
    min_price=Min('price'),
    total_revenue=Sum('price')
)
# {'total': 150, 'avg_price': 29.99, 'max_price': 99.99, ...}

# annotate() : statistiques par enregistrement
authors = Author.objects.annotate(
    book_count=Count('books'),
    avg_book_price=Avg('books__price'),
    total_revenue=Sum('books__price'),
    best_rating=Max('books__reviews__rating')
).filter(book_count__gte=3).order_by('-book_count')

for author in authors:
    print(f"{author.name}: {author.book_count} livres, moy {author.avg_book_price:.2f}€")

# Annotation conditionnelle avec Case/When
from django.db.models import Case, When, Value, IntegerField

books = Book.objects.annotate(
    price_tier=Case(
        When(price__lte=10, then=Value('budget')),
        When(price__lte=30, then=Value('mid')),
        When(price__gt=30, then=Value('premium')),
        default=Value('unknown'),
        output_field=models.CharField()
    )
)

# Group by avec values() + annotate()
from django.db.models.functions import TruncMonth

monthly_sales = Book.objects.filter(is_published=True) \
    .annotate(month=TruncMonth('publication_date')) \
    .values('month') \
    .annotate(count=Count('id'), revenue=Sum('price')) \
    .order_by('month')

for row in monthly_sales:
    print(f"{row['month'].strftime('%B %Y')}: {row['count']} livres, {row['revenue']:.2f}€")

Opérations en masse (bulk)

Pour insérer ou mettre à jour des milliers d'enregistrements, les méthodes bulk_create() et bulk_update() sont indispensables — elles réduisent drastiquement le nombre de requêtes SQL.

# ❌ MAUVAIS : 1000 INSERT individuels
for i in range(1000):
    Tag.objects.create(name=f'tag-{i}')

# ✅ BON : 1 seul INSERT avec bulk_create
tags = [Tag(name=f'tag-{i}') for i in range(1000)]
Tag.objects.bulk_create(tags, batch_size=100)  # INSERT par lots de 100

# bulk_create avec ignore_conflicts (pas d'erreur si doublon)
Tag.objects.bulk_create(tags, ignore_conflicts=True)

# bulk_create et récupérer les IDs créés (Django 4.1+)
created_tags = Tag.objects.bulk_create(tags, batch_size=100)
ids = [t.id for t in created_tags]  # IDs disponibles post-insertion

# ❌ MAUVAIS : N updates individuels
books = Book.objects.filter(is_published=False)
for book in books:
    book.is_published = True
    book.save()  # 1 UPDATE par livre!

# ✅ BON : 1 seul UPDATE
Book.objects.filter(is_published=False).update(is_published=True)

# bulk_update : mettre à jour différentes valeurs par objet
books = list(Book.objects.filter(author__id=1))
for book in books:
    book.price = round(book.price * 1.10, 2)  # +10%

Book.objects.bulk_update(books, ['price'], batch_size=100)

defer() et only() : champs partiels

Quand vous n'avez besoin que de certains champs d'un modèle, only() et defer() permettent de limiter le SELECT pour économiser bande passante et mémoire.

# only() : charger UNIQUEMENT les champs listés
# SELECT id, title, author_id FROM book
books = Book.objects.only('id', 'title', 'author_id')
for book in books:
    print(book.title)  # OK : champ chargé
    # print(book.price)  # ⚠️ Déclenche une requête supplémentaire!

# defer() : charger tout SAUF les champs listés
# Utile pour exclure les gros champs (TextField, JSONField)
books = Book.objects.defer('description', 'cover_image_data')
for book in books:
    print(book.title)  # OK
    print(book.price)  # OK
    # book.description déclenchera une requête si accédé

# values() : dictionnaires au lieu d'objets
books = Book.objects.values('id', 'title', 'author__name')
# [{'id': 1, 'title': '...', 'author__name': 'Alice'}, ...]

# values_list() : tuples (ou valeurs simples)
titles = Book.objects.values_list('title', flat=True)  # QuerySet de strings
# ['Titre 1', 'Titre 2', ...]

pairs = Book.objects.values_list('id', 'title')
# [(1, 'Titre 1'), (2, 'Titre 2'), ...]

# Usage réel : alimenter un select HTML
author_choices = Author.objects.values_list('id', 'name').order_by('name')
# Prêt pour un champ choices Django

Transactions et atomicité

Django gère les transactions SQL avec atomic(). Toute exception dans un bloc atomic() déclenche un rollback automatique.

from django.db import transaction

# Context manager : toute exception = rollback
def transfer_credits(from_user_id, to_user_id, amount):
    with transaction.atomic():
        from_user = User.objects.select_for_update().get(id=from_user_id)
        to_user = User.objects.select_for_update().get(id=to_user_id)

        if from_user.credits < amount:
            raise ValueError("Crédits insuffisants")

        from_user.credits = F('credits') - amount
        to_user.credits = F('credits') + amount
        from_user.save(update_fields=['credits'])
        to_user.save(update_fields=['credits'])
        # Commit automatique à la sortie du 'with'

# Décorateur @atomic sur une fonction
@transaction.atomic
def create_order_with_items(user_id, items):
    order = Order.objects.create(user_id=user_id)
    OrderItem.objects.bulk_create([
        OrderItem(order=order, product_id=item['id'], qty=item['qty'])
        for item in items
    ])
    return order

# Savepoints (rollback partiel)
def batch_import(records):
    with transaction.atomic():
        for record in records:
            try:
                sid = transaction.savepoint()  # Crée un point de sauvegarde
                process_record(record)
                transaction.savepoint_commit(sid)
            except Exception as e:
                transaction.savepoint_rollback(sid)  # Annule seulement ce record
                log_error(record, e)  # Continue avec les suivants

# select_for_update() : verrou de ligne (évite race conditions)
with transaction.atomic():
    product = Product.objects.select_for_update().get(id=product_id)
    if product.stock > 0:
        product.stock = F('stock') - 1
        product.save(update_fields=['stock'])

Index et mesure des performances

Créer des index efficaces

class Article(models.Model):
    title = models.CharField(max_length=200, db_index=True)  # Index simple
    slug = models.SlugField(unique=True)  # Index unique automatique
    author = models.ForeignKey(Author, on_delete=models.CASCADE)
    status = models.CharField(max_length=20, default='draft')
    published_at = models.DateTimeField(null=True, blank=True, db_index=True)
    views = models.PositiveIntegerField(default=0)

    class Meta:
        indexes = [
            # Index composé pour les requêtes status + published_at
            models.Index(fields=['status', '-published_at'],
                         name='article_status_pub_idx'),
            # Index partiel (PostgreSQL uniquement)
            models.Index(
                fields=['published_at'],
                condition=models.Q(status='published'),
                name='article_published_idx'
            ),
        ]
        # Contrainte unique composée
        constraints = [
            models.UniqueConstraint(
                fields=['author', 'slug'],
                name='unique_author_slug'
            )
        ]

Analyser les requêtes avec django-debug-toolbar et connection.queries

from django.db import connection, reset_queries
from django.conf import settings

def log_queries(func):
    """Décorateur pour auditer les requêtes SQL d'une vue."""
    def wrapper(*args, **kwargs):
        if settings.DEBUG:
            reset_queries()
            result = func(*args, **kwargs)
            queries = connection.queries
            total_time = sum(float(q['time']) for q in queries)
            print(f"Requêtes SQL: {len(queries)} | Temps total: {total_time:.3f}s")
            for i, q in enumerate(queries, 1):
                print(f"  [{i}] {q['time']}s: {q['sql'][:120]}")
            return result
        return func(*args, **kwargs)
    return wrapper

@log_queries
def get_books_api():
    return list(
        Book.objects.select_related('author')
                    .prefetch_related('categories')
                    .filter(is_published=True)
                    .order_by('-publication_date')[:20]
    )

# explain() : voir le plan d'exécution SQL
qs = Book.objects.filter(author__email='alice@example.com')
print(qs.explain())
print(qs.explain(analyze=True))  # Exécute réellement la requête (PostgreSQL)

Managers et QuerySets personnalisés

Les managers personnalisés permettent d'encapsuler la logique de requête au niveau du modèle, améliorant la réutilisabilité et la lisibilité.

from django.db import models
from django.utils import timezone

class ArticleQuerySet(models.QuerySet):
    def published(self):
        return self.filter(
            status='published',
            published_at__lte=timezone.now()
        )

    def by_author(self, author_id):
        return self.filter(author_id=author_id)

    def popular(self, threshold=1000):
        return self.filter(views__gte=threshold).order_by('-views')

    def with_stats(self):
        return self.annotate(
            comment_count=Count('comments'),
            avg_rating=Avg('reviews__rating')
        )

class ArticleManager(models.Manager):
    def get_queryset(self):
        return ArticleQuerySet(self.model, using=self._db)

    def published(self):
        return self.get_queryset().published()

    def trending(self):
        return self.get_queryset().published().popular(threshold=500) \
                   .order_by('-published_at')[:10]

class Article(models.Model):
    objects = ArticleManager()  # Remplace le manager par défaut

    status = models.CharField(max_length=20, default='draft')
    published_at = models.DateTimeField(null=True)
    views = models.PositiveIntegerField(default=0)
    # ...

# Usage propre et expressif
trending = Article.objects.trending()
user_articles = Article.objects.published().by_author(user.id).with_stats()

SQL brut quand ORM ne suffit pas

Certaines requêtes très complexes (fenêtres analytiques, requêtes récursives CTE, UPSERT) se prêtent mal à l'ORM. Django offre plusieurs niveaux d'accès au SQL brut.

from django.db import connection

# raw() : QuerySet avec SQL personnalisé (retourne des instances du modèle)
books = Book.objects.raw("""
    SELECT b.*, a.name as author_name
    FROM books_book b
    JOIN books_author a ON b.author_id = a.id
    WHERE b.is_published = TRUE
    ORDER BY b.publication_date DESC
    LIMIT 20
""")
for book in books:
    print(book.title, book.author_name)

# connection.execute() : SQL pur sans modèle
with connection.cursor() as cursor:
    cursor.execute("""
        SELECT
            DATE_TRUNC('month', published_at) as month,
            COUNT(*) as article_count,
            AVG(views) as avg_views
        FROM articles_article
        WHERE status = %s
        GROUP BY 1
        ORDER BY 1 DESC
    """, ['published'])

    columns = [col[0] for col in cursor.description]
    rows = cursor.fetchall()
    results = [dict(zip(columns, row)) for row in rows]

# UPSERT (INSERT OR UPDATE) avec on_conflict_do_update
# Django 4.1+ via update_or_create() ou via SQL brut
Article.objects.update_or_create(
    slug='mon-article',
    defaults={'title': 'Nouveau titre', 'views': F('views') + 1}
)

# PostgreSQL : bulk_create avec update_conflicts
Article.objects.bulk_create(
    articles,
    update_conflicts=True,
    unique_fields=['slug'],
    update_fields=['title', 'updated_at']
)

• Utilisez select_related() systématiquement pour les ForeignKey dans les boucles
• Vérifiez le nombre de requêtes avec connection.queries en développement
• Préférez bulk_create()/bulk_update() pour les insertions massives
• Utilisez F() pour les incréments atomiques sans race condition
• Encapsulez la logique de requête dans des managers/QuerySets personnalisés
• Ajoutez des index sur les champs utilisés dans filter() et order_by()
• Utilisez values() ou only() pour éviter de charger des gros champs inutilement
• Pensez à select_for_update() pour les opérations concurrentes critiques

Conclusion

Maîtriser Django ORM va bien au-delà du simple filter(). Les techniques avancées — select_related()/prefetch_related(), Q(), F(), annotate(), bulk operations, et transactions atomiques — sont celles qui font la différence entre une application qui rame à 1000 enregistrements et une qui tient à 10 millions.