Django ORM Patterns

Master Django ORM for building efficient, scalable database-driven applications with complex queries and relationships.

Model Definition

Define models with proper field types, constraints, and metadata.

from django.db import models from django.core.validators import MinValueValidator, MaxValueValidator

class User(models.Model): email = models.EmailField(unique=True, db_index=True) name = models.CharField(max_length=100) is_active = models.BooleanField(default=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True)

class Meta:
    ordering = ['-created_at']
    indexes = [
        models.Index(fields=['email']),
        models.Index(fields=['created_at', 'is_active']),
    ]
    verbose_name = 'User'
    verbose_name_plural = 'Users'

def __str__(self):
    return self.email

class Post(models.Model): title = models.CharField(max_length=200) content = models.TextField() author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='posts') published = models.BooleanField(default=False) views = models.PositiveIntegerField(default=0) created_at = models.DateTimeField(auto_now_add=True)

class Meta:
    ordering = ['-created_at']
    indexes = [
        models.Index(fields=['author', 'published']),
    ]

QuerySet API Basics

Use Django's QuerySet API for efficient database queries.

All records

users = User.objects.all()

Filtering

active_users = User.objects.filter(is_active=True) inactive_users = User.objects.exclude(is_active=True)

Get single record (raises exception if not found or multiple found)

user = User.objects.get(email='user@example.com')

Get or create

user, created = User.objects.get_or_create( email='user@example.com', defaults={'name': 'John Doe'} )

Update or create

user, created = User.objects.update_or_create( email='user@example.com', defaults={'name': 'Jane Doe', 'is_active': True} )

Chaining filters

posts = Post.objects.filter(published=True).filter(author__is_active=True)

Order by

users = User.objects.order_by('-created_at', 'name')

Limit results

recent_users = User.objects.all()[:10]

Count

user_count = User.objects.filter(is_active=True).count()

Exists

has_active_users = User.objects.filter(is_active=True).exists()

Q Objects for Complex Queries

Build complex queries with Q objects for OR and NOT operations.

from django.db.models import Q

OR queries

users = User.objects.filter( Q(name__icontains='john') | Q(email__icontains='john') )

AND with OR

users = User.objects.filter( Q(is_active=True) & (Q(name__icontains='john') | Q(email__icontains='john')) )

NOT queries

users = User.objects.filter(~Q(is_active=True))

Complex combinations

posts = Post.objects.filter( Q(published=True) & (Q(author__name__icontains='john') | Q(title__icontains='important')) & ~Q(views__lt=100) )

Dynamic query building

def search_users(name=None, email=None, is_active=None): query = Q() if name: query &= Q(name__icontains=name) if email: query &= Q(email__icontains=email) if is_active is not None: query &= Q(is_active=is_active) return User.objects.filter(query)

F Objects for Field References

Use F objects to reference model fields in queries and updates.

from django.db.models import F

Compare fields

posts = Post.objects.filter(views__gt=F('author__posts__count'))

Update based on current value

Post.objects.filter(published=True).update(views=F('views') + 1)

Avoid race conditions

post = Post.objects.get(id=1) post.views = F('views') + 1 post.save() post.refresh_from_db() # Get updated value

Complex expressions

from django.db.models import ExpressionWrapper, IntegerField

Post.objects.annotate( adjusted_views=ExpressionWrapper( F('views') * 2 + 10, output_field=IntegerField() ) )

Aggregation and Annotation

Perform database-level calculations and add computed fields.

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

Simple aggregation

from django.db.models import Avg avg_views = Post.objects.aggregate(Avg('views'))

Returns: {'views__avg': 42.5}

Multiple aggregations

stats = Post.objects.aggregate( total_posts=Count('id'), avg_views=Avg('views'), max_views=Max('views'), min_views=Min('views') )

Annotation (adds field to each object)

users = User.objects.annotate( post_count=Count('posts'), total_views=Sum('posts__views') )

for user in users: print(f"{user.name}: {user.post_count} posts, {user.total_views} views")

Filter by annotation

popular_users = User.objects.annotate( post_count=Count('posts') ).filter(post_count__gt=10)

Complex annotations

from django.db.models import Case, When, Value, CharField

