Django ORM
Mastery Guide

class Profile(models.Model):
    user     = models.OneToOneField(
                   User,
                   on_delete=models.CASCADE,
                   related_name='profile'   # user.profile
               )
    bio      = models.TextField(blank=True)
    avatar   = models.ImageField(upload_to='avatars/', blank=True)
    website  = models.URLField(blank=True)
    location = models.CharField(max_length=100, blank=True)

# ── Queries ──────────────────────────────────────────────────

# Forward access (Profile → User)
profile = Profile.objects.get(pk=1)
user = profile.user                    # access the related user

# Reverse access (User → Profile)
user = User.objects.get(username='alice')
profile = user.profile                 # because related_name='profile'

# Without related_name it would be: user.profile_set — but OneToOne
# creates a single object accessor, not a manager

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

    def __str__(self):
        return self.name


class Post(models.Model):
    title    = models.CharField(max_length=200)
    author   = models.ForeignKey(
                   User,
                   on_delete=models.CASCADE,
                   related_name='posts'        # user.posts.all()
               )
    category = models.ForeignKey(
                   Category,
                   on_delete=models.SET_NULL,
                   null=True,
                   blank=True,
                   related_name='posts'        # category.posts.all()
               )

# ── Queries ──────────────────────────────────────────────────────

# Forward: Post → Author (simple attribute access)
post = Post.objects.get(pk=1)
print(post.author.username)       # ⚠️ This does a 2nd query!
print(post.author_id)             # ✅ No extra query — stored as FK column

# Reverse: Author → Posts (returns a Manager)
alice = User.objects.get(username='alice')
alices_posts = alice.posts.all()           # QuerySet of all Alice's posts
alices_posts = alice.posts.filter(is_active=True)

# Category → Posts
tech = Category.objects.get(slug='tech')
tech_posts = tech.posts.order_by('-created_at')

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

    def __str__(self): return self.name


class Post(models.Model):
    title = models.CharField(max_length=200)
    tags  = models.ManyToManyField(Tag, related_name='posts', blank=True)

# ── Many-to-Many Queries ─────────────────────────────────────────

# ADD tags to a post
post = Post.objects.get(pk=1)
django_tag = Tag.objects.get(slug='django')
python_tag = Tag.objects.get(slug='python')

post.tags.add(django_tag)
post.tags.add(django_tag, python_tag)     # add multiple
post.tags.set([django_tag, python_tag])   # replace all

# REMOVE tags
post.tags.remove(python_tag)
post.tags.clear()                          # remove all

# QUERY: all tags of a post
post.tags.all()

# REVERSE: all posts with a tag
tag = Tag.objects.get(slug='django')
tag.posts.all()                            # uses related_name

# Filter posts that have a specific tag
Post.objects.filter(tags__slug='django')

class Student(models.Model):
    user = models.OneToOneField(User, on_delete=models.CASCADE)


class Course(models.Model):
    title    = models.CharField(max_length=200)
    students = models.ManyToManyField(
                   Student,
                   through='Enrollment',
                   related_name='courses'
               )


class Enrollment(models.Model):
    """The through/junction model with extra data."""
    student    = models.ForeignKey(Student, on_delete=models.CASCADE)
    course     = models.ForeignKey(Course, on_delete=models.CASCADE)
    enrolled_at = models.DateTimeField(auto_now_add=True)
    grade      = models.CharField(max_length=2, blank=True)  # A, B+, etc.
    completed  = models.BooleanField(default=False)

    class Meta:
        unique_together = ['student', 'course']  # one enrollment per combo

# ── Queries with through model ────────────────────────────────────

# Create enrollment (must use through model directly)
Enrollment.objects.create(student=student, course=course, grade='A')

# Get all students in a course (through course.students)
course.students.all()

# Get enrollment details
enrollment = Enrollment.objects.get(student=student, course=course)
print(enrollment.grade)

# All courses a student completed
student.courses.filter(enrollment__completed=True)

# WITHOUT related_name → Django auto-creates: user.post_set
class Post(models.Model):
    author = models.ForeignKey(User, on_delete=models.CASCADE)

user.post_set.all()     # works but ugly

# WITH related_name → you control the accessor name
class Post(models.Model):
    author = models.ForeignKey(User, on_delete=models.CASCADE,
                                related_name='posts')

user.posts.all()        # clean and readable ✅

