Django ORM #

pip install django psycopg2-binary   # PostgreSQL
pip install django mysqlclient        # MySQL

# myproject/settings.py

DATABASES = {
    "default": {
        "ENGINE":   "django.db.backends.postgresql",
        "NAME":     "myapp",
        "USER":     "postgres",
        "PASSWORD": "",
        "HOST":     "localhost",
        "PORT":     "5432",
        "OPTIONS": {
            "connect_timeout": 10,
        },
    }
}

# Available engines:
# django.db.backends.postgresql
# django.db.backends.mysql
# django.db.backends.sqlite3
# django.db.backends.oracle

# myapp/models.py

from django.db import models

class Kategori(models.Model):
    nama = models.CharField(max_length=100, unique=True)
    slug = models.SlugField(max_length=120, unique=True)

    class Meta:
        db_table            = "kategori"
        ordering            = ["nama"]
        verbose_name_plural = "Kategori"

    def __str__(self):
        return self.nama


class Pengguna(models.Model):
    nama        = models.CharField(max_length=100)
    email       = models.EmailField(unique=True)
    usia        = models.PositiveSmallIntegerField(null=True, blank=True)
    aktif       = models.BooleanField(default=True)
    bio         = models.TextField(blank=True, default="")
    dibuat_pada = models.DateTimeField(auto_now_add=True)
    diubah_pada = models.DateTimeField(auto_now=True)

    class Meta:
        db_table = "pengguna"
        indexes  = [
            models.Index(fields=["email"]),
            models.Index(fields=["aktif", "dibuat_pada"]),
        ]

    def __str__(self):
        return f"{self.nama} <{self.email}>"


class Tag(models.Model):
    nama = models.CharField(max_length=50, unique=True)

    def __str__(self):
        return self.nama


class Produk(models.Model):
    nama      = models.CharField(max_length=200)
    slug      = models.SlugField(max_length=220, unique=True)
    harga     = models.DecimalField(max_digits=15, decimal_places=2)
    stok      = models.PositiveIntegerField(default=0)
    aktif     = models.BooleanField(default=True)
    kategori  = models.ForeignKey(
        Kategori,
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
        related_name="produk"
    )
    tag = models.ManyToManyField(Tag, blank=True, related_name="produk")
    dibuat_pada = models.DateTimeField(auto_now_add=True)

    class Meta:
        db_table = "produk"
        ordering = ["-dibuat_pada"]

    def __str__(self):
        return self.nama


class Order(models.Model):
    class StatusChoices(models.TextChoices):
        PENDING    = "pending",    "Menunggu"
        DIPROSES   = "diproses",   "Diproses"
        DIKIRIM    = "dikirim",    "Dikirim"
        SELESAI    = "selesai",    "Selesai"
        DIBATALKAN = "dibatalkan", "Dibatalkan"

    pengguna    = models.ForeignKey(Pengguna, on_delete=models.PROTECT, related_name="orders")
    produk      = models.ManyToManyField(Produk, through="OrderItem")
    status      = models.CharField(
        max_length=20,
        choices=StatusChoices.choices,
        default=StatusChoices.PENDING
    )
    total       = models.DecimalField(max_digits=15, decimal_places=2, default=0)
    dibuat_pada = models.DateTimeField(auto_now_add=True)

    class Meta:
        db_table = "orders"
        ordering = ["-dibuat_pada"]

    def __str__(self):
        return f"Order #{self.pk} — {self.pengguna}"


class OrderItem(models.Model):
    order  = models.ForeignKey(Order, on_delete=models.CASCADE, related_name="items")
    produk = models.ForeignKey(Produk, on_delete=models.PROTECT)
    jumlah = models.PositiveIntegerField(default=1)
    harga  = models.DecimalField(max_digits=15, decimal_places=2)

    class Meta:
        db_table = "order_items"

# Text
models.CharField(max_length=200)         # short string, max_length required
models.TextField()                        # unlimited-length string
models.SlugField(max_length=220)         # URL-friendly string
models.EmailField()                       # automatic email format validation
models.URLField()                         # URL format validation

# Numbers
models.IntegerField()                     # -2^31 to 2^31-1
models.PositiveIntegerField()             # 0 to 2^31-1
models.PositiveSmallIntegerField()        # 0 to 32767
models.BigIntegerField()                  # for large IDs / timestamps
models.DecimalField(max_digits=15, decimal_places=2)  # money/high precision
models.FloatField()                       # floating point (avoid for money)

# Boolean
models.BooleanField(default=True)