User.objects.annotate( user_type=Case( When(post_count__gt=10, then=Value('prolific')), When(post_count__gt=5, then=Value('active')), default=Value('casual'), output_field=CharField() ) )

Prefetch and Select Related (N+1 Prevention)

Optimize queries by reducing database hits with eager loading.

Select related (for ForeignKey and OneToOne)

posts = Post.objects.select_related('author').all() for post in posts: print(post.author.name) # No additional query

Prefetch related (for ManyToMany and reverse ForeignKey)

from django.db.models import Prefetch

users = User.objects.prefetch_related('posts').all() for user in users: for post in user.posts.all(): # No additional query print(post.title)

Custom prefetch

users = User.objects.prefetch_related( Prefetch( 'posts', queryset=Post.objects.filter(published=True).order_by('-created_at') ) )

Multiple levels

posts = Post.objects.select_related( 'author' ).prefetch_related( 'author__posts' # Prefetch all posts by the same author )

Combining both

Post.objects.select_related('author').prefetch_related('tags')

Custom Managers and QuerySets

Create reusable query logic with custom managers and querysets.

from django.db import models

class PublishedQuerySet(models.QuerySet): def published(self): return self.filter(published=True)

def recent(self):
    return self.order_by('-created_at')[:10]

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

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

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

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

class Post(models.Model): # fields... objects = models.Manager() # Default manager published_posts = PublishedManager() # Custom manager

class Meta:
    base_manager_name = 'objects'

Usage

Post.published_posts.published().recent() Post.published_posts.published().by_author(user)

Chaining custom methods

class UserQuerySet(models.QuerySet): def active(self): return self.filter(is_active=True)

def with_posts(self):
    return self.annotate(post_count=Count('posts')).filter(post_count__gt=0)

User.objects.active().with_posts()

Transactions and Atomic Blocks

Ensure data consistency with database transactions.

from django.db import transaction

Atomic decorator

@transaction.atomic def create_user_with_post(email, name, post_title): user = User.objects.create(email=email, name=name) Post.objects.create(title=post_title, author=user) return user

Context manager

def update_user_posts(user_id): try: with transaction.atomic(): user = User.objects.select_for_update().get(id=user_id) user.posts.update(published=True) user.is_active = True user.save() except Exception as e: # Transaction is rolled back raise

Savepoints

from django.db import transaction

with transaction.atomic(): user = User.objects.create(email='user@example.com')

sid = transaction.savepoint()
try:
    Post.objects.create(title='Test', author=user)
except:
    transaction.savepoint_rollback(sid)
else:
    transaction.savepoint_commit(sid)

Select for update (locking)

with transaction.atomic(): user = User.objects.select_for_update().get(id=1) user.is_active = False user.save()

Advanced Select and Prefetch Patterns

Master complex query optimization with advanced eager loading techniques.

from django.db.models import Prefetch, Count, Q

Basic select_related (ForeignKey, OneToOne)

posts = Post.objects.select_related('author', 'category')

Multi-level select_related

comments = Comment.objects.select_related('post__author__profile')

Prefetch with custom queryset

users = User.objects.prefetch_related( Prefetch( 'posts', queryset=Post.objects.filter(published=True).select_related('category'), to_attr='published_posts' ) )

Multiple prefetch with different filters

authors = User.objects.prefetch_related( Prefetch( 'posts', queryset=Post.objects.filter(published=True), to_attr='published_posts' ), Prefetch( 'posts', queryset=Post.objects.filter(published=False), to_attr='draft_posts' ) )

Nested prefetch

posts = Post.objects.prefetch_related( Prefetch( 'comments', queryset=Comment.objects.select_related('author').prefetch_related( Prefetch( 'replies', queryset=Comment.objects.select_related('author') ) ) ) )

Prefetch with annotations

users = User.objects.prefetch_related( Prefetch( 'posts', queryset=Post.objects.annotate( comment_count=Count('comments') ).filter(comment_count__gt=0) ) )

Database Functions and Expressions

Leverage database functions for complex operations.

from django.db.models import F, Value, CharField, Case, When, Q from django.db.models.functions import Concat, Lower, Upper, Length, Substr, Coalesce

String operations

users = User.objects.annotate( full_name=Concat('first_name', Value(' '), 'last_name') )

users = User.objects.annotate( email_lower=Lower('email'), name_upper=Upper('name') )

String functions

posts = Post.objects.annotate( title_length=Length('title') ).filter(title_length__gt=50)

Substring

posts = Post.objects.annotate( title_preview=Substr('title', 1, 50) )

Coalesce (return first non-null value)

posts = Post.objects.annotate( display_name=Coalesce('custom_title', 'title', Value('Untitled')) )

Date functions

from django.db.models.functions import TruncDate, TruncMonth, ExtractYear, Now

posts = Post.objects.annotate( created_date=TruncDate('created_at'), created_month=TruncMonth('created_at'), created_year=ExtractYear('created_at') )

Date arithmetic

from datetime import timedelta from django.utils import timezone

recent_posts = Post.objects.filter( created_at__gte=timezone.now() - timedelta(days=7) )

Mathematical functions

from django.db.models.functions import Abs, Ceil, Floor, Round

products = Product.objects.annotate( price_rounded=Round('price'), discount_abs=Abs('discount') )

Conditional expressions

User.objects.annotate( user_type=Case( When(posts__count__gt=100, then=Value('power_user')), When(posts__count__gt=10, then=Value('active')), When(posts__count__gt=0, then=Value('casual')), default=Value('lurker'), output_field=CharField() ) )

Complex conditional updates

Post.objects.update( status=Case( When(Q(published=True) & Q(views__gt=1000), then=Value('viral')), When(Q(published=True) & Q(views__gt=100), then=Value('popular')), When(published=True, then=Value('published')), default=Value('draft'), output_field=CharField() ) )

Advanced Aggregation Patterns

Perform complex database-level calculations.

from django.db.models import ( Count, Sum, Avg, Max, Min, StdDev, Variance, Q, F, Value, CharField, When, Case ) from django.db.models.functions import Coalesce

Multiple aggregations with filters

stats = Post.objects.aggregate( total_posts=Count('id'), published_posts=Count('id', filter=Q(published=True)), draft_posts=Count('id', filter=Q(published=False)), avg_views=Avg('views'), max_views=Max('views'), total_views=Sum('views'), std_dev_views=StdDev('views') )

Conditional aggregation

User.objects.aggregate( active_users=Count('id', filter=Q(is_active=True)), inactive_users=Count('id', filter=Q(is_active=False)), avg_posts_active=Avg('posts__count', filter=Q(is_active=True)) )

Annotation with conditional aggregation

users = User.objects.annotate( published_post_count=Count('posts', filter=Q(posts__published=True)), draft_post_count=Count('posts', filter=Q(posts__published=False)), total_views=Sum('posts__views'), avg_post_views=Avg('posts__views') ).filter(published_post_count__gt=0)

Group by with annotation

from django.db.models.functions import TruncDate

daily_stats = Post.objects.annotate( date=TruncDate('created_at') ).values('date').annotate( post_count=Count('id'), total_views=Sum('views'), avg_views=Avg('views') ).order_by('-date')

Subquery aggregation

from django.db.models import OuterRef, Subquery

Get latest comment for each post

latest_comment = Comment.objects.filter( post=OuterRef('pk') ).order_by('-created_at')

posts = Post.objects.annotate( latest_comment_date=Subquery(latest_comment.values('created_at')[:1]), latest_comment_author=Subquery(latest_comment.values('author__name')[:1]) )

Complex nested aggregation

User.objects.annotate( total_post_views=Sum('posts__views'), total_comment_count=Count('posts__comments'), avg_comments_per_post=Case( When(posts__count=0, then=Value(0)), default=Count('posts__comments') / Count('posts', distinct=True) ) )

Database Indexes and Optimization

Optimize query performance with proper indexing.

class Post(models.Model): title = models.CharField(max_length=200, db_index=True) author = models.ForeignKey(User, on_delete=models.CASCADE) published = models.BooleanField(default=False) created_at = models.DateTimeField(auto_now_add=True)

class Meta:
    indexes = [
        # Single field
        models.Index(fields=['created_at']),

        # Composite index
        models.Index(fields=['author', 'published']),

        # Descending index
        models.Index(fields=['-created_at']),

        # Named index
        models.Index(fields=['title'], name='post_title_idx'),

        # Partial index (PostgreSQL)
        models.Index(
            fields=['author'],
            name='published_posts_idx',
            condition=models.Q(published=True)
        ),

        # Expression index (PostgreSQL)
        models.Index(
            Lower('title'),
            name='post_title_lower_idx'
        ),

        # Multi-column with includes (PostgreSQL)
        models.Index(
            fields=['author'],
            name='author_includes_idx',
            include=['title', 'created_at']
        ),
    ]

    # Unique together
    unique_together = [['author', 'title']]

    # Constraints (Django 2.2+)
    constraints = [
        models.UniqueConstraint(
            fields=['author', 'slug'],
            name='unique_author_slug'
        ),
        models.CheckConstraint(
            check=Q(views__gte=0),
            name='views_non_negative'
        ),
    ]

Query optimization techniques

Only load needed fields

posts = Post.objects.only('id', 'title', 'author_id')

Defer heavy fields

posts = Post.objects.defer('content', 'metadata')

Values and values_list for dictionaries/tuples

post_data = Post.objects.values('id', 'title', 'author__name') post_ids = Post.objects.values_list('id', flat=True)

Combine optimizations

posts = Post.objects.select_related('author').only( 'title', 'author__name' ).filter( published=True )

Use iterator() for large querysets

for post in Post.objects.iterator(chunk_size=1000): process_post(post)

Use explain() to analyze queries

print(Post.objects.filter(published=True).explain(analyze=True))

Model Inheritance Patterns

Implement proper model inheritance strategies.

from django.db import models

Abstract base classes (no database table)

class TimeStampedModel(models.Model): created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True)

