Django Interview Questions and Answers: 60 Questions from Easy to Hard

from decimal import Decimal
from django.db import transaction

class InsufficientFunds(Exception):
    pass

def debit(wallet_id, amount: Decimal):
    with transaction.atomic():
        w = (Wallet.objects.select_for_update()
             .get(pk=wallet_id))
        if w.balance < amount:
            raise InsufficientFunds()
        w.balance -= amount
        w.save(update_fields=["balance"])
        LedgerEntry.objects.create(wallet=w, amount=-amount)

from django.db.models import Prefetch

shortlisted = (Application.objects
               .filter(status="shortlisted")
               .select_related("candidate"))
jobs = Job.objects.prefetch_related(
    Prefetch("applications", queryset=shortlisted,
             to_attr="shortlisted_apps"))

for job in jobs:                 # 2 queries in total
    for app in job.shortlisted_apps:
        print(job.title, app.candidate.first_name)

from datetime import timedelta
from django.db.models import Exists, OuterRef, Subquery
from django.utils import timezone

latest = (Application.objects
          .filter(candidate=OuterRef("pk"))
          .order_by("-created_at")
          .values("created_at")[:1])
recent = Application.objects.filter(
    candidate=OuterRef("pk"),
    created_at__gte=timezone.now() - timedelta(days=30))

candidates = (Candidate.objects
              .annotate(last_applied=Subquery(latest))
              .filter(Exists(recent)))

from django.db import models
from django.utils import timezone

class JobQuerySet(models.QuerySet):
    def open(self):
        return self.filter(is_open=True,
                           last_date__gte=timezone.localdate())

    def in_city(self, city):
        return self.filter(city__iexact=city)

class Job(models.Model):
    title = models.CharField(max_length=200)
    city = models.CharField(max_length=80)
    is_open = models.BooleanField(default=True)
    last_date = models.DateField()

    objects = JobQuerySet.as_manager()

# Job.objects.open().in_city("Pune").order_by("-id")

from django.contrib.postgres.operations import (
    AddIndexConcurrently,
)
from django.db import migrations, models

class Migration(migrations.Migration):
    atomic = False
    dependencies = [("payments", "0031_payment_utr")]
    operations = [
        AddIndexConcurrently(
            "payment",
            models.Index(fields=["utr"], name="payment_utr_idx"),
        ),
    ]

import asyncio
import httpx
from django.http import JsonResponse

async def verify_vendor(request, gstin):
    async with httpx.AsyncClient(timeout=5) as client:
        gst, bank = await asyncio.gather(
            client.get(f"https://gst.example.com/v1/{gstin}"),
            client.get(f"https://bank.example.com/v1/{gstin}"),
        )
    vendor = await Vendor.objects.aget(gstin=gstin)
    return JsonResponse({"vendor": vendor.name,
                         "gst_ok": gst.status_code == 200,
                         "bank_ok": bank.status_code == 200})

import os

DEBUG = False
SECRET_KEY = os.environ["DJANGO_SECRET_KEY"]
ALLOWED_HOSTS = ["hr.example.com"]
CSRF_TRUSTED_ORIGINS = ["https://hr.example.com"]

SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")
SECURE_SSL_REDIRECT = True
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True
SECURE_HSTS_SECONDS = 3600  # raise once HTTPS is proven

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": os.environ["DB_NAME"],
        "CONN_MAX_AGE": 60,
        "CONN_HEALTH_CHECKS": True,
    }
}

from django.utils.html import format_html
from django.utils.safestring import mark_safe

# Unsafe: user text becomes live HTML
def badge_bad(name):
    return mark_safe(f"<b>{name}</b>")

# Safe: the argument is escaped
def badge(name):
    return format_html("<b>{}</b>", name)

{# template: hand data to JavaScript safely #}
{{ chart_data|json_script:"chart-data" }}
<script>
    const data = JSON.parse(
        document.getElementById("chart-data").textContent);
</script>

# Vulnerable: the input becomes part of the SQL
def search_bad(term):
    return Employee.objects.raw(
        f"SELECT * FROM hr_employee WHERE name = '{term}'")

# Safe: the driver sends term as a parameter
def search(term):
    return Employee.objects.raw(
        "SELECT * FROM hr_employee WHERE name = %s", [term])

SORTS = {"name": "name", "joined": "date_of_joining"}

def sorted_employees(key):
    return Employee.objects.order_by(SORTS.get(key, "name"))

import random
from django.core.cache import cache
from django.db import transaction

def headcount(org_id):
    key = f"org:{org_id}:headcount"
    return cache.get_or_set(
        key,
        lambda: Employee.objects.filter(org_id=org_id).count(),
        timeout=600 + random.randint(0, 60))

def employee_changed(org_id):
    transaction.on_commit(
        lambda: cache.delete(f"org:{org_id}:headcount"))

from rest_framework import viewsets
from rest_framework.permissions import IsAuthenticated

class OrgScopedViewSet(viewsets.ModelViewSet):
    permission_classes = [IsAuthenticated]

    def get_queryset(self):
        org = self.request.user.org
        return super().get_queryset().filter(org=org)

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

class InvoiceViewSet(OrgScopedViewSet):
    queryset = Invoice.objects.all()
    serializer_class = InvoiceSerializer

from django.db.models import Count

class JobSerializer(serializers.ModelSerializer):
    company = CompanySerializer(read_only=True)
    skills = serializers.SlugRelatedField(
        many=True, read_only=True, slug_field="name")
    applicant_count = serializers.IntegerField(read_only=True)

    class Meta:
        model = Job
        fields = ["id", "title", "company", "skills",
                  "applicant_count"]

class JobViewSet(viewsets.ReadOnlyModelViewSet):
    serializer_class = JobSerializer

    def get_queryset(self):
        return (Job.objects.select_related("company")
                .prefetch_related("skills")
                .annotate(applicant_count=Count("applications")))

import csv
import io
from django.http import StreamingHttpResponse

def csv_lines(rows):
    buf = io.StringIO()
    writer = csv.writer(buf)
    for row in rows:
        writer.writerow(row)
        yield buf.getvalue()
        buf.seek(0)
        buf.truncate(0)

def export_payments(request):
    rows = (Payment.objects.filter(org=request.user.org)
            .values_list("id", "utr", "amount", "paid_on")
            .iterator(chunk_size=2000))
    resp = StreamingHttpResponse(csv_lines(rows),
                                 content_type="text/csv")
    resp["Content-Disposition"] = 'attachment; filename="pay.csv"'
    return resp