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,
        },
    }
}

# Engine yang tersedia:
# 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"

# Teks
models.CharField(max_length=200)         # string pendek, wajib max_length
models.TextField()                        # string panjang tanpa batas
models.SlugField(max_length=220)         # URL-friendly string
models.EmailField()                       # validasi format email otomatis
models.URLField()                         # validasi format URL

# Angka
models.IntegerField()                     # -2^31 s.d. 2^31-1
models.PositiveIntegerField()             # 0 s.d. 2^31-1
models.PositiveSmallIntegerField()        # 0 s.d. 32767
models.BigIntegerField()                  # untuk ID besar / timestamp
models.DecimalField(max_digits=15, decimal_places=2)  # uang/presisi tinggi
models.FloatField()                       # floating point (hindari untuk uang)

# Boolean
models.BooleanField(default=True)

# Waktu
models.DateField()                        # tanggal saja
models.TimeField()                        # waktu saja
models.DateTimeField()                    # tanggal + waktu
models.DateTimeField(auto_now_add=True)   # set otomatis saat create, tidak bisa diubah
models.DateTimeField(auto_now=True)       # update otomatis setiap save()
models.DurationField()                    # selang waktu (timedelta Python)

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

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

class Produk(models.Model):
    kategori = models.ForeignKey(
        Kategori,
        on_delete=models.SET_NULL,  # jika kategori dihapus, set null
        null=True,
        blank=True,
        related_name="produk"       # akses balik: kategori.produk.all()
    )

# Pilihan on_delete:
# CASCADE     -- hapus semua produk jika kategori dihapus
# PROTECT     -- cegah penghapusan jika masih ada produk (raise ProtectedError)
# SET_NULL    -- set null (butuh null=True)
# SET_DEFAULT -- set ke nilai default
# DO_NOTHING  -- tidak melakukan apa-apa (berbahaya, bisa IntegrityError di 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)

# Akses dua arah
pengguna = Pengguna.objects.get(pk=1)
pengguna.profil.bio   # via related_name (raise RelatedObjectDoesNotExist jika belum ada)
profil.pengguna.nama  # akses balik ke pengguna

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

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

produk.tag.add(tag_baru)             # tambah relasi
produk.tag.remove(tag_baru)          # hapus relasi
produk.tag.set([tag1, tag2])         # ganti semua (hapus lama, tambah baru)
produk.tag.clear()                   # hapus semua relasi
semua_tag = produk.tag.all()         # ambil semua tag terkait

# Through model -- M2M dengan data tambahan di tabel pivot
class Order(models.Model):
    produk = models.ManyToManyField(Produk, through="OrderItem")

# Dengan through model, gunakan model pivot langsung untuk CRUD
OrderItem.objects.create(order=order, produk=produk, jumlah=2, harga=produk.harga)

# Buat file migrasi dari perubahan model
python manage.py makemigrations

# Terapkan semua migrasi yang belum diterapkan
python manage.py migrate

# Lihat status semua migrasi
python manage.py showmigrations

# Lihat SQL yang akan dijalankan migrasi tertentu
python manage.py sqlmigrate myapp 0001

# Rollback ke migrasi tertentu
python manage.py migrate myapp 0003

# QuerySet belum dievaluasi (belum ke DB)
qs = Pengguna.objects.filter(aktif=True)

# Baru dievaluasi saat:
list(qs)              # konversi ke list
for p in qs: ...      # iterasi
bool(qs)              # cek keberadaan
len(qs)               # hitung (lebih baik pakai .count())
qs[0]                 # indexing

# Ambil satu objek
pengguna = Pengguna.objects.get(pk=1)        # DoesNotExist jika tidak ada

# ANTI-PATTERN: get() tanpa try/except
pengguna = Pengguna.objects.get(pk=999)      # ✗ -- crash jika tidak ada

# BENAR: gunakan filter().first() untuk menghindari exception
pengguna = Pengguna.objects.filter(pk=999).first()   # ✓ -- None jika tidak ada

# Operasi umum
ada     = Pengguna.objects.filter(email="[email protected]").exists()  # lebih efisien dari 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

# Filter pada relasi FK -- ikuti dengan __
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")  # relasi bertingkat

# Q Objects untuk kondisi OR dan NOT
Pengguna.objects.filter(
    Q(nama__icontains="budi") | Q(email__icontains="budi")  # OR
)
Pengguna.objects.filter(~Q(email__endswith="@spam.com"))     # NOT