class Meta:
    abstract = True

class Post(TimeStampedModel): title = models.CharField(max_length=200) content = models.TextField() # Inherits created_at and updated_at

Multi-table inheritance (separate tables, joins required)

class BaseContent(models.Model): title = models.CharField(max_length=200) created_at = models.DateTimeField(auto_now_add=True)

class Article(BaseContent): # Has implicit OneToOne to BaseContent body = models.TextField() published = models.BooleanField(default=False)

class Video(BaseContent): duration = models.IntegerField() video_url = models.URLField()

Proxy models (same table, different behavior)

class PublishedPostManager(models.Manager): def get_queryset(self): return super().get_queryset().filter(published=True)

class Post(models.Model): title = models.CharField(max_length=200) published = models.BooleanField(default=False)

class Meta:
    ordering = ['-created_at']

class PublishedPost(Post): objects = PublishedPostManager()

class Meta:
    proxy = True
    ordering = ['-created_at']

def publish(self):
    self.published = True
    self.save()

When to use each:

- Abstract: Share fields/methods, no polymorphic queries

- Multi-table: Need polymorphic queries, different fields

- Proxy: Same fields, different managers/methods

Advanced QuerySet Methods

Master advanced QuerySet operations.

Bulk operations for performance

posts = [ Post(title=f'Post {i}', author=user) for i in range(1000) ] Post.objects.bulk_create(posts, batch_size=100)