# Time
models.DateField()                        # date only
models.TimeField()                        # time only
models.DateTimeField()                    # date + time
models.DateTimeField(auto_now_add=True)   # set automatically on create, immutable
models.DateTimeField(auto_now=True)       # updated automatically on every save()
models.DurationField()                    # time span (Python timedelta)

# File & Media
models.FileField(upload_to="files/")
models.ImageField(upload_to="images/")   # needs Pillow

# Others
models.JSONField()                        # native JSON (Django 3.1+)
models.UUIDField(default=uuid.uuid4, editable=False)

class Produk(models.Model):
    kategori = models.ForeignKey(
        Kategori,
        on_delete=models.SET_NULL,  # if the category is deleted, set null
        null=True,
        blank=True,
        related_name="produk"       # reverse access: kategori.produk.all()
    )

# on_delete choices:
# CASCADE     -- delete all products if the category is deleted
# PROTECT     -- prevent deletion if products still exist (raises ProtectedError)
# SET_NULL    -- set null (needs null=True)
# SET_DEFAULT -- set to the default value
# DO_NOTHING  -- do nothing (dangerous, can cause IntegrityError in the DB)

class Profil(models.Model):
    pengguna = models.OneToOneField(
        Pengguna,
        on_delete=models.CASCADE,
        related_name="profil"
    )
    bio     = models.TextField(blank=True)
    kota    = models.CharField(max_length=100, blank=True)
    website = models.URLField(blank=True)

# Two-way access
pengguna = Pengguna.objects.get(pk=1)
pengguna.profil.bio   # via related_name (raises RelatedObjectDoesNotExist if missing)
profil.pengguna.nama  # reverse access to the user

class Produk(models.Model):
    tag = models.ManyToManyField(Tag, blank=True, related_name="produk")

# M2M operations
produk    = Produk.objects.get(pk=1)
tag_baru  = Tag.objects.get(nama="python")

produk.tag.add(tag_baru)             # add a relationship
produk.tag.remove(tag_baru)          # remove a relationship
produk.tag.set([tag1, tag2])         # replace all (remove old, add new)
produk.tag.clear()                   # remove all relationships
semua_tag = produk.tag.all()         # fetch all related tags

# Through model -- M2M with extra data in the pivot table
class Order(models.Model):
    produk = models.ManyToManyField(Produk, through="OrderItem")

# With a through model, use the pivot model directly for CRUD
OrderItem.objects.create(order=order, produk=produk, jumlah=2, harga=produk.harga)

# Create migration files from model changes
python manage.py makemigrations

# Apply all pending migrations
python manage.py migrate

# See the status of all migrations
python manage.py showmigrations

# See the SQL that a specific migration will run
python manage.py sqlmigrate myapp 0001

# Roll back to a specific migration
python manage.py migrate myapp 0003

# The QuerySet isn't evaluated yet (not hitting the DB)
qs = Pengguna.objects.filter(aktif=True)

# Evaluated only when:
list(qs)              # converted to a list
for p in qs: ...      # iterated
bool(qs)              # existence check
len(qs)               # counted (better to use .count())
qs[0]                 # indexed

# Fetch a single object
pengguna = Pengguna.objects.get(pk=1)        # DoesNotExist if missing

# ANTI-PATTERN: get() without try/except
pengguna = Pengguna.objects.get(pk=999)      # ✗ -- crashes if missing

# CORRECT: use filter().first() to avoid the exception
pengguna = Pengguna.objects.filter(pk=999).first()   # ✓ -- None if missing

# Common operations
ada     = Pengguna.objects.filter(email="[email protected]").exists()  # more efficient than count() > 0
jumlah  = Pengguna.objects.filter(aktif=True).count()
halaman = Pengguna.objects.all()[10:20]      # LIMIT 10 OFFSET 10

from django.db.models import Q

# Field lookups
Pengguna.objects.filter(nama__icontains="budi")     # LIKE '%budi%' case-insensitive
Pengguna.objects.filter(nama__startswith="Budi")    # LIKE 'Budi%'
Pengguna.objects.filter(usia__gt=25)                # usia > 25
Pengguna.objects.filter(usia__gte=25)               # usia >= 25
Pengguna.objects.filter(usia__range=(20, 30))       # BETWEEN 20 AND 30
Pengguna.objects.filter(usia__in=[25, 28, 32])      # IN (25, 28, 32)
Pengguna.objects.filter(usia__isnull=True)          # IS NULL
Pengguna.objects.exclude(aktif=True)                # WHERE aktif != TRUE

