Django Mülakat Soruları

Django Mülakat Soruları: 5 Gerçek Case ile Bu Caseler için Mantıklı Çözüm Yöntemleri 

 

Seviye: Mid - Senior 

Django & Python mülakatlarında ve genel olarak projeler de sık sık karşılaşılan gerçek hayat caseleri ni ele aldığımız bu yazıda ele alınan her case için farklı bakış açıları ile uygulanmış yöntemleri ve bu yöntemler için oluşturulmuş kod örneklerini görebilirsiniz.

Temel Konular Şöyledir:

Django Python ORM Transactions Celery Caching Security

Bu sorular hangi seviyelerde daha çok çıkar?

Öncelikle eğer bir Junior iseniz bu tarz sorularla karşılaşmamanız lazımdır fakat işin seviyesinden bağımsız her pythonista ve django severin bu case ler için bu yöntemlere hakim olması onlara ileri seviye case lerde de çok şey katacaktır.

• Mid-level Backend Developer: Performans ve temel ölçekleme stratejileri, ORM ve cache sorunları

 

• Senior Backend / Tech Lead: Race conditions, transaction yönetimleri, idempotency, servis güvenilirliği, Lock mekanizması,  Celery ve Taskların doğru yönetimi

 

• Fullstack: Django based bir fullstack developer için API performansı, view/cache stratejileri ve genel olarak güvenlik diyebiliriz.

 

Bu case ler neden önemli?

Proje Konusu

Ele alacağımız proje bir doktor hasta randevu sistemi burda
Doctor, Patient, Appointment gibi genel olarak bir doktor sitesinde olması gereken modeller var ve bu modellerin birbiriyle one to many şeklinde ilişkilendirildğini farzedelim.

Case1 ORM Performansı: “Her doktorun randevu sayısını tek sorguda getir”

Problem

Doctor modelimizin one to many (1->N) ilişkiyle bağlı olduğu bir Appointment modelimiz mevcut. Doktorları API da listelerken onlara ait randevu sayılarını da burada göstermek istiyorsun. N+1 problemi yaşamadan bu sorunu nasıl çözebilirsin ? 

Best Practises:

1. annotate(Count(...)) ile tek SQL Sorgusu 

from django.db.models import Count, Q

qs = (
    Doctor.objects
    .annotate(appointment_count=Count('appointments'))
)

# Bonus olarak appointmenleri filtreleyedebiliriz:
qs = Doctor.objects.annotate(
    upcoming_count=Count('appointments', filter=Q(appointments__status='TAKEN'))
)

İpucu: Duplicate riskini önlemek için Count('appointments', distinct=True) kullanabilirsin.

Case 2 — Race Condition & Transaction: “Stok düş, negatif olmasın”

Problem

Doktorumuzun ilaç sattığını düşünelim. Medicine modelimiz için doktorumuz bir kampanya oluşturmuş ve insanlar akın akın siteye geliyor.(E-ticaret işi biliyorum ama başka türlü analatamadım :D) Bu koşullar altında Medicine.stock her seferinde 1 azalmalı ve asla negatif olmamalı bu durumu nasıl yaparız ?

Uygulanabilir Yöntemler

1. Lock + transaction.atomic() + select_for_update()

from django.db import transaction
from django.db.models import F
from django.core.exceptions import ValidationError

def reserve_stock(medicine_id: int, qty: int = 1):
    with transaction.atomic(): # ACID Prensiplerinin A sı Atomic
        product = Medicine.objects.select_for_update().get(pk=product_id)
        if medicine.stock < qty:
            raise ValidationError("Stok yetersiz lütfen kontrol edin!")
        medicine.stock = F('stock') - qty
        medicine.save(update_fields=['stock'])

2. Koşullu tek UPDATE (kilitsiz, yüksek trafik)

from django.db import transaction
from django.db.models import F

def reserve_stock_fast(medicine_id: int, qty: int = 1) -> bool:
    with transaction.atomic():
        updated = Medicine.objects.filter(pk=medicine_id, stock__gte=qty)\
                                 .update(stock=F('stock') - qty)
        return updated == 1

Bonus: Model Üzerinden Kontrol

from django.db.models import CheckConstraint, Q

class Medicine(models.Model):
    stock = models.IntegerField(default=0)
    class Meta:
        constraints = [
            CheckConstraint(check=Q(stock__gte=0), name='stock_non_negative'),
        ]

Case 3 — Celery Sağlamlığı: “429/5xx, timeout ve çift kayda karşı idempotent görev”

Problem

Dış API’den Instagram postlarını çekip DB’ye yazıyorsun. Zaman zaman 429/5xx, timeout ve tekrar eden payload problemi var.

Uygulanabilir Yöntemler

• Retry + exponential backoff + jitter
• Idempotent kayıt: unique(external_id) + update_or_create
• Lock (Redis/cache.add) + time limit + acks_late

# tasks.py
import requests
from celery import shared_task
from django.db import transaction, IntegrityError
from django.core.cache import cache
from requests import HTTPError, Timeout
from instagram.models import InstagramPost

RETRYABLE = (HTTPError, Timeout)

def _fetch(url, token, cursor=None):
    params = {"access_token": token}
    if cursor: params["after"] = cursor
    r = requests.get(url, params=params, timeout=15)
    if r.status_code == 429:
        raise Timeout("Rate limited")
    r.raise_for_status()
    return r.json()

def _lock_key(account_id): return f"ig_sync_lock:{account_id}"

@shared_task(bind=True, acks_late=True,
             autoretry_for=RETRYABLE,
             retry_kwargs={"max_retries": 5, "countdown": 10},
             retry_backoff=True, retry_jitter=True,
             time_limit=300, soft_time_limit=240)
