Python Developer Interview Questions (Part 2): Multithreading, Async, FastAPI, Django and Flask

Welcome to Part 2 of our Python developer interview series for 2–4 years of experience. Part 1 covered core Python and coding problems. This part covers the topics that separate a script writer from a backend developer: multithreading, multiprocessing and async programming, the three big web frameworks FastAPI, Django and Flask, and the scenario-based questions interviewers use to see how you solve real production problems.

As in Part 1, every example was run with Python 3.13 (FastAPI 0.162, Django 6.1, Flask 3.1, pytest 9.2), and the output below each example is the real output. Examples that start with a file name such as # app.py use threads, processes or a web framework, so run them on your own machine.

Interview Roadmap for Part 2

Section 3: Python Multithreading and Async Programming Interview Questions (Q31–Q42)

Q31 What is the difference between concurrency and parallelism?

Short answer: Concurrency is dealing with many tasks at once by switching between them; parallelism is doing many tasks at the same moment on several CPU cores. Threads and asyncio give you concurrency; multiprocessing gives you parallelism.

CONCURRENCY (one core, tasks take turns while others wait for I/O)
Core 1:  [A]──[B]──[A]──[C]──[B]──[A]

PARALLELISM (several cores, tasks really run at the same time)
Core 1:  [A][A][A][A]
Core 2:  [B][B][B][B]
Core 3:  [C][C][C][C]

Q32 What is the GIL? Show its effect with code.

Short answer: The Global Interpreter Lock lets only one thread execute Python bytecode at a time in standard CPython. Threads therefore do not speed up CPU-bound Python code, but processes do, because each process has its own interpreter and its own GIL.

# gil_demo.py
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

def cpu_task(n):
    return sum(i * i for i in range(n))

def timed(executor_cls, label):
    start = time.perf_counter()
    with executor_cls(max_workers=4) as pool:
        list(pool.map(cpu_task, [3_000_000] * 4))
    print(f"{label:10} {time.perf_counter() - start:.2f} s")

if __name__ == "__main__":
    start = time.perf_counter()
    for _ in range(4):
        cpu_task(3_000_000)
    print(f"{'sequential':10} {time.perf_counter() - start:.2f} s")
    timed(ThreadPoolExecutor, "threads")
    timed(ProcessPoolExecutor, "processes")

sequential 0.43 s
threads    0.43 s
processes  0.17 s

Timings come from a real run on a laptop; your numbers will differ, but the comparison stays the same.

Threads take about as long as the sequential loop because of the GIL; processes run on several cores. Strong answers also mention two details: C extensions such as NumPy release the GIL during heavy work, and Python 3.13 offers an experimental free-threaded build without the GIL, which most production systems do not use yet.

Q33 How do you create threads, wait for them and get their results?

Short answer: Create threading.Thread objects, start() them and join() them. Threads cannot return values directly, so in practice use ThreadPoolExecutor, which returns results through futures.

# threads_basic.py
import threading
import time
from concurrent.futures import ThreadPoolExecutor

def download(name, seconds):
    time.sleep(seconds)                 # simulates waiting for the network
    return f"{name} done"