Bulk update (Django 2.2+)

posts = Post.objects.filter(author=user) for post in posts: post.views += 1 Post.objects.bulk_update(posts, ['views'], batch_size=100)

Bulk create with returning IDs (PostgreSQL)

posts = Post.objects.bulk_create(posts, batch_size=100, ignore_conflicts=True)

Update with F expressions (atomic, no race conditions)

Post.objects.filter(id=1).update(views=F('views') + 1)

Get or create with complex lookups

user, created = User.objects.get_or_create( email='user@example.com', defaults={ 'name': 'John Doe', 'is_active': True } )

Update or create

post, created = Post.objects.update_or_create( author=user, slug='my-post', defaults={ 'title': 'My Post', 'content': 'Updated content' } )

In bulk (Django 4.1+)

Post.objects.bulk_create( posts, update_conflicts=True, update_fields=['title', 'content'], unique_fields=['author', 'slug'] )

Union, intersection, difference

published = Post.objects.filter(published=True) featured = Post.objects.filter(featured=True)

all_posts = published.union(featured) # Posts that are published OR featured both = published.intersection(featured) # Posts that are both only_published = published.difference(featured) # Published but not featured

Distinct

authors = Post.objects.values('author').distinct()

With PostgreSQL distinct on

posts = Post.objects.order_by('author', '-created_at').distinct('author')

