django-orm-patterns

---

name: django-orm-patterns description: Enforces Django ORM patterns with Pragmatic DDD: rich domain models, Managers, QuerySets, migrations, signals, performance. Activates when working with Django models, queries, or migrations. allowed-tools: Read, Grep, Glob, Bash(ruff:), Bash(mypy:), Bash(basedpyright:), Bash(make:), Bash(uv:), Bash(python:), Bash(pytest:), Bash(manage.py:), Bash(django-admin:*)

Django ORM Patterns (Pragmatic DDD)

These rules are NON-NEGOTIABLE. Violating any of them is a bug.

Models:

• NEVER put business logic in save() overrides for complex operations — use explicit model methods or services
• NEVER use null=True on CharField/TextField — use blank=True, default=""
• NEVER use ForeignKey without explicit on_delete
• NEVER use ForeignKey without related_name
• ALWAYS use db_index=True on fields used in filters/lookups (or Meta.indexes)
• ALWAYS create abstract base models for shared fields (TimeStampedModel with created_at, updated_at)
• ALWAYS define __str__ method on every model
• ALWAYS define class Meta: ordering, verbose_name, verbose_name_plural
• ALWAYS use UniqueConstraint / CheckConstraint in Meta.constraints — NEVER deprecated unique_together

Managers & QuerySets:

• NEVER put complex queries in views/services — encapsulate in custom QuerySet methods
• ALWAYS use MyQuerySet.as_manager() or Manager.from_queryset()
• ALWAYS name QuerySet methods as domain concepts (published(), active(), for_user(user))
• NEVER use raw SQL unless QuerySet API is genuinely insufficient — prefer .annotate(), .aggregate(), F(), Q()

Performance:

• NEVER cause N+1 queries — ALWAYS use select_related() (FK/OneToOne) and prefetch_related() (M2M/reverse FK)
• ALWAYS use only() / defer() for large models when only subset of fields needed
• ALWAYS use bulk_create() / bulk_update() for batch operations — NEVER loops with .save()
• ALWAYS use iterator() for large querysets in management commands / background tasks

Migrations:

• NEVER edit auto-generated migrations unless necessary (data migrations, RunSQL)
• ALWAYS separate data migrations from schema migrations
• ALWAYS use RunPython with reverse_code for reversible data migrations
• NEVER run migrate in production without review — ALWAYS review generated SQL with sqlmigrate

Signals:

• NEVER use signals for core business logic — use explicit model methods or services
• Signals ONLY for decoupled side effects (cache invalidation, audit logging, denormalization)
• ALWAYS use dispatch_uid to prevent duplicate signal registration

Quality:

• ALWAYS run the project's typecheck and test commands after any model change

These are the Django ORM patterns for AppVerk projects, following Pragmatic DDD where Django Models serve as rich domain models with business logic. All patterns target Python 3.13+, Django 5.1+.

Rich Domain Model Pattern

Abstract base model for shared timestamps. Models contain domain methods — they ARE the domain entities.

# apps/core/models.py
from django.db import models


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

    class Meta:
        abstract = True

# apps/orders/models.py
from django.db import models
from django.utils.translation import gettext_lazy as _

from apps.core.models import TimeStampedModel
from apps.orders.exceptions import DomainValidationError


class Order(TimeStampedModel):
    class Status(models.TextChoices):
        PENDING = "pending", _("Pending")
        CONFIRMED = "confirmed", _("Confirmed")
        CANCELLED = "cancelled", _("Cancelled")

    user = models.ForeignKey(
        "auth.User",
        on_delete=models.PROTECT,
        related_name="orders",
        db_index=True,
    )
    product = models.ForeignKey(
        "products.Product",
        on_delete=models.PROTECT,
        related_name="orders",
        db_index=True,
    )
    status = models.CharField(
        max_length=20,
        choices=Status.choices,
        default=Status.PENDING,
        db_index=True,
    )
    total_amount = models.DecimalField(max_digits=10, decimal_places=2)

    objects: "OrderManager"

    class Meta:
        ordering = ["-created_at"]
        verbose_name = _("order")
        verbose_name_plural = _("orders")
        constraints = [
            models.UniqueConstraint(
                fields=["user", "product"],
                condition=models.Q(status="pending"),
                name="unique_pending_order_per_user_product",
            ),
            models.CheckConstraint(
                check=models.Q(total_amount__gte=0),
                name="order_total_amount_non_negative",
            ),
        ]

    def __str__(self) -> str:
        return f"Order #{self.pk} — {self.user_id} ({self.status})"

    # --- Domain methods ---

    def can_cancel(self) -> bool:
        return self.status == self.Status.PENDING

    def cancel(self) -> None:
        if not self.can_cancel():
            raise DomainValidationError(
                f"Cannot cancel order in status '{self.status}'."
            )
        self.status = self.Status.CANCELLED
        self.save(update_fields=["status", "updated_at"])

