İleri Django ve DRF: ORM optimizasyonu, izinler ve test

Neden önemli?

Django'ya giriş dersinde proje kurmayı, modeli, migration'ı, admin'i ve temel ORM sorgularını gördün. İş ilanlarında "Django + DRF deneyimi" yazdığında beklenen bundan fazlasıdır: liste sayfası 400 sorgu attığında bunu ölçüp bir sorguya indirmek, izin onayı yarıda kalınca veritabanını tutarlı bırakmak, API'de doğrulama, sayfalama, filtreleme ve nesne düzeyinde izinleri doğru kurmak, kodu testlerle güvenceye almak. Bu derste Pusula İK'nın izin onay API'sini bu gözle tasarlayacağız.

Model ilişkileri, özel Manager ve QuerySet

Pusula İK'nın çekirdeği üç modeldir: Calisan kendi yöneticisine (aynı tabloya) bağlıdır, Izin bir çalışana bağlıdır. Tekrarlanan süzgeçleri her view'da yazmak yerine özel QuerySet metotlarına koy; zincirlenebilir kalırlar ve iş dilini koda taşırlar.

# izin/models.py
from django.db import models
from django.db.models import Q

class IzinQuerySet(models.QuerySet):
    def bekleyen(self):
        return self.filter(durum="beklemede")

    def gorunur(self, kullanici):                     # nesne düzeyinde yetki tek yerde
        c = kullanici.calisan
        return self.filter(Q(calisan=c) | Q(calisan__yonetici=c))

class Calisan(models.Model):
    ad = models.CharField(max_length=120)
    yonetici = models.ForeignKey("self", null=True, on_delete=models.SET_NULL, related_name="ekip")
    kullanici = models.OneToOneField("auth.User", on_delete=models.CASCADE)

class Izin(models.Model):
    TUR = [("yıllık", "Yıllık"), ("hastalık", "Hastalık"), ("mazeret", "Mazeret"), ("ücretsiz", "Ücretsiz")]
    calisan = models.ForeignKey(Calisan, on_delete=models.CASCADE, related_name="izinler")
    tur = models.CharField(max_length=20, choices=TUR)
    baslangic = models.DateField()
    gun = models.PositiveSmallIntegerField()
    durum = models.CharField(max_length=20, default="beklemede", db_index=True)

    objects = IzinQuerySet.as_manager()

# kullanım: Izin.objects.gorunur(request.user).bekleyen().order_by("baslangic")

ForeignKey çoğa-bir, OneToOneField bire-bir, ManyToManyField çoğa-çok ilişkidir; related_name ters yönün adıdır (yonetici.ekip.all()). on_delete seçimi bir iş kararıdır: CASCADE çalışan silinince izinleri de siler, PROTECT silmeyi engeller, SET_NULL bağı koparır.

Sorgu sayısını ölç: N+1 sorunu

QuerySet tembeldir ve ilişkili nesneye her erişim, önceden getirilmemişse yeni bir sorgu atar. 100 izinlik listede her satırda izin.calisan.ad yazdırmak 1 + 100 sorgu demektir. Önce ölç: Django'da django-debug-toolbar, testte assertNumQueries ya da connection.queries. sqlite3'te aynı ölçümü set_trace_callback ile yapabilirsin; aşağıdaki kutu gerçek Pusula İK verisinde üç yolu karşılaştırıyor.

import veri

con = veri.baglan("pusula_ik", tablolar=["ik_employees", "ik_leaves"])
sorgular = []
con.set_trace_callback(sorgular.append)

def n_arti_1(limit):                                     # for izin in Izin.objects.all()[:limit]: izin.calisan.ad
    izinler = con.execute("SELECT leave_id, emp_id FROM ik_leaves ORDER BY leave_id LIMIT ?", (limit,)).fetchall()
    return [(lid, con.execute("SELECT full_name FROM ik_employees WHERE emp_id = ?", (e,)).fetchone()[0])
            for lid, e in izinler]

def select_related(limit):                               # Izin.objects.select_related("calisan")
    return con.execute("""SELECT l.leave_id, e.full_name FROM ik_leaves l
                          JOIN ik_employees e ON e.emp_id = l.emp_id
                          ORDER BY l.leave_id LIMIT ?""", (limit,)).fetchall()