# Filtering on FK relationships -- follow with __
Produk.objects.filter(kategori__nama="Elektronik")
Produk.objects.filter(kategori__nama__icontains="elektr")
Order.objects.filter(pengguna__email="[email protected]")
Order.objects.filter(items__produk__nama__icontains="laptop")  # nested relationships

# Q Objects for OR and NOT conditions
Pengguna.objects.filter(
    Q(nama__icontains="budi") | Q(email__icontains="budi")  # OR
)
Pengguna.objects.filter(~Q(email__endswith="@spam.com"))     # NOT

# Complex combinations
Pengguna.objects.filter(
    Q(aktif=True) & (
        Q(nama__icontains="budi") | Q(email__icontains="budi")
    )
)

# CREATE
pengguna = Pengguna.objects.create(nama="Budi", email="[email protected]", usia=28)

# get_or_create -- fetch if it exists, create if not
pengguna, dibuat = Pengguna.objects.get_or_create(
    email="[email protected]",
    defaults={"nama": "Budi Santoso", "usia": 28}
)

# update_or_create -- update if it exists, create if not
pengguna, dibuat = Pengguna.objects.update_or_create(
    email="[email protected]",
    defaults={"nama": "Budi Santoso Wijaya", "usia": 29}
)

# UPDATE -- one object
pengguna = Pengguna.objects.get(pk=1)
pengguna.nama = "Budi Wijaya"

# ANTI-PATTERN: save() without update_fields
pengguna.save()                        # ✗ -- UPDATEs all columns, wasteful and can race

# CORRECT: include update_fields to UPDATE only specific columns
pengguna.save(update_fields=["nama"])  # ✓ -- UPDATE pengguna SET nama=... WHERE id=1

# Bulk update -- more efficient for many rows
Pengguna.objects.filter(usia__lt=18).update(aktif=False)

# DELETE
Pengguna.objects.get(pk=1).delete()              # delete one
Pengguna.objects.filter(aktif=False).delete()    # delete many

# ANTI-PATTERN: N+1 for ForeignKey
produk_list = Produk.objects.filter(aktif=True)  # 1 query

for produk in produk_list:
    print(produk.kategori.nama)   # ✗ -- 1 NEW query per product (N+1!)

# CORRECT: select_related JOINs in 1 query
produk_list = Produk.objects.filter(aktif=True).select_related("kategori")

for produk in produk_list:
    print(produk.kategori.nama)   # ✓ -- no extra queries

# Nested
Order.objects.select_related("pengguna", "pengguna__profil")

# ANTI-PATTERN: N+1 for ManyToMany
produk_list = Produk.objects.all()

for produk in produk_list:
    print(produk.tag.all())    # ✗ -- 1 query per product

# CORRECT: prefetch_related -- 2 queries total, merged in Python
produk_list = Produk.objects.all().prefetch_related("tag")

for produk in produk_list:
    print(produk.tag.all())    # ✓ -- no extra queries

# Combining both
Order.objects.select_related("pengguna").prefetch_related("items__produk")

# Prefetching with a custom queryset using Prefetch()
from django.db.models import Prefetch

produk_list = Produk.objects.prefetch_related(
    Prefetch(
        "tag",
        queryset=Tag.objects.order_by("nama"),
        to_attr="tag_terurut"   # store in a separate attribute so it can be filtered
    )
)

for produk in produk_list:
    for tag in produk.tag_terurut:  # a list, not a QuerySet
        print(tag.nama)

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

# Aggregating the whole QuerySet -- returns a dict
statistik = Pengguna.objects.filter(aktif=True).aggregate(
    total     = Count("id"),
    rata_usia = Avg("usia"),
    usia_min  = Min("usia"),
    usia_max  = Max("usia"),
)
# {'total': 50, 'rata_usia': 27.4, 'usia_min': 18, 'usia_max': 45}

# Annotation -- add computed values to each QuerySet object
pengguna_list = Pengguna.objects.annotate(
    jumlah_order  = Count("orders"),
    total_belanja = Coalesce(Sum("orders__total"), 0)
).filter(aktif=True).order_by("-total_belanja")

for p in pengguna_list:
    print(f"{p.nama}: {p.jumlah_order} orders, total Rp{p.total_belanja:,.0f}")

# Group by -- values() + annotate()
Order.objects.values("status").annotate(
    jumlah = Count("id")
).order_by("status")
# [{'status': 'pending', 'jumlah': 5}, {'status': 'selesai', 'jumlah': 12}]

# F Expression -- reference another column in a query without pulling it into Python
from django.db.models import F

# Raise all active product prices by 10% in one atomic query
Produk.objects.filter(aktif=True).update(harga=F("harga") * 1.1)