Reverse queryset

recent_first = Post.objects.order_by('-created_at') oldest_first = recent_first.reverse()

None queryset

empty = Post.objects.none() # Returns empty queryset

Raw SQL When Needed

Use raw SQL for complex queries that ORM cannot handle efficiently.

Raw queries

users = User.objects.raw('SELECT * FROM app_user WHERE is_active = %s', [True]) for user in users: print(user.name)

Execute custom SQL

from django.db import connection

def get_user_stats(): with connection.cursor() as cursor: cursor.execute(""" SELECT u.id, u.name, COUNT(p.id) as post_count FROM app_user u LEFT JOIN app_post p ON p.author_id = u.id GROUP BY u.id, u.name HAVING COUNT(p.id) > 5 """) columns = [col[0] for col in cursor.description] return [dict(zip(columns, row)) for row in cursor.fetchall()]

Combining ORM with raw SQL

User.objects.raw(""" SELECT * FROM app_user WHERE id IN ( SELECT DISTINCT author_id FROM app_post WHERE published = TRUE ) """)

Migrations Best Practices

Manage database schema changes safely and efficiently.

Create migration

python manage.py makemigrations

Custom migration

from django.db import migrations

def forwards_func(apps, schema_editor): User = apps.get_model('app', 'User') for user in User.objects.all(): user.is_active = True user.save()

def reverse_func(apps, schema_editor): pass

class Migration(migrations.Migration): dependencies = [ ('app', '0001_initial'), ]

operations = [
    migrations.RunPython(forwards_func, reverse_func),
]

Add field with default

class Migration(migrations.Migration): operations = [ migrations.AddField( model_name='user', name='status', field=models.CharField(max_length=20, default='active'), ), ]

Rename field

operations = [ migrations.RenameField( model_name='user', old_name='name', new_name='full_name', ), ]

Add index

operations = [ migrations.AddIndex( model_name='post', index=models.Index(fields=['author', 'created_at']), ), ]

When to Use This Skill

Use django-orm-patterns when building modern, production-ready applications that require advanced patterns, best practices, and optimal performance.

Signal Patterns and Best Practices

Use Django signals carefully for decoupled event handling.

from django.db.models.signals import post_save, pre_save, post_delete, m2m_changed from django.dispatch import receiver from django.db import transaction

Basic signal receiver

@receiver(post_save, sender=Post) def post_created_handler(sender, instance, created, **kwargs): if created: # Send notification notify_followers(instance.author, instance)

Pre-save validation

@receiver(pre_save, sender=User) def normalize_email(sender, instance, **kwargs): if instance.email: instance.email = instance.email.lower()

Conditional signal execution

@receiver(post_save, sender=Post) def update_stats(sender, instance, created, update_fields, **kwargs): # Skip if only certain fields updated if update_fields and 'views' in update_fields: return

# Update statistics
instance.author.update_post_count()

M2M changed signal

@receiver(m2m_changed, sender=Post.tags.through) def tags_changed(sender, instance, action, **kwargs): if action == 'post_add': # Tags were added pass elif action == 'post_remove': # Tags were removed pass

Avoid signals in transactions

@receiver(post_save, sender=Order) def send_confirmation_email(sender, instance, created, **kwargs): if created: # Wait for transaction to commit transaction.on_commit(lambda: send_email(instance))

Disconnect signals when needed

from django.test import TestCase

class PostTestCase(TestCase): def setUp(self): # Disconnect signal for testing post_save.disconnect(post_created_handler, sender=Post)

def tearDown(self):
    # Reconnect signal
    post_save.connect(post_created_handler, sender=Post)

Custom Field Types

Create reusable custom field types for complex data.

from django.db import models import json

JSON field (before Django 3.1)

class JSONField(models.TextField): def from_db_value(self, value, expression, connection): if value is None: return value return json.loads(value)

def to_python(self, value):
    if isinstance(value, dict):
        return value
    if value is None:
        return value
    return json.loads(value)

def get_prep_value(self, value):
    return json.dumps(value)

Encrypted field

from cryptography.fernet import Fernet