def prefetch_related(limit):                             # ikinci sorgu: WHERE emp_id IN (...)
    izinler = con.execute("SELECT leave_id, emp_id FROM ik_leaves ORDER BY leave_id LIMIT ?", (limit,)).fetchall()
    idler = sorted({e for _, e in izinler})
    yer = ",".join("?" * len(idler))
    ad = dict(con.execute(f"SELECT emp_id, full_name FROM ik_employees WHERE emp_id IN ({yer})", idler))
    return [(lid, ad[e]) for lid, e in izinler]

for f in (n_arti_1, select_related, prefetch_related):
    sorgular.clear()
    sonuc = f(100)
    print(f"{f.__name__:17} sorgu={len(sorgular):3}  ilk={sonuc[0]}")

annotate, aggregate, F ve Q

aggregate tüm kümeyi tek değere indirir ({"toplam": 1234}); annotate her satıra (ya da values() ile gruplanan her gruba) hesaplanmış bir alan ekler, yani SQL'deki GROUP BY'dır. Q nesneleri | (VEYA), & (VE) ve ~ (DEĞİL) ile birleşir; F bir sütuna Python'a çekmeden başvurur ve yarış durumunu önler.

from django.db.models import Count, Sum, Q, F

(Izin.objects
    .filter(baslangic__year=2024)
    .values("calisan__pozisyon__bolum__ad")
    .annotate(talep=Count("id"),
              gun=Sum("gun"),
              hastalik_gun=Sum("gun", filter=Q(tur="hastalık")))
    .order_by("-gun"))

Calisan.objects.filter(pk=418).update(izin_bakiyesi=F("izin_bakiyesi") - 3)   # UPDATE ... SET x = x - 3

import veri

con = veri.baglan("pusula_ik", tablolar=["ik_employees", "ik_leaves", "ik_positions", "ik_departments"])
satirlar = con.execute("""
    SELECT d.name, COUNT(*) AS talep, SUM(l.days) AS gun,
           SUM(CASE WHEN l.leave_type = 'hastalık' THEN l.days ELSE 0 END) AS hastalik_gun
    FROM ik_leaves l
    JOIN ik_employees e   ON e.emp_id = l.emp_id
    JOIN ik_positions p   ON p.position_id = e.position_id
    JOIN ik_departments d ON d.dept_id = p.dept_id
    WHERE l.start_date >= '2024-01-01'
    GROUP BY d.name ORDER BY gun DESC LIMIT 4""").fetchall()
for ad, talep, gun, hasta in satirlar:
    print(f"{ad:22} talep={talep:4} gün={gun:5} hastalık payı=%{100 * hasta / gun:.0f}")

Sum("gun", filter=Q(...)) SQL'de SUM(...) FILTER (WHERE ...) ya da CASE WHEN olarak çıkar. Çift yönlü alt çizgi (calisan__pozisyon__bolum__ad) her adımda bir JOIN ekler.

Sinyaller ve transaction.atomic

Sinyaller (post_save, pre_delete) bir model kaydedildiğinde başka bir kodu tetikler. Kolay görünür ama gizli yan etki üretir: kaydın nereden e-posta gönderdiğini bulmak zorlaşır, bulk_create ve update() sinyal tetiklemez, testler yavaşlar. Kural: kendi uygulamanın içinde açık bir servis fonksiyonu çağır; sinyali yalnız değiştiremediğin üçüncü taraf uygulamalara tepki için kullan.

Birden çok yazma işlemi "ya hep ya hiç" olmalıdır. İzin onayı hem talebin durumunu değiştirir hem bakiyeyi düşürür; ikincisi başarısız olursa birincisi geri alınmalıdır.

from django.db import transaction

def izni_onayla(talep_id, onaylayan):
    with transaction.atomic():
        talep = Izin.objects.select_for_update().get(pk=talep_id)    # satırı kilitle
        if talep.durum != "beklemede":
            raise ValueError("zaten işlenmiş")
        talep.durum = "onaylandi"
        talep.save(update_fields=["durum"])
        Calisan.objects.filter(pk=talep.calisan_id).update(izin_bakiyesi=F("izin_bakiyesi") - talep.gun)
        transaction.on_commit(lambda: bildirim_gonder.delay(talep_id))   # Celery: commit SONRASI