# ── When to use '+' as related_name ──────────────────────────────
# Use '+' to DISABLE the reverse relation (saves memory, avoids clashes)
class Post(models.Model):
    created_by = models.ForeignKey(User, on_delete=models.CASCADE,
                                     related_name='+')
# Now user.post_set does NOT exist — accessing it raises AttributeError
# Use '+' when you'll NEVER need reverse access from that field

# ── Multiple FKs to same model — need related_name! ──────────────
class Transfer(models.Model):
    from_account = models.ForeignKey('Account', on_delete=models.CASCADE,
                                       related_name='outgoing_transfers')
    to_account   = models.ForeignKey('Account', on_delete=models.CASCADE,
                                       related_name='incoming_transfers')
# Without distinct related_names, Django raises a clash error

# values() → list of dicts
Post.objects.values('id', 'title', 'author__username')
# [{'id': 1, 'title': 'Hello', 'author__username': 'alice'}, ...]

# values_list() → list of tuples
Post.objects.values_list('id', 'title')
# [(1, 'Hello'), (2, 'World'), ...]

# flat=True for single field → plain list
Post.objects.values_list('id', flat=True)
# [1, 2, 3, 4, ...]  ← Just a flat list of IDs

# Practical use: get all unique categories
Category.objects.values_list('name', flat=True)

# Practical use: build choices for a form
choices = list(Category.objects.values_list('id', 'name'))

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

# ── aggregate() → single result dict ────────────────────────────
result = Post.objects.aggregate(
    total=Count('id'),
    avg_views=Avg('view_count'),
    max_views=Max('view_count'),
    total_views=Sum('view_count'),
)
# {'total': 42, 'avg_views': 128.5, 'max_views': 5000, ...}

# ── annotate() → per-object value ───────────────────────────────

# Count how many posts each author has
from django.contrib.auth.models import User

authors = User.objects.annotate(post_count=Count('posts'))
for a in authors:
    print(a.username, a.post_count)

# Count posts per category, ordered by count
Category.objects\
    .annotate(post_count=Count('posts'))\
    .order_by('-post_count')

# Filter on annotations (HAVING clause)
popular_authors = User.objects\
    .annotate(post_count=Count('posts'))\
    .filter(post_count__gte=5)

# Annotate with Sum — total order value per customer
from shop.models import Customer, Order, OrderItem

customers = Customer.objects.annotate(
    total_spent=Sum('orders__items__price')
).order_by('-total_spent')

from django.db.models import F

# ❌ BAD — Requires two queries and a Python round-trip
post = Post.objects.get(pk=1)
post.view_count += 1
post.save()

# ✅ GOOD — Single atomic SQL UPDATE (race-condition safe!)
Post.objects.filter(pk=1).update(view_count=F('view_count') + 1)

# Compare two fields on same row
# Products where stock is less than reorder level
Product.objects.filter(stock__lt=F('reorder_level'))

# Sort by computed difference
Product.objects.annotate(
    diff=F('selling_price') - F('cost_price')
).order_by('-diff')   # most profitable first

from django.db.models import Q

# OR: posts about Python OR Django
Post.objects.filter(
    Q(title__icontains='python') | Q(title__icontains='django')
)

# AND with OR: active posts by Alice OR Bob
Post.objects.filter(
    Q(is_active=True),
    Q(author__username='alice') | Q(author__username='bob')
)

# NOT: exclude posts about "old"
Post.objects.filter(~Q(title__icontains='old'))

# Complex: (A OR B) AND NOT C
Post.objects.filter(
    (Q(view_count__gt=1000) | Q(is_featured=True)) &
    ~Q(author__is_staff=True)
)

# Dynamic Q building (e.g., from search form)
q = Q()
if search_term:
    q &= Q(title__icontains=search_term)
if category_id:
    q &= Q(category_id=category_id)
results = Post.objects.filter(q)

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

# Annotate a "popularity" tier per post
posts = Post.objects.annotate(
    popularity=Case(
        When(view_count__gte=10000, then=Value('viral')),
        When(view_count__gte=1000,  then=Value('popular')),
        When(view_count__gte=100,   then=Value('growing')),
        default=Value('new'),
        output_field=CharField(),
    )
)