class EncryptedField(models.TextField): def init(self, *args, **kwargs): self.cipher_suite = Fernet(settings.FIELD_ENCRYPTION_KEY) super().init(*args, **kwargs)

def from_db_value(self, value, expression, connection):
    if value is None:
        return value
    return self.cipher_suite.decrypt(value.encode()).decode()

def get_prep_value(self, value):
    if value is None:
        return value
    return self.cipher_suite.encrypt(value.encode()).decode()

Usage

class User(models.Model): metadata = JSONField(default=dict) ssn = EncryptedField()

Array field (PostgreSQL)

from django.contrib.postgres.fields import ArrayField

class Post(models.Model): tags = ArrayField(models.CharField(max_length=50), default=list) ratings = ArrayField(models.IntegerField(), default=list)

class Meta:
    indexes = [
        models.Index(fields=['tags']),
    ]

Query array fields

posts = Post.objects.filter(tags__contains=['django']) posts = Post.objects.filter(tags__overlap=['python', 'django'])

Query Debugging and Profiling

Debug and optimize database queries effectively.

from django.db import connection, reset_queries from django.test.utils import override_settings import time

Log all queries

@override_settings(DEBUG=True) def analyze_queries(func): def wrapper(*args, **kwargs): reset_queries() start = time.time() result = func(*args, **kwargs) end = time.time()

    print(f"Function: {func.__name__}")
    print(f"Number of queries: {len(connection.queries)}")
    print(f"Time taken: {end - start:.2f}s")

    for query in connection.queries:
        print(f"SQL: {query['sql']}")
        print(f"Time: {query['time']}s\n")

    return result
return wrapper

Usage

@analyze_queries def get_user_posts(user_id): user = User.objects.get(id=user_id) posts = user.posts.all() return list(posts)

Django Debug Toolbar integration

Add to INSTALLED_APPS

INSTALLED_APPS = [ 'debug_toolbar', ]

Middleware

MIDDLEWARE = [ 'debug_toolbar.middleware.DebugToolbarMiddleware', ]

Explain queries

queryset = Post.objects.filter(published=True) print(queryset.explain()) # Basic explain print(queryset.explain(verbose=True)) # Verbose print(queryset.explain(analyze=True)) # Actually run query

Query count assertion in tests

from django.test import TestCase from django.test.utils import override_settings

class PostTestCase(TestCase): def test_query_count(self): with self.assertNumQueries(3): # Should execute exactly 3 queries user = User.objects.get(id=1) posts = list(user.posts.all()) comments = list(Comment.objects.filter(post__in=posts))

Find duplicate queries

def find_duplicate_queries(): from collections import Counter

queries = [q['sql'] for q in connection.queries]
duplicates = [q for q, count in Counter(queries).items() if count > 1]

for sql in duplicates:
    print(f"Duplicate query: {sql}")

Advanced Manager Patterns

Build sophisticated custom managers for complex business logic.

from django.db import models from django.db.models import Q, Count, Avg

class PostQuerySet(models.QuerySet): def published(self): return self.filter(published=True)

def draft(self):
    return self.filter(published=False)

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

def popular(self, min_views=100):
    return self.filter(views__gte=min_views)

def recent(self, days=7):
    from django.utils import timezone
    from datetime import timedelta
    cutoff = timezone.now() - timedelta(days=days)
    return self.filter(created_at__gte=cutoff)

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

def optimized(self):
    return self.select_related('author').prefetch_related('comments')

class PostManager(models.Manager.from_queryset(PostQuerySet)): def get_queryset(self): return super().get_queryset().filter(deleted_at__isnull=True)

def with_deleted(self):
    return super().get_queryset()

class Post(models.Model): title = models.CharField(max_length=200) author = models.ForeignKey(User, on_delete=models.CASCADE) published = models.BooleanField(default=False) views = models.IntegerField(default=0) deleted_at = models.DateTimeField(null=True, blank=True) created_at = models.DateTimeField(auto_now_add=True)

objects = PostManager()

def soft_delete(self):
    from django.utils import timezone
    self.deleted_at = timezone.now()
    self.save()

Usage - methods chain naturally

recent_popular = Post.objects.published().recent().popular().with_stats() author_drafts = Post.objects.by_author(user).draft().optimized()

Multiple manager pattern

class AllPostsManager(models.Manager): def get_queryset(self): return super().get_queryset()