import sqlite3

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE talep (id INTEGER PRIMARY KEY, durum TEXT)")
con.execute("CREATE TABLE bakiye (emp_id INTEGER PRIMARY KEY, gun INTEGER CHECK (gun >= 0))")
con.execute("INSERT INTO talep VALUES (1, 'beklemede')")
con.execute("INSERT INTO bakiye VALUES (418, 2)")
con.commit()

try:
    with con:                                     # transaction.atomic karşılığı
        con.execute("UPDATE talep SET durum = 'onaylandi' WHERE id = 1")
        con.execute("UPDATE bakiye SET gun = gun - 5 WHERE emp_id = 418")   # CHECK ihlali
except sqlite3.IntegrityError as e:
    print("geri alındı:", e)
print(con.execute("SELECT durum FROM talep").fetchone(), con.execute("SELECT gun FROM bakiye").fetchone())

with con: blok hatasız biterse commit, hata çıkarsa rollback yapar; talep "beklemede" kaldı. transaction.on_commit önemlidir: Celery görevini atomic bloğun içinde doğrudan çağırırsan işçi, henüz commit edilmemiş kaydı bulamayabilir.

Django REST Framework: serializer'dan router'a

DRF'te bir isteğin yolu: router URL'i ViewSet eylemine eşler → kimlik doğrulama → izin sınıfları (has_permission) → throttling → get_queryset ve filtreler → nesne alınırsa has_object_permission → serializer doğrulaması → kaydetme → sayfalanmış yanıt.

# izin/api.py
from rest_framework import serializers, viewsets, permissions, decorators, response, status
from django_filters.rest_framework import DjangoFilterBackend

class IzinSerializer(serializers.ModelSerializer):
    calisan_ad = serializers.CharField(source="calisan.ad", read_only=True)

    class Meta:
        model = Izin
        fields = ["id", "calisan_ad", "tur", "baslangic", "gun", "durum"]
        read_only_fields = ["durum"]

    def validate_gun(self, deger):                       # alan düzeyi
        if not 1 <= deger <= 30:
            raise serializers.ValidationError("1 ile 30 gün arası olmalı.")
        return deger

    def validate(self, veri):                            # nesne düzeyi (alanlar arası)
        if veri["tur"] == "hastalık" and veri["gun"] > 10:
            raise serializers.ValidationError("10 günden uzun hastalık izni rapor ister.")
        return veri

class YoneticiMi(permissions.BasePermission):
    def has_object_permission(self, request, view, obj):
        if request.method in permissions.SAFE_METHODS:
            return True
        return obj.calisan.yonetici_id == request.user.calisan.id

class IzinViewSet(viewsets.ModelViewSet):
    serializer_class = IzinSerializer
    permission_classes = [permissions.IsAuthenticated, YoneticiMi]
    filter_backends = [DjangoFilterBackend]
    filterset_fields = ["tur", "durum"]
    throttle_scope = "izin"

    def get_queryset(self):
        return Izin.objects.gorunur(self.request.user).select_related("calisan")

    def perform_create(self, serializer):
        serializer.save(calisan=self.request.user.calisan)

    @decorators.action(detail=True, methods=["post"])
    def onayla(self, request, pk=None):
        izin = self.get_object()                         # has_object_permission burada çalışır
        izni_onayla(izin.pk, request.user)
        return response.Response({"durum": "onaylandi"}, status=status.HTTP_200_OK)

# urls.py
# router = routers.DefaultRouter(); router.register("izinler", IzinViewSet, basename="izin")
# settings.py
# REST_FRAMEWORK = {"DEFAULT_PAGINATION_CLASS": "rest_framework.pagination.PageNumberPagination",
#                   "PAGE_SIZE": 20,
#                   "DEFAULT_THROTTLE_CLASSES": ["rest_framework.throttling.ScopedRateThrottle"],
#                   "DEFAULT_THROTTLE_RATES": {"izin": "60/min"}}

Serializer'ın doğrulama sırası: her alan için tip dönüşümü ve alan doğrulayıcıları → validate_<alan> → tüm alanlar geçerliyse validate(). Hatalar {"alan": ["mesaj"], "non_field_errors": [...]} biçiminde toplanır ve 400 ile döner.

import datetime as dt

TURLER = {"yıllık", "hastalık", "mazeret", "ücretsiz"}