start = time.perf_counter()
threads = [threading.Thread(target=download, args=(f"file{i}", 0.5)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()                            # wait for all threads to finish
print(f"3 downloads with Thread: {time.perf_counter() - start:.1f} s")

with ThreadPoolExecutor(max_workers=3) as pool:
    results = list(pool.map(download, ["a", "b", "c"], [0.5, 0.5, 0.5]))
print(results)
print("daemon thread example:", threading.Thread(target=print, daemon=True).daemon)

3 downloads with Thread: 0.5 s
['a done', 'b done', 'c done']
daemon thread example: True

Timings come from a real run on a laptop; your numbers will differ, but the comparison stays the same.

A daemon thread is killed automatically when the main program exits. Use it for background helpers such as heartbeats, never for work that must finish, like writing to a database.

Q34 What is a race condition? How does a Lock fix it?

Short answer: A race condition happens when several threads read and write shared data at the same time, and the result depends on timing. A threading.Lock makes the read-modify-write sequence atomic: only one thread can be inside it at a time.

# race_condition.py
import threading
import time

balance = 0
lock = threading.Lock()

def deposit_unsafe(times):
    global balance
    for _ in range(times):
        current = balance
        time.sleep(0)                  # gives other threads a chance to run, like real I/O would
        balance = current + 1

def deposit_safe(times):
    global balance
    for _ in range(times):
        with lock:                     # only one thread at a time
            current = balance
            time.sleep(0)
            balance = current + 1

for func in (deposit_unsafe, deposit_safe):
    balance = 0
    threads = [threading.Thread(target=func, args=(1000,)) for _ in range(4)]
    [t.start() for t in threads]
    [t.join() for t in threads]
    print(f"{func.__name__:15} balance={balance}  correct={balance == 4000}")

deposit_unsafe  balance=1024  correct=False
deposit_safe    balance=4000  correct=True

The unsafe number changes on every run; the safe one is always 4000.

Interview tip: count += 1 looks like one step but is several bytecode instructions (load, add, store), so even it is not thread-safe.

Q35 What is a deadlock, and how do you avoid it?

Short answer: A deadlock happens when two threads each hold one lock and wait forever for the other’s lock. Avoid it by always acquiring locks in the same order, keeping locks short, using timeouts, and using RLock when the same thread needs to re-enter.

# deadlock.py
import threading

accounts_lock = threading.Lock()
audit_lock = threading.Lock()

def transfer():
    # Rule: always take accounts_lock first, then audit_lock, in every function
    with accounts_lock:
        with audit_lock:
            return "transfer ok"

print(transfer())

# A timeout turns a silent hang into an error you can handle
accounts_lock.acquire()
got_it = accounts_lock.acquire(timeout=0.5)
print("second acquire succeeded:", got_it)
accounts_lock.release()

# RLock: the same thread may acquire it again (useful for recursive code)
rlock = threading.RLock()
with rlock:
    with rlock:
        print("RLock re-entered safely")

transfer ok
second acquire succeeded: False
RLock re-entered safely

Q36 How do you implement the producer–consumer pattern?

Short answer: Use queue.Queue: it is thread-safe, so producers put() work and consumers get() it without any extra locks. A special “sentinel” value tells consumers to stop.

# producer_consumer.py
import queue
import threading

tasks = queue.Queue(maxsize=10)        # maxsize gives back-pressure
results = []
results_lock = threading.Lock()
STOP = object()

def producer(n):
    for order_id in range(1, n + 1):
        tasks.put(order_id)
    for _ in range(2):                 # one STOP per consumer
        tasks.put(STOP)

def consumer(name):
    while True:
        item = tasks.get()
        if item is STOP:
            break
        with results_lock:
            results.append(item * 100)  # "process" the order

workers = [threading.Thread(target=consumer, args=(f"w{i}",)) for i in range(2)]
[w.start() for w in workers]
producer(6)
[w.join() for w in workers]
print("processed:", sorted(results))

processed: [100, 200, 300, 400, 500, 600]

Q37 When do you use ThreadPoolExecutor? What is the difference between map() and submit()?

Short answer: Use a thread pool for I/O-bound work: API calls, file downloads, database queries with blocking drivers. map() returns results in input order; submit() with as_completed() gives each result as soon as it is ready and lets you handle errors one by one.

# thread_pool.py
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_price(product):
    delays = {"laptop": 0.3, "phone": 0.1, "tablet": 0.2}
    if product == "watch":
        raise ValueError("product not found")
    time.sleep(delays[product])
    return product, delays[product] * 1000

with ThreadPoolExecutor(max_workers=4) as pool:
    futures = {pool.submit(fetch_price, p): p for p in ["laptop", "phone", "tablet", "watch"]}
    for future in as_completed(futures):
        try:
            name, price = future.result()
            print(f"finished: {name} -> {price:.0f}")
        except ValueError as err:
            print(f"failed:   {futures[future]} -> {err}")

failed:   watch -> product not found
finished: phone -> 100
finished: tablet -> 200
finished: laptop -> 300

Q38 How does multiprocessing work? Why do you need if __name__ == "__main__"?

Short answer: multiprocessing starts separate Python processes, each with its own memory and GIL, so CPU-bound work runs in parallel. On Windows and macOS new processes import your script again, so the start-up code must sit under if __name__ == "__main__": or it would start processes endlessly.

# multiprocessing_demo.py
import os
from multiprocessing import Pool, Value

def resize(image_id):
    return f"image{image_id} processed by a worker process"

def count_up(counter, n):
    for _ in range(n):
        with counter.get_lock():       # shared memory needs a lock too
            counter.value += 1

if __name__ == "__main__":
    with Pool(processes=3) as pool:
        for line in pool.map(resize, [1, 2, 3]):
            print(line)
    counter = Value("i", 0)
    count_up(counter, 5)
    print("shared counter:", counter.value)
    print("main process pid is an int:", isinstance(os.getpid(), int))

image1 processed by a worker process
image2 processed by a worker process
image3 processed by a worker process
shared counter: 5
main process pid is an int: True

Remember: arguments and results are pickled to cross process boundaries, so lambdas and open connections cannot be sent, and sending huge objects is slow. Processes are worth it when each task does a lot of computation compared with the data it moves.

Q39 What is asyncio? Show the difference between sequential and concurrent awaits.

Short answer: asyncio runs many coroutines on one thread using an event loop. When a coroutine hits await on I/O, the loop runs another one. asyncio.gather() runs several awaitables concurrently.

# asyncio_basics.py
import asyncio
import time

async def call_service(name, seconds):
    await asyncio.sleep(seconds)          # non-blocking wait, like a network call
    return f"{name} ok"

async def main():
    start = time.perf_counter()
    for name in ["users", "orders", "payments"]:
        await call_service(name, 0.5)     # one after another
    print(f"sequential: {time.perf_counter() - start:.1f} s")

    start = time.perf_counter()
    results = await asyncio.gather(*(call_service(n, 0.5) for n in ["users", "orders", "payments"]))
    print(f"gather:     {time.perf_counter() - start:.1f} s ->", results)

asyncio.run(main())

sequential: 1.5 s
gather:     0.5 s -> ['users ok', 'orders ok', 'payments ok']

Q40 How do you use tasks, timeouts and cancellation in asyncio?

Short answer: asyncio.create_task() schedules a coroutine to run in the background. asyncio.timeout() (Python 3.11+) or asyncio.wait_for() cancels work that takes too long, and the cancelled coroutine receives CancelledError so it can clean up.

# asyncio_tasks.py
import asyncio

async def slow_report():
    try:
        await asyncio.sleep(5)
        return "report ready"
    except asyncio.CancelledError:
        print("slow_report: cancelled, cleaning up")
        raise                              # always re-raise CancelledError

async def heartbeat():
    for i in range(3):
        print("heartbeat", i)
        await asyncio.sleep(0.1)

async def main():
    beat = asyncio.create_task(heartbeat())   # runs in the background
    try:
        async with asyncio.timeout(0.5):
            await slow_report()
    except TimeoutError:
        print("main: report took too long")
    await beat

asyncio.run(main())

heartbeat 0
heartbeat 1
heartbeat 2
slow_report: cancelled, cleaning up
main: report took too long

Q41 How do you limit how many requests run at the same time in asyncio?

Short answer: Use asyncio.Semaphore. It allows only N coroutines inside the protected block at once, which protects third-party APIs and your own database from overload.

# semaphore_demo.py
import asyncio

running = 0
peak = 0
limit = asyncio.Semaphore(3)

async def fetch(page):
    global running, peak
    async with limit:                     # at most 3 at a time
        running += 1
        peak = max(peak, running)
        await asyncio.sleep(0.1)
        running -= 1
        return page

async def main():
    pages = await asyncio.gather(*(fetch(p) for p in range(10)))
    print("pages fetched:", len(pages), "| peak concurrent requests:", peak)

asyncio.run(main())

pages fetched: 10 | peak concurrent requests: 3

Q42 What happens if you call blocking code inside async code? When do you choose threads, processes or asyncio?

Short answer: A blocking call such as time.sleep() or requests.get() inside a coroutine freezes the whole event loop. Move it to a thread with await asyncio.to_thread(...), or use an async library such as httpx or asyncpg.

# blocking_in_async.py
import asyncio
import time

def blocking_report():
    time.sleep(0.5)                       # a blocking library call
    return "report"

async def main():
    start = time.perf_counter()
    await asyncio.gather(*(asyncio.to_thread(blocking_report) for _ in range(3)))
    print(f"3 blocking calls via to_thread: {time.perf_counter() - start:.1f} s")

    start = time.perf_counter()
    async def bad():
        time.sleep(0.5)                   # blocks the event loop!
    await asyncio.gather(bad(), bad(), bad())
    print(f"3 blocking calls inside async:  {time.perf_counter() - start:.1f} s")

asyncio.run(main())

3 blocking calls via to_thread: 0.5 s
3 blocking calls inside async:  1.5 s

Quick Revision: Q31–Q42

Section 4: FastAPI, Django and Flask Interview Questions (Q43–Q55)

Q43 Django vs Flask vs FastAPI: which one would you choose?

Short answer: Django for full web applications that need an ORM, admin panel and authentication out of the box; Flask for small, flexible services; FastAPI for high-performance, async APIs with automatic validation and documentation.

Q44 Explain the Django request–response lifecycle.

Short answer: The web server passes the request to Django through WSGI or ASGI; it travels through the middleware stack, the URL resolver picks a view, the view uses the ORM and a template or serializer, and the response travels back through the middleware in reverse order.

Browser ──▶ Web server (Nginx) ──▶ Gunicorn / Uvicorn (WSGI / ASGI)
                                            │
                                            ▼
                   Middleware (security, session, auth, CSRF) ─── request
                                            │
                                            ▼
                   URL resolver (urls.py) ──▶ View (function / class)
                                            │
                                   ORM ◀────┴────▶ Template / Serializer
                                            │
                                            ▼
                   Middleware (in reverse order) ─── response
                                            │
Browser ◀──────────────────────────────────┘

# middleware.py
# A custom middleware that times every request
import logging
import time

logger = logging.getLogger(__name__)

class RequestTimingMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response          # runs once, at start-up

    def __call__(self, request):
        start = time.perf_counter()
        response = self.get_response(request)     # the view runs here
        ms = (time.perf_counter() - start) * 1000
        response["X-Response-Time-ms"] = f"{ms:.0f}"
        logger.info("%s %s %s %.0fms", request.method, request.path, response.status_code, ms)
        return response

# settings.py: add "myapp.middleware.RequestTimingMiddleware" to MIDDLEWARE

Q45 What is the N+1 query problem? How do select_related and prefetch_related fix it?

Short answer: N+1 happens when you load N objects with one query and then run one more query per object to read a related field. select_related uses a SQL JOIN for ForeignKey and OneToOne; prefetch_related runs one extra query for reverse and many-to-many relations and joins them in Python.

# django_orm_demo.py
import django
from django.conf import settings

settings.configure(
    DATABASES={"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}},
    INSTALLED_APPS=["django.contrib.contenttypes"],
    DEFAULT_AUTO_FIELD="django.db.models.AutoField",
)
django.setup()

from django.db import connection, models
from django.test.utils import CaptureQueriesContext

class Author(models.Model):
    name = models.CharField(max_length=50)
    class Meta:
        app_label = "shop"

class Book(models.Model):
    title = models.CharField(max_length=100)
    price = models.IntegerField(default=499)
    stock = models.IntegerField(default=10)
    author = models.ForeignKey(Author, on_delete=models.CASCADE, related_name="books")
    class Meta:
        app_label = "shop"

with connection.schema_editor() as editor:
    editor.create_model(Author)
    editor.create_model(Book)

for name in ["Asha", "Ravi", "Meera"]:
    author = Author.objects.create(name=name)
    Book.objects.create(title=f"{name}'s Python Notes", author=author)
    Book.objects.create(title=f"{name}'s Django Tips", author=author, price=799)

with CaptureQueriesContext(connection) as q:
    for book in Book.objects.all():
        _ = book.author.name                     # one extra query per book
print("without select_related:", len(q.captured_queries), "queries")

with CaptureQueriesContext(connection) as q:
    for book in Book.objects.select_related("author"):
        _ = book.author.name                     # author comes from the JOIN
print("with select_related:   ", len(q.captured_queries), "query")

with CaptureQueriesContext(connection) as q:
    for author in Author.objects.prefetch_related("books"):
        _ = [b.title for b in author.books.all()]
print("with prefetch_related: ", len(q.captured_queries), "queries")

without select_related: 7 queries
with select_related:    1 query
with prefetch_related:  2 queries

With 6 books, the slow version runs 7 queries. With 6,000 books it would run 6,001 and the page would take seconds. Tools such as Django Debug Toolbar or CaptureQueriesContext in tests catch this early.

Q46 Are QuerySets lazy? What are F and Q expressions?

Short answer: Yes. A QuerySet runs SQL only when you iterate, slice, call len(), list() or similar. F() refers to a column value inside the database, which allows race-free updates; Q() builds OR / NOT conditions.

# django_orm_demo.py
from django.db.models import F, Q

with CaptureQueriesContext(connection) as q:
    cheap = Book.objects.filter(price__lt=600)     # no SQL yet
    cheap = cheap.order_by("title")                # still no SQL
    print("queries before use:", len(q.captured_queries))
    print("cheap books:", cheap.count())
    print("queries after count():", len(q.captured_queries))

# Race-free stock update: the subtraction happens inside the database
Book.objects.filter(title__startswith="Asha").update(stock=F("stock") - 1)
print("Asha stock:", list(Book.objects.filter(title__startswith="Asha").values_list("stock", flat=True)))

# OR condition with Q objects
result = Book.objects.filter(Q(author__name="Ravi") | Q(price__gt=700)).count()
print("Ravi's books or price > 700:", result)

queries before use: 0
cheap books: 3
queries after count(): 1
Asha stock: [9, 9]
Ravi's books or price > 700: 4

Why F() matters: book.stock = book.stock - 1; book.save() reads the value into Python first, so two users buying at the same moment can both write the same number. update(stock=F("stock") - 1) lets the database do it atomically.

Q47 How do you handle transactions in Django?

Short answer: Wrap related writes in transaction.atomic() so they all succeed or all roll back. Use select_for_update() to lock rows you are about to change, and transaction.on_commit() to send emails or tasks only after the data is really saved.

# views.py
from django.db import transaction
from django.http import JsonResponse

from .models import Order, Product
from .tasks import send_order_email

def place_order(request, product_id):
    with transaction.atomic():
        product = Product.objects.select_for_update().get(pk=product_id)   # row lock until commit
        if product.stock < 1:
            return JsonResponse({"error": "Out of stock"}, status=409)
        product.stock -= 1
        product.save(update_fields=["stock"])
        order = Order.objects.create(product=product, user=request.user)
        transaction.on_commit(lambda: send_order_email.delay(order.id))     # only if the commit succeeds
    return JsonResponse({"order_id": order.id}, status=201)

Q48 What security features does Django give you, and how do you avoid SQL injection?

# django_orm_demo.py
name = "Asha' OR '1'='1"                         # a malicious input

unsafe_sql = f"SELECT * FROM shop_author WHERE name = '{name}'"
print("unsafe:", unsafe_sql)

with connection.cursor() as cursor:              # safe: the value is passed as a parameter
    cursor.execute("SELECT COUNT(*) FROM shop_author WHERE name = %s", [name])
    print("rows matched with params:", cursor.fetchone()[0])
print("rows via ORM:", Author.objects.filter(name=name).count())

unsafe: SELECT * FROM shop_author WHERE name = 'Asha' OR '1'='1'
rows matched with params: 0
rows via ORM: 0

Q49 How do you structure a Flask application? Explain the app factory, blueprints and contexts.

Short answer: Use an application factory (create_app()) so tests and environments can build their own app, split features into blueprints, and remember Flask’s two contexts: the application context (current_app, g) and the request context (request, session).

# flask_app.py
from flask import Blueprint, Flask, current_app, g, jsonify, request

orders_bp = Blueprint("orders", __name__, url_prefix="/orders")

@orders_bp.before_app_request
def load_user():
    g.user = request.headers.get("X-User", "guest")        # g lives for one request

@orders_bp.get("/<int:order_id>")
def get_order(order_id):
    return jsonify(order_id=order_id, user=g.user, env=current_app.config["ENV_NAME"])

@orders_bp.post("/")
def create_order():
    data = request.get_json(silent=True) or {}
    if "product_id" not in data:
        return jsonify(error="product_id is required"), 400
    return jsonify(status="created", product_id=data["product_id"]), 201

def create_app(env_name="development"):
    app = Flask(__name__)
    app.config["ENV_NAME"] = env_name
    app.register_blueprint(orders_bp)
    return app

client = create_app("testing").test_client()
print(client.get("/orders/7", headers={"X-User": "asha"}).get_json())
response = client.post("/orders/", json={})
print(response.status_code, response.get_json())
print(client.post("/orders/", json={"product_id": 3}).status_code)

{'env': 'testing', 'order_id': 7, 'user': 'asha'}
400 {'error': 'product_id is required'}
201

Q50 How does FastAPI validate path, query and body parameters?

Short answer: FastAPI reads the type hints. Path and query parameters are converted and checked from the function signature, and request bodies are validated by Pydantic models. Invalid input returns 422 Unprocessable Entity with a clear error, before your code runs.

# fastapi_validation.py
from fastapi import FastAPI, Query
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field

app = FastAPI()

class OrderIn(BaseModel):
    product_id: int
    quantity: int = Field(gt=0, le=10)
    coupon: str | None = None

@app.get("/products/{product_id}")
def get_product(product_id: int, currency: str = Query("INR", min_length=3, max_length=3)):
    return {"product_id": product_id, "currency": currency}

@app.post("/orders", status_code=201)
def create_order(order: OrderIn):
    return {"status": "created", **order.model_dump()}

client = TestClient(app)
print(client.get("/products/7?currency=USD").json())
r = client.get("/products/abc")
print(r.status_code, r.json()["detail"][0]["msg"])
print(client.post("/orders", json={"product_id": 1, "quantity": 2}).json())
r = client.post("/orders", json={"product_id": 1, "quantity": 50})
print(r.status_code, r.json()["detail"][0]["msg"])

{'product_id': 7, 'currency': 'USD'}
422 Input should be a valid integer, unable to parse string as an integer
{'status': 'created', 'product_id': 1, 'quantity': 2, 'coupon': None}
422 Input should be less than or equal to 10

Q51 How does dependency injection work in FastAPI? How do you override a dependency in tests?

Short answer: Depends() declares what an endpoint needs: a database session, settings, the current user. A dependency with yield runs set-up before the request and clean-up after it. In tests, app.dependency_overrides swaps the real dependency for a fake one.

# fastapi_dependencies.py
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

class FakeDB:
    def __init__(self, name):
        self.name = name
    def get_user(self, user_id):
        return {"id": user_id, "source": self.name}

def get_db():
    db = FakeDB("production-db")
    print("open connection")
    try:
        yield db                        # the endpoint runs here
    finally:
        print("close connection")       # always runs after the response

@app.get("/users/{user_id}")
def read_user(user_id: int, db: FakeDB = Depends(get_db)):
    return db.get_user(user_id)

client = TestClient(app)
print(client.get("/users/1").json())

app.dependency_overrides[get_db] = lambda: FakeDB("test-db")
print(client.get("/users/1").json())
app.dependency_overrides.clear()

open connection
close connection
{'id': 1, 'source': 'production-db'}
{'id': 1, 'source': 'test-db'}

Q52 In FastAPI, when do you use async def and when def?

Short answer: Use async def when everything inside is non-blocking (async database drivers, httpx.AsyncClient). Use plain def when you call blocking libraries: FastAPI runs def endpoints in a thread pool so they do not block the event loop. The worst choice is blocking code inside async def.

# endpoints.py
import httpx
import requests
from fastapi import FastAPI

app = FastAPI()

@app.get("/good-async")
async def good_async():
    async with httpx.AsyncClient() as client:         # non-blocking
        r = await client.get("https://api.example.com/rates", timeout=5)
    return r.json()

@app.get("/good-sync")
def good_sync():
    r = requests.get("https://api.example.com/rates", timeout=5)   # blocking, but FastAPI runs it in a thread pool
    return r.json()

@app.get("/bad")
async def bad():
    r = requests.get("https://api.example.com/rates", timeout=5)   # blocks the whole event loop!
    return r.json()

Q53 How do you add middleware and background tasks in FastAPI?

Short answer: Middleware wraps every request (timing, logging, request IDs). BackgroundTasks runs small jobs after the response has been sent, such as sending an email, so the user does not wait for them.

# fastapi_middleware.py
import time
import uuid

from fastapi import BackgroundTasks, FastAPI, Request
from fastapi.testclient import TestClient

app = FastAPI()
sent_emails = []

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4())[:8])
    start = time.perf_counter()
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Process-Time"] = f"{(time.perf_counter() - start) * 1000:.1f}ms"
    return response

def send_welcome_email(email: str):
    sent_emails.append(email)               # imagine an SMTP call here

@app.post("/signup")
def signup(email: str, background: BackgroundTasks):
    background.add_task(send_welcome_email, email)
    return {"message": "account created"}

client = TestClient(app)
r = client.post("/signup?email=asha@example.com", headers={"X-Request-ID": "req-42"})
print(r.json(), "| request id:", r.headers["X-Request-ID"])
print("emails sent after the response:", sent_emails)

{'message': 'account created'} | request id: req-42
emails sent after the response: ['asha@example.com']

For heavy or important jobs (video processing, payment reconciliation) use a real task queue such as Celery, RQ or a cloud queue: BackgroundTasks runs in the same process and is lost if the server restarts.

Q54 How do you protect FastAPI endpoints with bearer tokens?

Short answer: Use the OAuth2PasswordBearer (or HTTPBearer) security scheme to read the token from the Authorization header, verify it in a dependency, and return 401 if it is missing or invalid. In real projects the token is a signed JWT verified with a library such as PyJWT.

# fastapi_auth.py
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

FAKE_TOKENS = {"token-asha": {"username": "asha", "role": "admin"}}   # a JWT decode would go here

def current_user(token: str = Depends(oauth2_scheme)):
    user = FAKE_TOKENS.get(token)
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token",
                            headers={"WWW-Authenticate": "Bearer"})
    return user