# Filter based on a column-to-column comparison
Produk.objects.filter(stok__lt=F("harga"))  # products whose stock is less than their price

# Ordering
Pengguna.objects.order_by("nama")            # ASC
Pengguna.objects.order_by("-nama")           # DESC
Pengguna.objects.order_by("usia", "-nama")   # multi-column
Pengguna.objects.order_by()                  # remove the default ordering from Meta

# values() -- returns dicts, not model instances
Pengguna.objects.values("id", "nama", "email")
# <QuerySet [{'id': 1, 'nama': 'Budi', 'email': '...'}, ...]>

# values_list() -- returns tuples
Pengguna.objects.values_list("id", "nama")
# <QuerySet [(1, 'Budi'), (2, 'Sari'), ...]>

# values_list flat -- one column as a flat list
email_list = list(Pengguna.objects.values_list("email", flat=True))
# ['[email protected]', '[email protected]', ...]

# only() and defer() -- control which columns are loaded
Pengguna.objects.only("id", "nama", "email")   # load only these columns
Pengguna.objects.defer("bio")                   # load everything except bio

# Distinct
Produk.objects.values("kategori_id").distinct()

from django.db import transaction

# Decorator -- the whole function in one transaction
@transaction.atomic
def proses_order(pengguna_id: int, items: list[dict]) -> Order:
    pengguna = Pengguna.objects.get(pk=pengguna_id)
    order    = Order.objects.create(pengguna=pengguna, total=0)
    total    = 0

    for item in items:
        # select_for_update() -- lock the row so other processes can't change it
        produk = Produk.objects.select_for_update().get(pk=item["produk_id"])

        if produk.stok < item["jumlah"]:
            raise ValueError(f"Insufficient stock for {produk.nama}")

        produk.stok -= item["jumlah"]
        produk.save(update_fields=["stok"])

        subtotal = produk.harga * item["jumlah"]
        OrderItem.objects.create(
            order=order, produk=produk,
            jumlah=item["jumlah"], harga=produk.harga
        )
        total += subtotal

    order.total = total
    order.save(update_fields=["total"])
    return order

# Context manager -- transaction only within a specific block
def update_status_order(order_id: int, status_baru: str):
    with transaction.atomic():
        order = Order.objects.select_for_update().get(pk=order_id)
        order.status = status_baru
        order.save(update_fields=["status"])

# Savepoint -- nested transactions, partial rollback
def operasi_kompleks():
    with transaction.atomic():        # outer transaction
        buat_order()

        try:
            with transaction.atomic():    # savepoint
                kirim_email_notifikasi()  # if this fails...
        except Exception:
            pass                          # ... the order is still saved (only the email is rolled back)

from django.db import connection

# Manager.raw() -- returns model instances
pengguna_list = Pengguna.objects.raw(
    "SELECT * FROM pengguna WHERE usia > %s ORDER BY nama",
    [25]
)
for p in pengguna_list:
    print(p.nama)   # still Pengguna objects with all their methods

# connection.cursor() -- for non-model queries (reports, complex aggregations)
def laporan_penjualan_bulanan() -> list[dict]:
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT
                DATE_TRUNC('month', o.dibuat_pada) AS bulan,
                COUNT(o.id)                         AS jumlah_order,
                SUM(o.total)                        AS total_pendapatan,
                COUNT(DISTINCT o.pengguna_id)       AS pengguna_unik
            FROM orders o
            WHERE o.status = %s
            GROUP BY bulan
            ORDER BY bulan DESC
            LIMIT 12
        """, ["selesai"])

        kolom = [desc[0] for desc in cursor.description]
        return [dict(zip(kolom, baris)) for baris in cursor.fetchall()]

from django.db import models

class PenggunaQuerySet(models.QuerySet):
    def aktif(self):
        return self.filter(aktif=True)

    def dewasa(self):
        return self.filter(usia__gte=18)

    def dengan_orders(self):
        return self.annotate(
            jumlah_order=models.Count("orders")
        ).filter(jumlah_order__gt=0)

    def top_pembeli(self, limit: int = 10):
        return (
            self.annotate(total_belanja=models.Sum("orders__total"))
            .order_by("-total_belanja")[:limit]
        )


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

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


class Pengguna(models.Model):
    # ... fields ...
    objects = PenggunaManager()


# Usage -- chainable because every method returns a QuerySet
Pengguna.objects.aktif().dewasa().dengan_orders().order_by("-dibuat_pada")
Pengguna.objects.aktif().top_pembeli(limit=5)