# Kombinasi kompleks
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 -- ambil jika ada, buat jika belum
pengguna, dibuat = Pengguna.objects.get_or_create(
    email="[email protected]",
    defaults={"nama": "Budi Santoso", "usia": 28}
)

# update_or_create -- update jika ada, buat jika belum
pengguna, dibuat = Pengguna.objects.update_or_create(
    email="[email protected]",
    defaults={"nama": "Budi Santoso Wijaya", "usia": 29}
)

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

# ANTI-PATTERN: save() tanpa update_fields
pengguna.save()                        # ✗ -- UPDATE semua kolom, boros dan bisa race condition

# BENAR: sertakan update_fields untuk UPDATE kolom tertentu saja
pengguna.save(update_fields=["nama"])  # ✓ -- UPDATE pengguna SET nama=... WHERE id=1

# Bulk update -- lebih efisien untuk banyak baris
Pengguna.objects.filter(usia__lt=18).update(aktif=False)

# DELETE
Pengguna.objects.get(pk=1).delete()              # hapus satu
Pengguna.objects.filter(aktif=False).delete()    # hapus banyak

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

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

# BENAR: select_related JOIN dalam 1 query
produk_list = Produk.objects.filter(aktif=True).select_related("kategori")

for produk in produk_list:
    print(produk.kategori.nama)   # ✓ -- tidak ada query tambahan

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

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

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

# BENAR: prefetch_related -- 2 query total, digabung di Python
produk_list = Produk.objects.all().prefetch_related("tag")

for produk in produk_list:
    print(produk.tag.all())    # ✓ -- tidak ada query tambahan

# Kombinasi keduanya
Order.objects.select_related("pengguna").prefetch_related("items__produk")

# Prefetch dengan queryset kustom menggunakan 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"   # simpan ke atribut terpisah agar bisa difilter
    )
)

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

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

# Agregasi seluruh QuerySet -- kembalikan 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}

# Anotasi -- tambahkan nilai kalkulasi ke setiap objek QuerySet
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} order, 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 -- referensi kolom lain dalam query tanpa menariknya ke Python
from django.db.models import F

# Update harga semua produk naik 10% dalam satu query atomik
Produk.objects.filter(aktif=True).update(harga=F("harga") * 1.1)

# Filter berdasarkan perbandingan antar kolom
Produk.objects.filter(stok__lt=F("harga"))  # produk yang stoknya kurang dari harganya

# Ordering
Pengguna.objects.order_by("nama")            # ASC
Pengguna.objects.order_by("-nama")           # DESC
Pengguna.objects.order_by("usia", "-nama")   # multi-kolom
Pengguna.objects.order_by()                  # hapus default ordering dari Meta

# values() -- kembalikan dict, bukan model instance
Pengguna.objects.values("id", "nama", "email")
# <QuerySet [{'id': 1, 'nama': 'Budi', 'email': '...'}, ...]>

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

# values_list flat -- satu kolom sebagai flat list
email_list = list(Pengguna.objects.values_list("email", flat=True))
# ['[email protected]', '[email protected]', ...]

# only() dan defer() -- kontrol kolom yang di-load
Pengguna.objects.only("id", "nama", "email")   # load kolom ini saja
Pengguna.objects.defer("bio")                   # load semua kecuali bio

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

from django.db import transaction

# Decorator -- seluruh fungsi dalam satu transaksi
@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() -- kunci baris agar tidak diubah proses lain
        produk = Produk.objects.select_for_update().get(pk=item["produk_id"])

        if produk.stok < item["jumlah"]:
            raise ValueError(f"Stok {produk.nama} tidak mencukupi")

        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 -- transaksi dalam blok tertentu saja
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 -- transaksi bersarang, rollback parsial
def operasi_kompleks():
    with transaction.atomic():        # transaksi luar
        buat_order()

        try:
            with transaction.atomic():    # savepoint
                kirim_email_notifikasi()  # jika ini gagal...
        except Exception:
            pass                          # ... order tetap tersimpan (hanya email yang dibatalkan)

from django.db import connection

# Manager.raw() -- kembalikan 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)   # tetap objek Pengguna dengan semua method-nya

# connection.cursor() -- untuk query non-model (report, aggregasi kompleks)
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()


# Penggunaan -- chainable karena setiap method mengembalikan QuerySet
Pengguna.objects.aktif().dewasa().dengan_orders().order_by("-dibuat_pada")
Pengguna.objects.aktif().top_pembeli(limit=5)