def admin_only(user: dict = Depends(current_user)):
    if user["role"] != "admin":
        raise HTTPException(status_code=403, detail="Admins only")
    return user

@app.get("/reports")
def reports(user: dict = Depends(admin_only)):
    return {"report": "sales", "requested_by": user["username"]}

client = TestClient(app)
print(client.get("/reports").status_code)
print(client.get("/reports", headers={"Authorization": "Bearer wrong"}).json())
print(client.get("/reports", headers={"Authorization": "Bearer token-asha"}).json())

401
{'detail': 'Invalid token'}
{'report': 'sales', 'requested_by': 'asha'}

Q55 How do you test Python code with pytest? Explain fixtures, parametrize and mocking.

Short answer: Fixtures provide reusable set-up, @pytest.mark.parametrize runs one test with many inputs, and unittest.mock (or pytest’s monkeypatch) replaces slow or external dependencies such as payment gateways.

# test_shop.py
import sys
from unittest.mock import Mock

import pytest

def apply_discount(price, percent):
    if not 0 <= percent <= 100:
        raise ValueError("percent must be 0-100")
    return round(price * (1 - percent / 100), 2)

def charge(gateway, amount):
    response = gateway.pay(amount)
    return "paid" if response["ok"] else "failed"

@pytest.fixture
def cart():
    return {"items": [499, 799], "currency": "INR"}