Managers & QuerySets

Encapsulate all non-trivial query logic in custom QuerySet methods named after domain concepts.

# apps/orders/managers.py
from django.db import models


class OrderQuerySet(models.QuerySet["Order"]):  # type: ignore[name-defined]
    def active(self) -> "OrderQuerySet":
        return self.exclude(status="cancelled")

    def for_user(self, user: "AbstractBaseUser") -> "OrderQuerySet":  # type: ignore[name-defined]
        return self.filter(user=user)

    def with_totals(self) -> "OrderQuerySet":
        from django.db.models import Sum
        return self.annotate(item_total=Sum("items__price"))


OrderManager = OrderQuerySet.as_manager()

# apps/orders/models.py  (add to Order)
objects = OrderQuerySet.as_manager()

Usage — chain QuerySet methods in views or services:

# In a service or ViewSet.get_queryset()
queryset = (
    Order.objects
    .for_user(request.user)
    .active()
    .select_related("product")
    .order_by("-created_at")
)

Model Definition Patterns

ForeignKey conventions

# Good: explicit on_delete, related_name, db_index
user = models.ForeignKey(
    "auth.User",
    on_delete=models.PROTECT,   # PROTECT prevents accidental cascades
    related_name="orders",
    db_index=True,
)

Composite indexes

class Meta:
    indexes = [
        models.Index(fields=["user", "status"], name="order_user_status_idx"),
        models.Index(fields=["created_at", "status"], name="order_created_status_idx"),
    ]

Constraints

class Meta:
    constraints = [
        # Unique constraint (replaces deprecated unique_together)
        models.UniqueConstraint(fields=["user", "product"], name="unique_user_product"),
        # Check constraint enforced at DB level
        models.CheckConstraint(
            check=models.Q(quantity__gt=0),
            name="order_quantity_positive",
        ),
    ]

Performance Patterns

select_related vs prefetch_related

Relationship	Use	Notes
ForeignKey / OneToOne	select_related()	SQL JOIN; one query
ManyToMany / reverse FK	prefetch_related()	Separate query; Python join
Filtered reverse FK	Prefetch(queryset=...)	Custom queryset per prefetch

# Prefetch with custom queryset (avoid loading all items)
from django.db.models import Prefetch
from apps.orders.models import Order, OrderItem

orders = Order.objects.prefetch_related(
    Prefetch(
        "items",
        queryset=OrderItem.objects.filter(is_active=True).only("id", "price"),
        to_attr="active_items",
    )
)

Bulk operations

# bulk_create — never loop with .save()
items = [OrderItem(order=order, product_id=pid, price=p) for pid, p in products]
OrderItem.objects.bulk_create(items, batch_size=500)

# bulk_update — only update changed fields
for item in items:
    item.price = calculate_price(item)
OrderItem.objects.bulk_update(items, fields=["price"], batch_size=500)

iterator() for large datasets

# Management command / background task — avoids loading entire QS into memory
for order in Order.objects.filter(status="pending").iterator(chunk_size=200):
    process_order(order)

Migration Patterns

Data migration with RunPython

# apps/orders/migrations/0005_backfill_total_amount.py
from django.db import migrations, models


def backfill_totals(apps, schema_editor):
    Order = apps.get_model("orders", "Order")
    for order in Order.objects.iterator(chunk_size=500):
        order.total_amount = order.items.aggregate(
            total=models.Sum("price")
        )["total"] or 0
        order.save(update_fields=["total_amount"])


def reverse_backfill(apps, schema_editor):
    Order = apps.get_model("orders", "Order")
    Order.objects.update(total_amount=0)


class Migration(migrations.Migration):
    dependencies = [("orders", "0004_order_total_amount")]

    operations = [
        migrations.RunPython(backfill_totals, reverse_code=reverse_backfill),
    ]