class Post(models.Model): # ... fields ... objects = PostManager() # Default, excludes deleted all_objects = AllPostsManager() # Includes deleted

Database-Specific Features

Leverage PostgreSQL-specific features when available.

from django.contrib.postgres.fields import ArrayField, JSONField, HStoreField from django.contrib.postgres.search import SearchVector, SearchQuery, SearchRank from django.contrib.postgres.aggregates import ArrayAgg, StringAgg

Full-text search

class Post(models.Model): title = models.CharField(max_length=200) content = models.TextField() search_vector = SearchVectorField(null=True)

class Meta:
    indexes = [
        GinIndex(fields=['search_vector']),
    ]

Update search vector

from django.contrib.postgres.search import SearchVector

Post.objects.update( search_vector=SearchVector('title', weight='A') + SearchVector('content', weight='B') )

Search

query = SearchQuery('django') posts = Post.objects.annotate( rank=SearchRank('search_vector', query) ).filter(search_vector=query).order_by('-rank')

Array aggregation

authors = User.objects.annotate( post_titles=ArrayAgg('posts__title', distinct=True), tags_list=StringAgg('posts__tags', delimiter=', ', distinct=True) )

JSON operations

from django.contrib.postgres.fields.jsonb import KeyTextTransform

users = User.objects.annotate( city=KeyTextTransform('city', 'metadata') ).filter(city='New York')

Range fields

from django.contrib.postgres.fields import IntegerRangeField, DateRangeField

class Event(models.Model): name = models.CharField(max_length=200) date_range = DateRangeField() capacity = IntegerRangeField()

from django.db.models import Q from datetime import date

Find events happening on a specific date

events = Event.objects.filter(date_range__contains=date(2024, 1, 15))

Find overlapping events

events = Event.objects.filter( date_range__overlap=(date(2024, 1, 1), date(2024, 1, 31)) )

Django ORM Best Practices

• Use select_related and prefetch_related - Always optimize queries to prevent N+1 problems

• Index frequently queried fields - Add database indexes for fields used in filters and joins

• Use get_or_create carefully - Wrap in transactions when dealing with race conditions

• Avoid queries in loops - Batch operations and use bulk methods when possible

• Use only() and defer() wisely - Load only necessary fields for large models

• Leverage F() expressions - Use database-level operations to avoid race conditions

• Use transactions for data integrity - Wrap related operations in atomic blocks

• Create custom managers - Encapsulate common query patterns in reusable managers

• Use exists() for checks - More efficient than count() when only checking existence

• Monitor query performance - Use django-debug-toolbar to identify slow queries

• Implement soft deletes with managers - Use custom managers to hide deleted records

• Use database functions - Leverage Django's database functions for complex operations

• Batch database operations - Use bulk_create and bulk_update for large datasets

• Use iterator() for large datasets - Avoid loading entire querysets into memory

• Apply database constraints - Use CheckConstraint and UniqueConstraint for data integrity

Django ORM Common Pitfalls

• N+1 query problem - Forgetting to use select_related or prefetch_related causes excessive queries

• Loading too much data - Using .all() without pagination can cause memory issues

• Inefficient updates - Using save() in loops instead of bulk_update or update()

• Missing database indexes - Slow queries on unindexed fields in large tables

• Incorrect use of get() - Not handling DoesNotExist or MultipleObjectsReturned exceptions

• Lazy evaluation confusion - Querysets are lazy; understand when they actually execute

• Transaction isolation issues - Not using select_for_update when needed for locking

• Mixing F() with save() - Must call refresh_from_db() after saving F() expressions

• Inefficient aggregations - Running Python calculations instead of database aggregations

• Migration conflicts - Not coordinating migrations in team environments

• Signal performance issues - Signals in tight loops can cause performance problems

• Overusing signals - Prefer explicit calls over implicit signal-based logic

• Not using transactions with signals - Signals fire before transaction commit by default

• Incorrect distinct() usage - Using distinct() without understanding its implications

• Ignoring database-specific features - Missing out on PostgreSQL full-text search, arrays, etc.

Resources

• Django ORM Documentation

• Django QuerySet API Reference

• Django Database Optimization

• Django Migration Guide

• Django Database Transactions