def test_cart_total(cart):
    assert sum(cart["items"]) == 1298

@pytest.mark.parametrize("price, percent, expected", [(1000, 10, 900.0), (499, 0, 499.0), (200, 100, 0.0)])
def test_apply_discount(price, percent, expected):
    assert apply_discount(price, percent) == expected

def test_invalid_discount():
    with pytest.raises(ValueError):
        apply_discount(100, 150)

def test_charge_uses_gateway():
    gateway = Mock()                                  # no real payment is made
    gateway.pay.return_value = {"ok": True}
    assert charge(gateway, 1000) == "paid"
    gateway.pay.assert_called_once_with(1000)

def test_charge_failure():
    gateway = Mock()
    gateway.pay.return_value = {"ok": False}
    assert charge(gateway, 9000) == "failed"

if __name__ == "__main__":
    sys.exit(pytest.main([__file__, "-q", "-p", "no:cacheprovider"]))

.......                                                                  [100%]
7 passed in 0.02s

Quick Revision: Q43–Q55

Section 5: Python Scenario-Based Interview Questions (Q56–Q60)

These questions have no single right answer. Interviewers want a structured approach: measure → find the cause → fix → prevent. Each answer below includes a small, real example of the tool you would use.

Q56 An API endpoint suddenly takes 4 seconds. How do you find the slow part?