# Conditional ordering: active posts first
posts = Post.objects.annotate(
    priority=Case(
        When(is_active=True, then=Value(0)),
        default=Value(1),
        output_field=IntegerField(),
    )
).order_by('priority', '-created_at')

from django.db.models import OuterRef, Subquery

# Annotate each author with the title of their LATEST post
latest_post = Post.objects\
    .filter(author=OuterRef('pk'))\
    .order_by('-created_at')\
    .values('title')[:1]

authors = User.objects.annotate(
    latest_post_title=Subquery(latest_post)
)

from django.db import connection

# Method 1: Model.objects.raw() — returns model instances
posts = Post.objects.raw(
    'SELECT * FROM blog_post WHERE view_count > %s', [100]
)

# Method 2: connection.cursor() — for arbitrary SQL
with connection.cursor() as cursor:
    cursor.execute('SELECT COUNT(*) FROM blog_post')
    row = cursor.fetchone()
    total = row[0]

# ⚠️ ALWAYS use parameterized queries — NEVER string format!
# ❌ BAD (SQL Injection risk!)
cursor.execute(f'SELECT * FROM blog_post WHERE title = "{user_input}"')

# ✅ GOOD (parameterized)
cursor.execute('SELECT * FROM blog_post WHERE title = %s', [user_input])

# Models: Order → OrderItem → Product → Category

# Get orders that contain products in the "Electronics" category
Order.objects.filter(
    items__product__category__name='Electronics'
)

# Get users who wrote posts tagged with 'django'
User.objects.filter(posts__tags__slug='django').distinct()
# .distinct() because one user might match multiple times via M2M

# Filter across 4 levels deep
# Company → Department → Employee → Task → status='done'
Company.objects.filter(
    departments__employees__tasks__status='done'
)

# Annotate with nested count
from django.db.models import Count

Order.objects.annotate(
    item_count=Count('items'),               # direct relation
    product_count=Count('items__product', distinct=True)  # nested
)

# Order by related field
Post.objects.order_by('author__username')   # sort by author's name

# Filter + annotate across relations
Category.objects\
    .filter(posts__is_active=True)\
    .annotate(active_post_count=Count('posts'))\
    .filter(active_post_count__gt=5)\
    .order_by('-active_post_count')

# ❌ N+1 PROBLEM — 1 query for posts + N queries for authors
posts = Post.objects.all()  # Query 1: SELECT * FROM post

for post in posts:
    print(post.author.username)
    # Query 2: SELECT * FROM user WHERE id=1
    # Query 3: SELECT * FROM user WHERE id=2
    # Query 4: SELECT * FROM user WHERE id=3
    # ... N more queries for N posts!

# With 1000 posts → 1001 queries! ❌❌❌


# ✅ SOLUTION: select_related() — Fixes it with a JOIN
posts = Post.objects.select_related('author').all()
# SELECT post.*, user.* FROM post JOIN user ... — only 1 query!

for post in posts:
    print(post.author.username)  # No extra query — already loaded!

from django.db.models import Prefetch

# ── select_related (FK & OneToOne → JOIN) ─────────────────────
posts = Post.objects\
    .select_related('author')\           # author FK
    .select_related('author__profile')\   # chain: author → profile
    .select_related('category')           # category FK

# Shorthand: all in one call
posts = Post.objects.select_related('author', 'category')

# ── prefetch_related (M2M & Reverse FK) ──────────────────────
posts = Post.objects\
    .select_related('author')\           # FK → JOIN
    .prefetch_related('tags')             # M2M → separate query

# ── Prefetch() object for fine control ────────────────────────
active_posts_prefetch = Prefetch(
    'posts',
    queryset=Post.objects.filter(is_active=True).select_related('category'),
    to_attr='active_posts'   # store result in .active_posts instead of .posts
)

authors = User.objects.prefetch_related(active_posts_prefetch)
for author in authors:
    for post in author.active_posts:   # uses prefetched list
        print(post.title)

# only() — load ONLY specified fields (defer the rest)
posts = Post.objects.only('id', 'title', 'created_at')
# Accessing post.body will trigger an extra query!

# defer() — load everything EXCEPT specified fields
posts = Post.objects.defer('body')   # skip the big text column

# count() — efficient COUNT(*) SQL — don't use len(qs)!
total = Post.objects.filter(is_active=True).count()  # ✅
total = len(Post.objects.filter(is_active=True))      # ❌ loads all rows