Use RunSQL only when Django ORM cannot express the operation (e.g., DB-specific functions, index creation with custom options).

Signals

Use signals for decoupled side effects only — never for core business logic.

Good uses: cache invalidation, audit logging, denormalization, sending notifications. Bad uses: changing the entity's own state, enforcing business rules, triggering other domain operations.

# apps/orders/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver

from apps.orders.models import Order


@receiver(
    post_save,
    sender=Order,
    dispatch_uid="orders.invalidate_order_cache",  # prevents duplicate registration
)
def invalidate_order_cache(sender, instance: Order, **kwargs) -> None:
    from django.core.cache import cache
    cache.delete(f"order:{instance.pk}")

Connect in AppConfig.ready():

# apps/orders/apps.py
from django.apps import AppConfig


class OrdersConfig(AppConfig):
    name = "apps.orders"

    def ready(self) -> None:
        import apps.orders.signals  # noqa: F401

Optional Repository Pattern

Use when you need to: decouple services from Django ORM (e.g. testing without DB), or enforce a strict application-layer boundary.

For most views/services, direct QuerySet access is sufficient and preferred.

# apps/orders/repositories.py
from typing import Protocol

from apps.orders.exceptions import EntityNotFoundError
from apps.orders.models import Order


class OrderRepositoryProtocol(Protocol):
    def get_by_id(self, order_id: int) -> Order: ...
    def list_for_user(self, user_id: int) -> list[Order]: ...
    def save(self, order: Order) -> None: ...


class DjangoOrderRepository:
    def get_by_id(self, order_id: int) -> Order:
        try:
            return Order.objects.select_related("product").get(pk=order_id)
        except Order.DoesNotExist:
            raise EntityNotFoundError(f"Order {order_id} not found.")

    def list_for_user(self, user_id: int) -> list[Order]:
        return list(Order.objects.for_user(user_id).active().select_related("product"))

    def save(self, order: Order) -> None:
        order.save()

Testing Django ORM

Use factory_boy with DjangoModelFactory for all test data. Never call Model.objects.create() directly in tests.

# apps/orders/tests/factories.py
import factory
from factory.django import DjangoModelFactory

from apps.orders.models import Order


class UserFactory(DjangoModelFactory):
    class Meta:
        model = "auth.User"

    username = factory.Sequence(lambda n: f"user_{n}")
    email = factory.LazyAttribute(lambda o: f"{o.username}@example.com")


class OrderFactory(DjangoModelFactory):
    class Meta:
        model = Order

    user = factory.SubFactory(UserFactory)
    product = factory.SubFactory("apps.products.tests.factories.ProductFactory")
    status = Order.Status.PENDING
    total_amount = factory.Faker("pydecimal", left_digits=4, right_digits=2, positive=True)

Unit test for a model method — no DB needed:

# apps/orders/tests/test_models.py
import pytest

from apps.orders.exceptions import DomainValidationError
from apps.orders.models import Order


def test_can_cancel_returns_true_when_pending() -> None:
    order = Order(status=Order.Status.PENDING)
    assert order.can_cancel() is True


def test_cancel_raises_when_already_cancelled() -> None:
    order = Order(status=Order.Status.CANCELLED)
    with pytest.raises(DomainValidationError):
        order.cancel()

Integration test for a QuerySet method:

@pytest.mark.django_db
class TestOrderQuerySet:
    def test_for_user_returns_only_own_orders(self) -> None:
        user = UserFactory()
        own = OrderFactory(user=user)
        OrderFactory()  # another user's order

        result = list(Order.objects.for_user(user))

        assert result == [own]

    def test_active_excludes_cancelled(self) -> None:
        OrderFactory(status=Order.Status.PENDING)
        OrderFactory(status=Order.Status.CANCELLED)

        assert Order.objects.active().count() == 1

Assert query count to catch N+1 regressions:

@pytest.mark.django_db
def test_list_orders_no_n_plus_one(django_assert_num_queries) -> None:
    user = UserFactory()
    OrderFactory.create_batch(5, user=user)

    with django_assert_num_queries(2):  # 1 orders + 1 prefetch products
        orders = list(
            Order.objects.for_user(user).select_related("product")
        )
    assert len(orders) == 5