Short answer: Measure before guessing. Check the logs and APM traces, reproduce the request, then profile the code path. In Python, cProfile shows which functions take the most time.

# profile_endpoint.py
import cProfile
import pstats
import time

def load_user():
    time.sleep(0.05)

def load_orders():
    for _ in range(20):          # N+1: one query per order
        time.sleep(0.02)

def build_response():
    return {"ok": True}

def endpoint():
    load_user()
    load_orders()
    return build_response()

profiler = cProfile.Profile()
profiler.enable()
endpoint()
profiler.disable()

stats = pstats.Stats(profiler)
ranking = sorted(
    ((func[2], data[3]) for func, data in stats.stats.items() if func[2] in {"load_user", "load_orders", "build_response"}),
    key=lambda item: item[1], reverse=True,
)
for name, cumulative in ranking:
    print(f"{name:15} {cumulative:.2f} s")

load_orders     0.48 s
load_user       0.06 s
build_response  0.00 s

Timings come from a real run on a laptop; your numbers will differ, but the comparison stays the same.

Typical causes, in the order you should check them: N+1 database queries, a missing index, a slow external API without a timeout, large JSON serialisation, and blocking code inside async endpoints. Fix the biggest one, measure again, then add a performance test or alert so it cannot come back silently.

Q57 Memory keeps growing until the service restarts. How do you find the leak?