# exists() — efficient EXISTS check — don't use if qs:
if Post.objects.filter(author=user).exists():  # ✅ SELECT 1 LIMIT 1
    print('has posts')

# iterator() — for large QuerySets, don't cache in memory
for post in Post.objects.all().iterator(chunk_size=500):
    process(post)
# Streams 500 rows at a time — great for data exports

# explain() — show the query execution plan
print(Post.objects.filter(is_active=True).explain())

# bulk_create — insert many rows in one query
posts = [
    Post(title=f'Post {i}', slug=f'post-{i}', author=author, body='...')
    for i in range(1000)
]
Post.objects.bulk_create(posts, batch_size=100)
# ✅ ~10 queries instead of 1000!

# bulk_update — update many rows efficiently
posts = list(Post.objects.filter(is_active=False))
for post in posts:
    post.is_active = True

Post.objects.bulk_update(posts, ['is_active'], batch_size=100)

# update() on QuerySet — even better for simple updates
Post.objects.filter(is_active=False).update(is_active=True)
# 1 single SQL UPDATE — fastest option

class Post(models.Model):
    # db_index=True → single column index
    slug       = models.SlugField(unique=True)          # unique implies index
    created_at = models.DateTimeField(db_index=True)    # frequent order_by
    is_active  = models.BooleanField(db_index=True)    # frequent filter

    class Meta:
        indexes = [
            # Composite index for queries that filter+order together
            models.Index(fields=['is_active', '-created_at'],
                          name='post_active_date_idx'),
        ]

# ⚠️ Index Rules:
# ✅ Index: columns in WHERE, ORDER BY, JOIN conditions
# ✅ Index: high-cardinality columns (many unique values)
# ❌ Don't index: columns with few unique values (e.g., boolean)
#    unless combined in a composite index
# ❌ Don't over-index: each index slows down INSERT/UPDATE/DELETE

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

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

    def popular(self, min_views=1000):
        return self.filter(view_count__gte=min_views)


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

    # Delegate custom methods
    def active(self): return self.get_queryset().active()
    def popular(self): return self.get_queryset().popular()


class Post(models.Model):
    # ... fields ...
    objects = PostManager()   # Replace default manager

# Now you can chain cleanly:
Post.objects.active().popular().order_by('-created_at')

from django.db.models.signals import post_save, pre_delete
from django.dispatch import receiver
from .models import Post

# Auto-create Profile when a User is created
@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    if created:
        Profile.objects.create(user=instance)

# Log before a Post is deleted
@receiver(pre_delete, sender=Post)
def log_post_deletion(sender, instance, **kwargs):
    DeletionLog.objects.create(
        model='Post', object_id=instance.id, title=instance.title
    )

# Connect signals in AppConfig.ready()
class BlogConfig(AppConfig):
    def ready(self):
        import blog.signals   # ← make sure this is imported

from django.db import transaction

# Decorator — entire function is atomic
@transaction.atomic
def transfer_funds(from_account, to_account, amount):
    from_account.balance -= amount
    from_account.save()
    to_account.balance += amount
    to_account.save()
    # If any exception → both saves are rolled back ✅

# Context manager — finer control
with transaction.atomic():
    order = Order.objects.create(customer=customer)
    for item_data in cart:
        OrderItem.objects.create(order=order, **item_data)
    cart.clear()

# Savepoints — nested atomic blocks
with transaction.atomic():
    order = Order.objects.create(...)
    try:
        with transaction.atomic():    # savepoint
            send_email(order)          # might fail
    except EmailError:
        pass                           # rolls back only email, not order

with transaction.atomic():
    # Locks the row until transaction ends — prevents race conditions
    account = Account.objects\
        .select_for_update()\
        .get(pk=account_id)
    account.balance -= amount
    account.save()

# Django 4.1+ supports async ORM operations
import asyncio
from django.http import JsonResponse

async def post_list(request):
    # Async versions of common operations
    posts = await Post.objects.filter(is_active=True)\
                              .aall()    # async all()

    post = await Post.objects.aget(pk=1)  # async get()

    await Post.objects.acreate(          # async create()
        title='Async Post', slug='async', author=user, body='...'
    )

    return JsonResponse({'count': await Post.objects.acount()})

# Note: prefetch_related / select_related work async too
# For loops over QuerySets use: async for post in Post.objects.all()