def dogrula(girdi):
    hatalar, temiz = {}, {}
    for alan, cevir in (("tur", str), ("baslangic", dt.date.fromisoformat), ("gun", int)):
        if alan not in girdi:
            hatalar.setdefault(alan, []).append("Bu alan zorunlu.")
            continue
        try:
            temiz[alan] = cevir(girdi[alan])
        except (TypeError, ValueError):
            hatalar.setdefault(alan, []).append("Geçersiz değer.")
    if "tur" in temiz and temiz["tur"] not in TURLER:
        hatalar.setdefault("tur", []).append(f"{temiz['tur']!r} geçerli bir seçim değil.")
    if not hatalar and temiz["tur"] == "hastalık" and temiz["gun"] > 10:
        hatalar["non_field_errors"] = ["10 günden uzun hastalık izni rapor ister."]
    return (temiz, {}) if not hatalar else ({}, hatalar)

print(dogrula({"tur": "yıllık", "baslangic": "2025-07-01", "gun": "5"}))
print(dogrula({"tur": "tatil", "baslangic": "2025-13-01"}))
print(dogrula({"tur": "hastalık", "baslangic": "2025-02-03", "gun": 14}))

Ayarlar, yönetim komutları ve test

Tek bir settings.py ile geliştirme ve üretimi yönetmek hataya açıktır. Yaygın yapı: settings/base.py (ortak), dev.py ve prod.py (from .base import * + farklar); hangisinin yükleneceğini DJANGO_SETTINGS_MODULE belirler, sırlar ortam değişkeninden okunur (Docker dersindeki 12-factor ilkesi). Tekrarlanan bakım işleri (gece bakiye hesabı, eski kayıt temizliği) için özel yönetim komutu yaz:

# izin/management/commands/bakiye_yenile.py   ->   python manage.py bakiye_yenile --yil 2025
from django.core.management.base import BaseCommand

class Command(BaseCommand):
    help = "Yıllık izin bakiyelerini yeniden hesaplar"

    def add_arguments(self, parser):
        parser.add_argument("--yil", type=int, required=True)

    def handle(self, *args, yil, **kw):
        n = bakiye_hesapla(yil)
        self.stdout.write(self.style.SUCCESS(f"{n} çalışanın bakiyesi güncellendi"))

# tests/test_izin_api.py      ->  pip install pytest-django factory-boy ; pytest
import factory, pytest
from rest_framework.test import APIClient

class CalisanFactory(factory.django.DjangoModelFactory):
    class Meta:
        model = Calisan
    ad = factory.Faker("name", locale="tr_TR")
    kullanici = factory.SubFactory(UserFactory)

@pytest.mark.django_db
def test_yonetici_ekibinin_iznini_onaylar(django_assert_num_queries):
    yonetici = CalisanFactory()
    izin = IzinFactory(calisan__yonetici=yonetici)
    api = APIClient(); api.force_authenticate(yonetici.kullanici)
    r = api.post(f"/api/izinler/{izin.pk}/onayla/")
    assert r.status_code == 200
    izin.refresh_from_db(); assert izin.durum == "onaylandi"

@pytest.mark.django_db
def test_baskasi_onaylayamaz():
    izin = IzinFactory()
    api = APIClient(); api.force_authenticate(CalisanFactory().kullanici)
    assert api.post(f"/api/izinler/{izin.pk}/onayla/").status_code == 404   # görünmüyor

Factory'ler test verisini tek satırda, ilişkileriyle birlikte üretir; sabit JSON fixture'larından daha bakımı kolaydır. @pytest.mark.django_db her testi kendi transaction'ında çalıştırıp sonunda geri alır.

Celery, statik ve medya dosyaları

E-posta, PDF rapor, toplu içe aktarma gibi uzun işler isteğin içinde yapılmaz; Celery görevine verilir ve transaction.on_commit ile commit'ten sonra kuyruğa atılır. Broker, yeniden deneme ve ölü mektup kuyruğu ayrıntılarını Redis ve kuyruk dersinde göreceksin. Statik dosyalar (CSS, JS) collectstatic ile tek klasöre toplanır ve WhiteNoise ya da CDN ile sunulur; kullanıcı yüklemeleri (medya, ör. hastalık raporu) ise MEDIA_ROOT ya da S3 benzeri nesne deposuna gider ve asla uygulama sunucusunun kod klasöründe tutulmaz.