Short answer: Watch memory over time, then compare snapshots with tracemalloc to see which line keeps allocating. In long-running Python services the usual culprits are caches without a size limit, global lists that keep growing and objects kept alive by references.

# find_leak.py
import tracemalloc

_cache = {}                                   # grows forever: the leak

def handle_request(request_id):
    _cache[request_id] = "x" * 10_000         # stores every response
    return len(_cache)

tracemalloc.start()
before = tracemalloc.take_snapshot()
for i in range(500):
    handle_request(i)
after = tracemalloc.take_snapshot()

top = after.compare_to(before, "lineno")[0]
print("biggest growth at line:", top.traceback[0].lineno)
print(f"memory added: about {top.size_diff / 1_000_000:.0f} MB in {top.count_diff} blocks")

biggest growth at line: 7
memory added: about 5 MB in 501 blocks

Fix: give the cache a limit (functools.lru_cache(maxsize=...), cachetools.TTLCache) or move it to Redis with an expiry time. Gunicorn’s max_requests setting is a useful safety net, but not a fix.

Q58 You need to import a CSV file with 1 million rows into the database. How do you do it?

Short answer: Never load the whole file into memory and never insert row by row. Stream the file, validate each row, insert in batches (for example 1,000 rows per query) inside transactions, and run it as a background job with progress tracking.