def sync_ig(self, account_id, url, token, cursor=None):
    if not cache.add(_lock_key(account_id), "1", timeout=120):
        raise self.retry(countdown=15)
    try:
        payload = _fetch(url, token, cursor)
        with transaction.atomic():
            for item in payload.get("data", []):
                ext = item["id"]
                defaults = {"caption": item.get("caption",""),
                            "taken_at": item.get("timestamp"),
                            "media_url": item.get("media_url")}
                InstagramPost.objects.update_or_create(external_id=ext, defaults=defaults)
        next_cur = payload.get("paging", {}).get("cursors", {}).get("after")
        if next_cur:
            sync_ig.apply_async(args=[account_id, url, token, next_cur], countdown=10)
    except IntegrityError:
        pass
    finally:
        cache.delete(_lock_key(account_id))

# models.py
class InstagramPost(models.Model):
    external_id = models.CharField(max_length=64, unique=True, db_index=True)
    caption = models.TextField(blank=True)
    taken_at = models.DateTimeField(null=True, blank=True)
    media_url = models.URLField(null=True, blank=True)
    updated_at = models.DateTimeField(auto_now=True)

Case 4 — Cache Stratejisi: “Liste & detay hızlı; invalidation tutarlı”

Problem

Doktor/klinik listesi ve detay sayfasını cache’lemek istiyorsun; güncellemede tutarlı invalidation gerek. Dogpile (cache stampede) yakınlarında da dayanıklı olmalı.

Uygulanabilir Yöntemler

1. View cache (kolay): cache_page
2. Low-level cache: Versiyonlu key + filtre hash
3. Detay: Key’i updated_at ile bağlı

# urls/views.py — kolay kazanım
from django.views.decorators.cache import cache_page
from django.utils.decorators import method_decorator

@method_decorator(cache_page(120), name='dispatch')  # 2 dk
class DoctorListView(ListView):
    model = Doctor
    paginate_by = 20

# services/cache_keys.py
def doctor_list_key(page, filters_hash, version):
    return f"doc:list:v{version}:p{page}:{filters_hash}"

# services/doctor_query.py
import hashlib
from django.core.cache import cache
from django.db.models import Count

def _filters_hash(params: dict) -> str:
    raw = "&".join(f"{k}={v}" for k,v in sorted(params.items()))
    return hashlib.md5(raw.encode()).hexdigest()

def get_doctor_list(params: dict, page: int, version: int):
    key = doctor_list_key(page, _filters_hash(params), version)
    if (data := cache.get(key)): return data
    qs = Doctor.objects.annotate(appointment_count=Count('appointments'))
    # ... filtreler, paginate, serialize
    data = serialize(paginate(qs, page))
    cache.set(key, data, 300)  # 5 dk
    return data

# detail cache (updated_at ile bağla)
def doctor_detail_key(pk, ts): return f"doc:detail:{pk}:{int(ts)}"

def get_doctor_detail(pk):
    obj = Doctor.objects.get(pk=pk)
    key = doctor_detail_key(pk, obj.updated_at.timestamp())
    if (c := cache.get(key)): return c
    data = serialize(obj)
    cache.set(key, data, 1800)  # 30 dk
    return data

# signals.py — liste versiyonunu bump et
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from django.core.cache import cache
from doctor.models import Doctor

LIST_VERSION_KEY = "doc:list:version"

def bump_list_version():
    if cache.get(LIST_VERSION_KEY) is None:
        cache.set(LIST_VERSION_KEY, 1)
    cache.incr(LIST_VERSION_KEY)

@receiver(post_save, sender=Doctor)
@receiver(post_delete, sender=Doctor)
def invalidate_list(sender, **kwargs):
    bump_list_version()

Case 5 — Güvenlik Sağlamlaştırma

Problem

Ürünün prod’da. CSRF/XSS, rate limit, dosya yükleme riskleri ve gizli anahtar yönetimi gibi başlıklar kontrol altında olmalı nasıl yaparsın ?

Uygulanabilir Yöntemler

• CSRF/XSS/SQLi: Django varsayılanlarını koru; dış HTML’de sanitize (örn. bleach), raw SQL’de parametrize et.

 

• AuthZ: DRF permission’ları ve obje düzeyi kontroller.

 

• Rate limit: DRF throttling + Nginx limit_req.

 

• File upload: Tip doğrulama, boyut limiti, SVG sanitize, farklı domain’den servis et.

 

• Secrets: SECRET_KEY, DB parolaları, API anahtarları env’de; rotasyon ve minimum yetki.

 

• HTTP güvenliği: HSTS, nosniff, referrer policy, CORS.

 

• Logging/Audit: Kritik aksiyonları iz kaydı olarak tut.

# settings.py — prod sertleştirme
SECURE_SSL_REDIRECT = True
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True
SECURE_HSTS_SECONDS = 31536000
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True
SECURE_CONTENT_TYPE_NOSNIFF = True
SECURE_REFERRER_POLICY = "strict-origin-when-cross-origin"
X_FRAME_OPTIONS = "DENY"
ALLOWED_HOSTS = ["example.com", "www.example.com"]

# DRF throttling örneği
REST_FRAMEWORK = {
  "DEFAULT_THROTTLE_CLASSES": [
    "rest_framework.throttling.AnonRateThrottle",
    "rest_framework.throttling.UserRateThrottle",
  ],
  "DEFAULT_THROTTLE_RATES": {
    "anon": "30/min",
    "user": "120/min",
  },
}

Ek: Dosya yüklemelerini ayrı bir domain/subdomain’den servis ederek cookie’lerle aynı origin’i paylaşmamayı düşün.