# csv_import.py
import csv
import io
import sqlite3
from itertools import batched

# 100,000 fake rows instead of a real 1 GB file, so the example runs quickly
fake_file = io.StringIO("id,email,amount\n" + "".join(f"{i},user{i}@mail.com,{i % 500}\n" for i in range(1, 100_001)))

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE payments (id INTEGER PRIMARY KEY, email TEXT, amount INTEGER)")

reader = csv.DictReader(fake_file)        # streams row by row
batches = bad_rows = 0
for batch in batched(reader, 1000):
    rows = []
    for row in batch:
        if "@" not in row["email"]:
            bad_rows += 1
            continue
        rows.append((int(row["id"]), row["email"], int(row["amount"])))
    with db:                               # one transaction per batch
        db.executemany("INSERT INTO payments VALUES (?, ?, ?)", rows)
    batches += 1

print("batches:", batches, "| bad rows:", bad_rows)
print("rows in table:", db.execute("SELECT COUNT(*) FROM payments").fetchone()[0])

batches: 100 | bad rows: 0
rows in table: 100000

In Django use bulk_create(objs, batch_size=1000); with PostgreSQL the fastest option is COPY. Mention idempotency too: if the job crashes at row 600,000, it should be able to restart without creating duplicates.

Q59 Users see random 500 errors in production, but you cannot reproduce them. What do you do?

Short answer: Make every request traceable: structured logs with a request ID, error tracking (Sentry or similar) with the full stack trace, and metrics that show when and where errors happen. Then look for patterns: one endpoint, one server, one time of day, one type of input.

# structured_logging.py
import json
import logging
import sys
import uuid

class JsonFormatter(logging.Formatter):
    def format(self, record):
        entry = {"level": record.levelname, "message": record.getMessage(), "request_id": getattr(record, "request_id", "-")}
        if record.exc_info:
            entry["error"] = repr(record.exc_info[1])
        return json.dumps(entry)

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
logger = logging.getLogger("api")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

def handle(payload):
    request_id = "req-" + uuid.uuid5(uuid.NAMESPACE_URL, str(payload)).hex[:6]
    extra = {"request_id": request_id}
    logger.info("order received", extra=extra)
    try:
        return 100 / payload["quantity"]
    except Exception:
        logger.exception("order failed", extra=extra)
        return None

handle({"quantity": 4})
handle({"quantity": 0})          # the rare input that causes the "random" 500

{"level": "INFO", "message": "order received", "request_id": "req-4d134c"}
{"level": "INFO", "message": "order received", "request_id": "req-d069a9"}
{"level": "ERROR", "message": "order failed", "request_id": "req-d069a9", "error": "ZeroDivisionError('division by zero')"}

With logs like this, the “random” error turns out to be a specific input (quantity 0) on a specific request. Common real causes: unexpected nulls, timeouts to a third-party API, database connection limits under load, and race conditions that appear only with real traffic.

Q60 A mobile client retries a payment request and creates duplicate orders. How do you prevent it?

Short answer: Make the endpoint idempotent: the client sends a unique Idempotency-Key header, the server stores the key with the first response, and any retry with the same key returns the stored response instead of creating a new order. A unique database constraint is the final safety net.

# idempotency.py
from fastapi import FastAPI, Header
from fastapi.testclient import TestClient

app = FastAPI()
orders = []
responses_by_key = {}                 # use Redis or a DB table with an expiry in production

@app.post("/payments", status_code=201)
def create_payment(amount: int, idempotency_key: str = Header(...)):
    if idempotency_key in responses_by_key:
        return responses_by_key[idempotency_key]       # a retry: same answer, no new order
    order = {"order_id": len(orders) + 1, "amount": amount}
    orders.append(order)
    responses_by_key[idempotency_key] = order
    return order

client = TestClient(app)
headers = {"Idempotency-Key": "pay-7f3a"}
print(client.post("/payments?amount=999", headers=headers).json())
print(client.post("/payments?amount=999", headers=headers).json())   # network retry
print("orders created:", len(orders))

{'order_id': 1, 'amount': 999}
{'order_id': 1, 'amount': 999}
orders created: 1

Quick Revision: Q56–Q60

Tips to Crack Your Python Developer Interview

• Say “I/O-bound or CPU-bound?” out loud before any concurrency answer.
• Bring one production story: a slow query you fixed, a bug you traced, a deployment that went wrong and what you changed afterwards.
• Know your framework’s internals one level deeper than the tutorial: middleware, ORM queries, dependency injection.
• Write tests in the live round if you have time; it shows professional habits.
• Be honest: “I have not used Celery in production, but here is how I would approach it” beats guessing.

Final Thoughts

At 2–4 years, interviewers are not looking for someone who has memorised every API. They want a developer who understands why threads do not speed up CPU work, why an ORM loop can be slow, and how to find the real cause of a production problem. If you can explain the outputs in this guide in your own words, you are well prepared.

Conclusion

Together with Part 1, these Python developer interview questions for 2–4 years experience cover the whole interview: core Python, coding problems, multithreading, async programming, FastAPI, Django, Flask and real-world scenarios. Run the examples, change them and break them; that is how the answers become yours. If you are moving towards AI roles, continue with our AI Agent Developer interview questions.

Which Python question surprised you most in an interview? Share it in the comments below, so we can add it and help other developers prepare.