Advanced Topics &
Professional Mastery

Table of Contents

I. Graph Algorithms in Python

1.1 Graph Representations

# Adjacency list (most common in Python)
graph = {
    "A": ["B", "C"],
    "B": ["A", "D", "E"],
    "C": ["A", "F"],
    "D": ["B"],
    "E": ["B", "F"],
    "F": ["C", "E"],
}

# Weighted graph
weighted = {
    "A": [("B", 4), ("C", 2)],
    "B": [("A", 4), ("C", 5), ("D", 10)],
    "C": [("A", 2), ("B", 5), ("D", 3)],
    "D": [("B", 10), ("C", 3)],
}

1.2 Breadth-First Search (BFS)

Explores level by level using a queue. Finds shortest paths in unweighted graphs.

from collections import deque

def bfs(graph, start):
    visited = {start}
    parent = {start: None}
    order = []
    q = deque([start])

    while q:
        node = q.popleft()
        order.append(node)
        for neighbour in graph[node]:
            if neighbour not in visited:
                visited.add(neighbour)
                parent[neighbour] = node
                q.append(neighbour)

    return order, parent

def shortest_path(parent, target):
    path = []
    while target is not None:
        path.append(target)
        target = parent[target]
    return path[::-1]

order, parent = bfs(graph, "A")
print("BFS order :", order)               # ['A', 'B', 'C', 'D', 'E', 'F']
print("A → F path:", shortest_path(parent, "F"))   # ['A', 'C', 'F']

1.3 Depth-First Search (DFS)

Explores as deep as possible before backtracking. Uses recursion or an explicit stack.

def dfs_recursive(graph, node, visited=None):
    if visited is None:
        visited = set()
    visited.add(node)
    print(node, end=" ")
    for neighbour in graph[node]:
        if neighbour not in visited:
            dfs_recursive(graph, neighbour, visited)
    return visited

def dfs_iterative(graph, start):
    visited = set()
    stack = [start]
    order = []
    while stack:
        node = stack.pop()
        if node not in visited:
            visited.add(node)
            order.append(node)
            stack.extend(reversed(graph[node]))
    return order

print("Recursive:", end=" ")
dfs_recursive(graph, "A")           # A B D E F C
print()
print("Iterative:", dfs_iterative(graph, "A"))   # ['A', 'B', 'D', 'E', 'F', 'C']

1.4 Cycle Detection

def has_cycle_undirected(graph):
    visited = set()

    def dfs(node, parent):
        visited.add(node)
        for neighbour in graph[node]:
            if neighbour not in visited:
                if dfs(neighbour, node):
                    return True
            elif neighbour != parent:
                return True
        return False

    for node in graph:
        if node not in visited:
            if dfs(node, None):
                return True
    return False

def has_cycle_directed(graph):
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {node: WHITE for node in graph}

    def dfs(node):
        color[node] = GRAY
        for neighbour in graph[node]:
            if color[neighbour] == GRAY:
                return True                 # back edge → cycle
            if color[neighbour] == WHITE and dfs(neighbour):
                return True
        color[node] = BLACK
        return False

    return any(color[n] == WHITE and dfs(n) for n in graph)

1.5 Topological Sort

Linear ordering of vertices such that for every edge \(u \to v\), \(u\) comes before \(v\). Only defined on DAGs (directed acyclic graphs).

from collections import deque

def topo_sort(graph):
    indegree = {node: 0 for node in graph}
    for node in graph:
        for neighbour in graph[node]:
            indegree[neighbour] += 1

    q = deque([n for n, d in indegree.items() if d == 0])
    order = []

    while q:
        node = q.popleft()
        order.append(node)
        for neighbour in graph[node]:
            indegree[neighbour] -= 1
            if indegree[neighbour] == 0:
                q.append(neighbour)

    if len(order) != len(graph):
        raise ValueError("Graph has a cycle — no topological order")
    return order

dag = {
    "shirt": ["tie", "belt"],
    "tie": ["jacket"],
    "belt": ["jacket"],
    "jacket": [],
    "pants": ["belt", "shoes"],
    "shoes": [],
}
print(topo_sort(dag))
# ['shirt', 'pants', 'tie', 'belt', 'shoes', 'jacket']

1.6 Dijkstra's Shortest Path

Finds shortest paths from a source in a graph with non-negative weights.

import heapq

def dijkstra(graph, start):
    dist = {node: float("inf") for node in graph}
    dist[start] = 0
    parent = {start: None}
    heap = [(0, start)]

    while heap:
        d, node = heapq.heappop(heap)
        if d > dist[node]:
            continue                        # stale entry
        for neighbour, weight in graph[node]:
            new_dist = d + weight
            if new_dist < dist[neighbour]:
                dist[neighbour] = new_dist
                parent[neighbour] = node
                heapq.heappush(heap, (new_dist, neighbour))

    return dist, parent

d, p = dijkstra(weighted, "A")
for node in sorted(d):
    print(f"A → {node}: {d[node]}")

# A → A: 0
# A → B: 4
# A → C: 2
# A → D: 5   (via C)

1.7 Union–Find (Disjoint Set Union)

Efficiently tracks connectivity. Used in Kruskal's MST algorithm, cycle detection, and clustering.

class UnionFind:
    def __init__(self, n):
        self.parent = list(range(n))
        self.rank = [0] * n

    def find(self, x):
        while self.parent[x] != x:
            self.parent[x] = self.parent[self.parent[x]]  # path compression
            x = self.parent[x]
        return x

    def union(self, a, b):
        ra, rb = self.find(a), self.find(b)
        if ra == rb:
            return False
        if self.rank[ra] < self.rank[rb]:
            ra, rb = rb, ra
        self.parent[rb] = ra
        if self.rank[ra] == self.rank[rb]:
            self.rank[ra] += 1
        return True

    def connected(self, a, b):
        return self.find(a) == self.find(b)

uf = UnionFind(6)
uf.union(0, 1); uf.union(1, 2); uf.union(3, 4)
print(uf.connected(0, 2))   # True
print(uf.connected(0, 3))   # False

1.8 Minimum Spanning Tree (Kruskal)

def kruskal(n, edges):
    """edges = list of (weight, u, v)."""
    edges.sort()
    uf = UnionFind(n)
    mst = []
    total = 0
    for w, u, v in edges:
        if uf.union(u, v):
            mst.append((u, v, w))
            total += w
        if len(mst) == n - 1:
            break
    return mst, total

edges = [(1, 0, 1), (4, 0, 2), (2, 1, 2), (5, 1, 3),
         (3, 2, 3), (7, 2, 4), (6, 3, 4)]
mst, total = kruskal(5, edges)
print("MST:", mst)
print("Total weight:", total)      # 1 + 2 + 3 + 6 = 12

1.9 Algorithm Complexity Summary

II. Advanced String Algorithms

2.1 The Naive Pattern Search — \(O(nm)\)

def naive_search(text, pattern):
    n, m = len(text), len(pattern)
    matches = []
    for i in range(n - m + 1):
        if text[i:i + m] == pattern:
            matches.append(i)
    return matches

print(naive_search("ababcabcab", "abc"))    # [2, 5]

2.2 KMP (Knuth–Morris–Pratt) — \(O(n + m)\)

Preprocesses the pattern to build a failure function (LPS array) so that after a mismatch, the search resumes without re-checking characters.

def build_lps(pattern):
    """Longest proper prefix which is also a suffix, for each position."""
    lps = [0] * len(pattern)
    length = 0
    i = 1
    while i < len(pattern):
        if pattern[i] == pattern[length]:
            length += 1
            lps[i] = length
            i += 1
        elif length != 0:
            length = lps[length - 1]
        else:
            lps[i] = 0
            i += 1
    return lps

def kmp_search(text, pattern):
    if not pattern:
        return []
    lps = build_lps(pattern)
    matches = []
    i = j = 0
    while i < len(text):
        if text[i] == pattern[j]:
            i += 1; j += 1
            if j == len(pattern):
                matches.append(i - j)
                j = lps[j - 1]
        elif j != 0:
            j = lps[j - 1]
        else:
            i += 1
    return matches

print(build_lps("ABABCABAB"))     # [0,0,1,2,0,1,2,3,4]
print(kmp_search("ABABDABACDABABCABAB", "ABABCABAB"))   # [10]

2.3 Rabin–Karp — Rolling Hash

def rabin_karp(text, pattern, base=256, mod=10**9 + 7):
    n, m = len(text), len(pattern)
    if m > n:
        return []

    # Pre-compute base^(m-1) mod
    high = pow(base, m - 1, mod)
    ph = 0
    th = 0
    for i in range(m):
        ph = (ph * base + ord(pattern[i])) % mod
        th = (th * base + ord(text[i])) % mod

    matches = []
    for i in range(n - m + 1):
        if ph == th and text[i:i + m] == pattern:
            matches.append(i)
        if i < n - m:
            th = (th - ord(text[i]) * high) * base + ord(text[i + m])
            th %= mod
    return matches

print(rabin_karp("the quick brown fox", "quick"))   # [4]

2.4 Trie (Prefix Tree)

A tree where each node represents a prefix. Enables \(O(m)\) insert and search (where \(m\) is word length).

class TrieNode:
    __slots__ = ("children", "is_end")
    def __init__(self):
        self.children = {}
        self.is_end = False

class Trie:
    def __init__(self):
        self.root = TrieNode()

    def insert(self, word):
        node = self.root
        for ch in word:
            node = node.children.setdefault(ch, TrieNode())
        node.is_end = True

    def search(self, word):
        node = self._find(word)
        return node is not None and node.is_end

    def starts_with(self, prefix):
        return self._find(prefix) is not None

    def _find(self, s):
        node = self.root
        for ch in s:
            if ch not in node.children:
                return None
            node = node.children[ch]
        return node

    def autocomplete(self, prefix):
        node = self._find(prefix)
        if node is None:
            return []
        results = []
        self._collect(node, prefix, results)
        return results

    def _collect(self, node, prefix, out):
        if node.is_end:
            out.append(prefix)
        for ch, child in node.children.items():
            self._collect(child, prefix + ch, out)

t = Trie()
for w in ["cat", "car", "card", "care", "dog", "do"]:
    t.insert(w)

print(t.search("cat"))            # True
print(t.search("ca"))             # False
print(t.autocomplete("car"))      # ['car', 'card', 'care']
print(t.autocomplete("do"))       # ['do', 'dog']

2.5 Longest Palindromic Substring — Manacher's Algorithm

Naive \(O(n^3)\), DP \(O(n^2)\), Manacher's \(O(n)\). Manacher is rarely required in interviews but good to know.

def longest_palindrome(s):
    """DP O(n^2) approach — clean and interview-friendly."""
    n = len(s)
    if n < 2:
        return s
    dp = [[False] * n for _ in range(n)]
    start, max_len = 0, 1

    for i in range(n):
        dp[i][i] = True

    for i in range(n - 1):
        if s[i] == s[i + 1]:
            dp[i][i + 1] = True
            start, max_len = i, 2

    for length in range(3, n + 1):
        for i in range(n - length + 1):
            j = i + length - 1
            if s[i] == s[j] and dp[i + 1][j - 1]:
                dp[i][j] = True
                start, max_len = i, length

    return s[start:start + max_len]

print(longest_palindrome("babad"))    # 'bab' or 'aba'
print(longest_palindrome("cbbd"))     # 'bb'

2.6 Algorithm Comparison

III. Large-Scale Data — Dask, Polars, PySpark

3.1 The Scaling Problem

When data exceeds available RAM, pandas fails. Three approaches scale Python data processing:

3.2 Polars — Fast Single-Machine DataFrames

pip install polars

import polars as pl

# Lazy API — the key to efficiency
lf = (
    pl.scan_csv("sales.csv")          # lazy, doesn't load yet
    .filter(pl.col("amount") > 0)
    .group_by("region")
    .agg([
        pl.col("amount").sum().alias("total"),
        pl.col("amount").mean().alias("avg"),
        pl.len().alias("orders"),
    ])
    .sort("total", descending=True)
)

df = lf.collect()                     # executes the plan efficiently
print(df)

3.3 Dask — Parallel pandas at Scale

pip install "dask[complete]"

import dask.dataframe as dd

# Dask reads CSV in chunks and builds a task graph
ddf = dd.read_csv("large_*.csv")

result = (
    ddf[ddf["amount"] > 0]
    .groupby("region")["amount"]
    .agg(["sum", "mean", "count"])
    .compute()                        # trigger execution
)
print(result)

3.4 PySpark — Distributed Big Data

pip install pyspark

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .appName("SalesAnalysis")
         .getOrCreate())

df = spark.read.csv("s3://bucket/sales/*.csv", header=True, inferSchema=True)

result = (
    df.filter(F.col("amount") > 0)
      .groupBy("region")
      .agg(
          F.sum("amount").alias("total"),
          F.avg("amount").alias("avg"),
          F.count("*").alias("orders"),
      )
      .orderBy(F.desc("total"))
)

result.show()
spark.stop()

3.5 Out-of-Core Processing with Generators

For simple cases, no framework is needed — process in chunks with a generator.

def read_in_chunks(filepath, chunk_size=100_000):
    with open(filepath) as f:
        chunk = []
        for line in f:
            chunk.append(line)
            if len(chunk) == chunk_size:
                yield chunk
                chunk = []
        if chunk:
            yield chunk

def count_errors(filepath):
    total = 0
    for chunk in read_in_chunks(filepath):
        total += sum(1 for line in chunk if "ERROR" in line)
    return total

# Memory usage stays constant regardless of file size.

3.6 DataFrame Libraries — Feature Comparison

IV. Deep Learning with PyTorch

4.1 Deep Learning vs Classical ML

pip install torch torchvision

4.2 Tensors — The Fundamental Data Structure

import torch

# Creation
a = torch.tensor([[1, 2, 3], [4, 5, 6]])
print(a.shape, a.dtype)              # torch.Size([2, 3]) torch.int64

# From NumPy
import numpy as np
b = torch.from_numpy(np.array([1.0, 2.0, 3.0]))
print(b.dtype)                       # torch.float32

# Zeros, ones, random
print(torch.zeros(2, 3))
print(torch.ones(2, 3))
print(torch.randn(2, 3))             # standard normal

# Operations (similar to NumPy)
x = torch.tensor([1.0, 2.0, 3.0])
y = torch.tensor([4.0, 5.0, 6.0])
print(x + y)                         # tensor([5., 7., 9.])
print(x * y)                         # element-wise
print(torch.dot(x, y))               # tensor(32.)

# GPU
device = "cuda" if torch.cuda.is_available() else "cpu"
x = x.to(device)

4.3 Autograd — Automatic Differentiation

import torch

x = torch.tensor(3.0, requires_grad=True)
y = x ** 2 + 2 * x + 1

y.backward()                         # computes dy/dx
print(x.grad)                        # tensor(8.)  →  2*3 + 2 = 8

4.4 Building a Neural Network

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

# 1. Model
class Net(nn.Module):
    def __init__(self, in_features, hidden, out_classes):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(in_features, hidden),
            nn.ReLU(),
            nn.Linear(hidden, hidden),
            nn.ReLU(),
            nn.Linear(hidden, out_classes),
        )

    def forward(self, x):
        return self.net(x)

# 2. Data
X = torch.randn(1000, 20)
y = torch.randint(0, 3, (1000,))
loader = DataLoader(TensorDataset(X, y), batch_size=32, shuffle=True)

# 3. Training setup
device = "cuda" if torch.cuda.is_available() else "cpu"
model = Net(20, 64, 3).to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)

# 4. Training loop
for epoch in range(5):
    total_loss = 0.0
    for xb, yb in loader:
        xb, yb = xb.to(device), yb.to(device)
        optimizer.zero_grad()
        logits = model(xb)
        loss = criterion(logits, yb)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    print(f"Epoch {epoch + 1}: loss = {total_loss / len(loader):.4f}")

4.5 The Training Loop — Six Steps

4.6 Convolutional Neural Networks (CNN)

class ConvNet(nn.Module):
    def __init__(self, num_classes=10):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(3, 32, kernel_size=3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),
            nn.Conv2d(32, 64, kernel_size=3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),
        )
        self.classifier = nn.Sequential(
            nn.Flatten(),
            nn.Linear(64 * 8 * 8, 128),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(128, num_classes),
        )

    def forward(self, x):
        return self.classifier(self.features(x))

4.7 Saving and Loading Models

# Save state dict (recommended)
torch.save(model.state_dict(), "model.pt")

# Load
model = Net(20, 64, 3)
model.load_state_dict(torch.load("model.pt", map_location="cpu"))
model.eval()

model.eval()
with torch.no_grad():
    sample = torch.randn(1, 20)
    logits = model(sample)
    predicted_class = logits.argmax(dim=1).item()
    print("Predicted:", predicted_class)

4.8 Common Building Blocks

V. LLM APIs & Generative AI

5.1 What is an LLM?

5.2 Basic LLM API Call (OpenAI-compatible)

pip install openai

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a helpful Python tutor."},
        {"role": "user",   "content": "Explain list comprehensions briefly."},
    ],
    temperature=0.7,
    max_tokens=200,
)

print(response.choices[0].message.content)

5.3 Chat Roles and Message Format

5.4 Streaming Responses

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Write a haiku about Python."}],
    stream=True,
)

for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)
print()

5.5 Structured Output with Pydantic

from pydantic import BaseModel
from openai import OpenAI

class Student(BaseModel):
    name: str
    roll: int
    cgpa: float

client = OpenAI()

text = "Aarav Kumar, roll number 101, has a CGPA of 8.7."

response = client.beta.chat.completions.parse(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "Extract the student information."},
        {"role": "user",   "content": text},
    ],
    response_format=Student,
)

student = response.choices[0].message.parsed
print(student.name, student.roll, student.cgpa)    # Aarav Kumar 101 8.7

5.6 Function Calling (Tools)

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"},
                    "unit": {"type": "string", "enum": ["c", "f"]},
                },
                "required": ["city"],
            },
        },
    }
]

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Weather in Delhi?"}],
    tools=tools,
)

tool_call = response.choices[0].message.tool_calls[0]
print(tool_call.function.name)              # get_weather
print(tool_call.function.arguments)         # {"city": "Delhi"}

# Call the actual function, then send the result back to the model.

5.7 Embeddings and Semantic Search

import numpy as np

def embed(texts):
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return np.array([d.embedding for d in response.data])

docs = [
    "Python is a high-level programming language.",
    "Cats are common household pets.",
    "Machine learning is a subset of AI.",
]
doc_vecs = embed(docs)

query = "What is Python?"
q_vec = embed([query])[0]

# Cosine similarity
similarities = doc_vecs @ q_vec / (
    np.linalg.norm(doc_vecs, axis=1) * np.linalg.norm(q_vec)
)
best = int(np.argmax(similarities))
print("Best match:", docs[best])

5.8 Retrieval-Augmented Generation (RAG)

def rag_answer(question, knowledge_base, k=3):
    # 1. Embed the question
    q_vec = embed([question])[0]

    # 2. Retrieve top-k chunks
    doc_vecs = embed(knowledge_base)
    sims = doc_vecs @ q_vec / (
        np.linalg.norm(doc_vecs, axis=1) * np.linalg.norm(q_vec)
    )
    top_k_idx = np.argsort(-sims)[:k]
    context = "\n\n".join(knowledge_base[i] for i in top_k_idx)

    # 3. Prompt the LLM with context
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system",
             "content": "Answer using ONLY the provided context."},
            {"role": "user",
             "content": f"Context:\n{context}\n\nQuestion: {question}"},
        ],
    )
    return response.choices[0].message.content

5.9 Prompt Engineering Basics

VI. GraphQL & gRPC

6.1 REST, GraphQL and gRPC Compared

6.2 GraphQL with Strawberry and FastAPI

pip install "strawberry-graphql[fastapi]" fastapi uvicorn

from typing import Optional
import strawberry
from strawberry.fastapi import GraphQLRouter
from fastapi import FastAPI

@strawberry.type
class Student:
    id: int
    name: str
    cgpa: float
    branch: str

DB = [
    Student(id=1, name="Aarav", cgpa=8.7, branch="CSE"),
    Student(id=2, name="Diya",  cgpa=9.3, branch="ECE"),
]

@strawberry.type
class Query:
    @strawberry.field
    def student(self, id: int) -> Optional[Student]:
        return next((s for s in DB if s.id == id), None)

    @strawberry.field
    def students(self, branch: Optional[str] = None) -> list[Student]:
        if branch:
            return [s for s in DB if s.branch == branch]
        return DB

@strawberry.type
class Mutation:
    @strawberry.mutation
    def add_student(self, name: str, cgpa: float, branch: str) -> Student:
        new = Student(id=len(DB) + 1, name=name, cgpa=cgpa, branch=branch)
        DB.append(new)
        return new

schema = strawberry.Schema(query=Query, mutation=Mutation)
graphql_app = GraphQLRouter(schema)

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

# Run: uvicorn main:app --reload
# GraphiQL UI: http://localhost:8000/graphql

Sample GraphQL query

query {
  students(branch: "CSE") {
    id
    name
    cgpa
  }
}

Sample mutation

mutation {
  addStudent(name: "Kabir", cgpa: 7.9, branch: "MEC") {
    id
    name
  }
}

6.3 gRPC with Python

pip install grpcio grpcio-tools

// student.proto
syntax = "proto3";

package student;

service StudentService {
  rpc GetStudent (StudentRequest) returns (StudentReply);
  rpc ListStudents (Empty) returns (stream StudentReply);
}

message StudentRequest { int32 id = 1; }
message Empty {}

message StudentReply {
  int32 id = 1;
  string name = 2;
  float cgpa = 3;
}

# Generate Python stubs
python -m grpc_tools.protoc -I. \
    --python_out=. --grpc_python_out=. student.proto

# server.py
import grpc
from concurrent import futures
import student_pb2, student_pb2_grpc

class StudentServicer(student_pb2_grpc.StudentServiceServicer):
    def GetStudent(self, request, context):
        return student_pb2.StudentReply(id=request.id, name="Aarav", cgpa=8.7)

    def ListStudents(self, request, context):
        for name in ["Aarav", "Diya", "Kabir"]:
            yield student_pb2.StudentReply(id=0, name=name, cgpa=8.0)

server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
student_pb2_grpc.add_StudentServiceServicer_to_server(StudentServicer(), server)
server.add_insecure_port("[::]:50051")
server.start()
server.wait_for_termination()

# client.py
import grpc
import student_pb2, student_pb2_grpc

with grpc.insecure_channel("localhost:50051") as channel:
    stub = student_pb2_grpc.StudentServiceStub(channel)
    reply = stub.GetStudent(student_pb2.StudentRequest(id=1))
    print(reply.name, reply.cgpa)         # Aarav 8.7

    for s in stub.ListStudents(student_pb2.Empty()):
        print("Streamed:", s.name)

6.4 When to Choose What

VII. Property-Based & Mutation Testing

7.1 The Limits of Example-Based Testing

7.2 Hypothesis — Property-Based Testing

pip install hypothesis

from hypothesis import given, strategies as st

def sort_list(xs):
    return sorted(xs)

@given(st.lists(st.integers()))
def test_sort_preserves_length(xs):
    assert len(sort_list(xs)) == len(xs)

@given(st.lists(st.integers()))
def test_sort_is_ordered(xs):
    result = sort_list(xs)
    for i in range(len(result) - 1):
        assert result[i] <= result[i + 1]

@given(st.lists(st.integers()))
def test_sort_preserves_multiset(xs):
    from collections import Counter
    assert Counter(sort_list(xs)) == Counter(xs)

@given(st.lists(st.integers()), st.lists(st.integers()))
def test_sort_idempotent(xs, ys):
    combined = xs + ys
    assert sort_list(combined) == sort_list(sort_list(combined))

7.3 Strategies — Generating Data

7.4 Composite Strategies

from hypothesis import strategies as st
from hypothesis import given

@st.composite
def student(draw):
    return {
        "name":   draw(st.text(min_size=1, max_size=20)),
        "age":    draw(st.integers(min_value=15, max_value=100)),
        "cgpa":   draw(st.floats(min_value=0, max_value=10)),
    }

@given(student())
def test_student_has_valid_fields(s):
    assert 1 <= len(s["name"]) <= 20
    assert 15 <= s["age"] <= 100
    assert 0 <= s["cgpa"] <= 10

7.5 Mutation Testing — Testing the Tests

pip install mutmut

mutmut run --paths-to-mutate=src/
mutmut results
mutmut show 3

Common mutation types

7.6 Fuzz Testing

pip install atheris

# Atheris is Google's coverage-guided fuzzer for Python.
# import atheris
# import sys
#
# def TestOneInput(data):
#     fdp = atheris.FuzzedDataProvider(data)
#     text = fdp.ConsumeUnicodeNoSurrogates(100)
#     # Call code that should not crash on any input.
#     try:
#         import json
#         json.loads(text)
#     except json.JSONDecodeError:
#         pass
#
# atheris.Setup(sys.argv, TestOneInput)
# atheris.Fuzz()

7.7 Testing Strategy — The Pyramid

VIII. System Design for Python Services

8.1 The System Design Interview

8.2 Back-of-the-Envelope Numbers

8.3 Standard Building Blocks

8.4 Caching Strategies

import redis, json, hashlib

cache = redis.Redis(host="localhost", port=6379, decode_responses=True)

def get_student(db, student_id):
    key = f"student:{student_id}"
    cached = cache.get(key)
    if cached:
        return json.loads(cached)          # cache hit

    student = db.get(student_id)           # cache miss
    if student:
        cache.setex(key, 300, json.dumps(student))   # 5-min TTL
    return student

8.5 Database Patterns

# SQLAlchemy connection pool
from sqlalchemy import create_engine

engine = create_engine(
    "postgresql://user:pass@host/db",
    pool_size=20,           # persistent connections
    max_overflow=10,        # extra connections under load
    pool_pre_ping=True,     # verify connection is alive before use
    pool_recycle=1800,      # recycle connections after 30 min
)

8.6 Resilience Patterns

pip install pybreaker

import pybreaker
import requests

breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=30)

@breaker
def call_remote_api():
    r = requests.get("https://api.example.com/data", timeout=5)
    r.raise_for_status()
    return r.json()

# After 5 failures, the circuit opens for 30 seconds and calls fail fast.

8.7 Worked Design — URL Shortener

import string, hashlib

ALPHABET = string.digits + string.ascii_letters     # 62 chars

def encode_base62(num: int) -> str:
    if num == 0:
        return ALPHABET[0]
    out = []
    while num > 0:
        num, rem = divmod(num, 62)
        out.append(ALPHABET[rem])
    return "".join(reversed(out))

def shorten(long_url: str) -> str:
    digest = hashlib.sha256(long_url.encode()).hexdigest()
    numeric = int(digest[:16], 16)             # 64-bit slice
    return encode_base62(numeric)[:7]

print(shorten("https://example.com/very/long/path"))
# e.g., 'jK9aBc2'

IX. Advanced Patterns — CQRS, Event Sourcing, Saga

9.1 CQRS — Command Query Responsibility Segregation

class CommandHandler:
    """Handles writes: validates and persists."""
    def __init__(self, db, event_bus):
        self.db = db
        self.event_bus = event_bus

    def create_task(self, task_id, title, owner):
        self.db.execute(
            "INSERT INTO tasks (id, title, owner) VALUES (?, ?, ?)",
            (task_id, title, owner))
        self.event_bus.publish("task.created",
                               {"id": task_id, "title": title, "owner": owner})

class QueryHandler:
    """Handles reads: uses a denormalised projection."""
    def __init__(self, read_db):
        self.read_db = read_db

    def list_tasks(self, owner):
        return self.read_db.execute(
            "SELECT id, title FROM task_view WHERE owner = ?", (owner,)).fetchall()

9.2 Event Sourcing

from dataclasses import dataclass, field
from typing import List
from datetime import datetime

@dataclass
class Event:
    type: str
    data: dict
    ts: datetime = field(default_factory=datetime.utcnow)

class BankAccount:
    def __init__(self, account_id):
        self.id = account_id
        self.balance = 0
        self.events: List[Event] = []

    def _apply(self, event: Event):
        if event.type == "deposited":
            self.balance += event.data["amount"]
        elif event.type == "withdrawn":
            self.balance -= event.data["amount"]

    def deposit(self, amount):
        if amount <= 0:
            raise ValueError("Deposit must be positive")
        ev = Event("deposited", {"amount": amount})
        self.events.append(ev)
        self._apply(ev)

    def withdraw(self, amount):
        if amount > self.balance:
            raise ValueError("Insufficient funds")
        ev = Event("withdrawn", {"amount": amount})
        self.events.append(ev)
        self._apply(ev)

    @classmethod
    def replay(cls, account_id, events):
        acc = cls(account_id)
        for ev in events:
            acc._apply(ev)
        acc.events = events
        return acc

acc = BankAccount("A1")
acc.deposit(1000)
acc.withdraw(300)
print(acc.balance)               # 700

# Rebuild the same state from the event log
rebuilt = BankAccount.replay("A1", acc.events)
print(rebuilt.balance)           # 700

9.3 Saga Pattern — Distributed Transactions

Long-running business processes that span multiple services must handle partial failure. A saga is a sequence of local transactions, each with a compensating action for rollback.

class OrderSaga:
    def __init__(self, payment, inventory, shipping):
        self.payment = payment
        self.inventory = inventory
        self.shipping = shipping
        self.compensations = []

    def run(self, order):
        try:
            self.payment.charge(order)
            self.compensations.append(lambda: self.payment.refund(order))

            self.inventory.reserve(order)
            self.compensations.append(lambda: self.inventory.release(order))

            self.shipping.schedule(order)
            return "success"

        except Exception as e:
            print(f"Saga failed: {e}. Rolling back...")
            for undo in reversed(self.compensations):
                try:
                    undo()
                except Exception as undo_err:
                    print(f"Compensation failed: {undo_err}")
            return "rolled_back"

9.4 Outbox Pattern — Reliable Event Publishing

-- Write the domain change and the event in ONE transaction.
BEGIN;
  UPDATE accounts SET balance = balance - 100 WHERE id = 1;
  INSERT INTO outbox (event_type, payload, status)
       VALUES ('account.debited', '{"account_id": 1, "amount": 100}', 'pending');
COMMIT;

-- A separate worker publishes outbox rows to Kafka/RabbitMQ and marks them 'sent'.

Problem solved: without the outbox, a crash between DB write and message publish causes an event to be lost.

9.5 Pattern Selection Guide

X. Python in the Cloud — Serverless, IaC

10.1 Compute Models in the Cloud

10.2 AWS Lambda with Python

# handler.py
import json

def lambda_handler(event, context):
    """Entry point for AWS Lambda."""
    name = event.get("queryStringParameters", {}).get("name", "World")
    return {
        "statusCode": 200,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"message": f"Hello, {name}!"}),
    }

# Deployment with the AWS SAM CLI
# 1. Install SAM CLI, then initialise the project:
sam init --runtime python3.12

# 2. Build and deploy
sam build
sam deploy --guided

10.3 Infrastructure as Code with Terraform

# main.tf
provider "aws" { region = "ap-south-1" }

resource "aws_lambda_function" "hello" {
  function_name = "hello-python"
  role          = aws_iam_role.lambda_role.arn
  handler       = "handler.lambda_handler"
  runtime       = "python3.12"
  filename      = "package.zip"
  timeout       = 30
  memory_size   = 256
}

resource "aws_iam_role" "lambda_role" {
  name = "lambda-role"
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Action    = "sts:AssumeRole"
      Effect    = "Allow"
      Principal = { Service = "lambda.amazonaws.com" }
    }]
  })
}

terraform init
terraform plan
terraform apply

10.4 Serverless Framework

# serverless.yml
service: task-api
provider:
  name: aws
  runtime: python3.12
  region: ap-south-1
  environment:
    DATABASE_URL: ${env:DATABASE_URL}

functions:
  api:
    handler: handler.lambda_handler
    events:
      - httpApi:
          path: /{proxy+}
          method: ANY

serverless deploy
serverless logs -f api --tail

10.5 Cold Starts and Optimisation

# Lazy imports to reduce cold-start time
import json

def lambda_handler(event, context):
    # Only import heavy libraries when actually needed
    import pandas as pd
    df = pd.DataFrame({"a": [1, 2, 3]})
    return {"statusCode": 200, "body": json.dumps({"sum": int(df["a"].sum())})}

10.6 Cost Optimisation

10.7 Observability in the Cloud

# Structured logging for CloudWatch
import json, logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    logger.info(json.dumps({
        "request_id": context.aws_request_id,
        "path":       event.get("path"),
        "method":     event.get("httpMethod"),
    }))
    return {"statusCode": 200, "body": "ok"}

AWS X-Ray, CloudWatch Logs Insights, and Datadog's Lambda extension provide tracing and metrics for serverless functions.

10.8 Choosing a Deployment Model

XI. Technical Interview Preparation

11.1 The Interview Loop

11.2 The DSA Problem-Solving Framework

11.3 Python-Specific Interview Questions

# BROKEN
def append_to(value, target=[]):
    target.append(value)
    return target

print(append_to(1))         # [1]
print(append_to(2))         # [1, 2]  ← surprise!

# FIXED
def append_to(value, target=None):
    if target is None:
        target = []
    target.append(value)
    return target

print(append_to(1))         # [1]
print(append_to(2))         # [2]

11.4 Behavioural Questions — STAR

Common prompts

• Tell me about a time you solved a difficult bug.
• Describe a conflict with a teammate and how you resolved it.
• What is the most impactful project you have worked on?
• Tell me about a time you failed and what you learned.
• How do you stay updated with technology?

11.5 Coding Round Checklist

11.6 Recommended Practice Plan

XII. Capstone — End-to-End Production System

A complete reference project tying together every concept in Units I–VI. This is what a "professional Python engineer" ships.

12.1 System Overview

12.2 Architecture

Client → ALB → FastAPI (ECS) ─┬─ Postgres (metadata)
                              ├─ Redis (cache)
                              ├─ Qdrant / pgvector (embeddings)
                              └─ OpenAI API (LLM)

Background workers:
- Ingest worker  (queue-driven) → chunk + embed + upsert
- Cleanup worker (scheduled)     → prune old embeddings

Observability:
- Prometheus /metrics → Grafana
- Structured JSON logs → CloudWatch
- OpenTelemetry → Jaeger

12.3 Project Structure

doc-qa/
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
├── .env.example
├── src/
│   └── docqa/
│       ├── __init__.py
│       ├── main.py             # FastAPI app
│       ├── config.py           # Settings (pydantic-settings)
│       ├── database.py         # SQLAlchemy engine + session
│       ├── models.py           # ORM models
│       ├── schemas.py          # Pydantic
│       ├── rag.py              # Chunk + embed + retrieve + answer
│       ├── cache.py            # Redis wrapper
│       ├── metrics.py          # Prometheus instrumentation
│       ├── logging_config.py   # JSON logs
│       ├── workers/
│       │   ├── ingest.py
│       │   └── cleanup.py
│       └── auth.py             # JWT
└── tests/
    ├── conftest.py
    ├── test_api.py
    ├── test_rag.py
    └── test_auth.py

12.4 Configuration

# src/docqa/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url:   str = "postgresql://user:pass@db:5432/docqa"
    redis_url:      str = "redis://cache:6379/0"
    openai_api_key: str
    jwt_secret:     str
    jwt_expire_min: int = 60
    chunk_size:     int = 500
    chunk_overlap:  int = 80
    top_k:          int = 4
    log_level:      str = "INFO"

    class Config:
        env_file = ".env"

settings = Settings()

12.5 RAG Module

# src/docqa/rag.py
import numpy as np
from openai import OpenAI
from .config import settings

client = OpenAI(api_key=settings.openai_api_key)

def chunk(text, size=500, overlap=80):
    words = text.split()
    step  = size - overlap
    return [
        " ".join(words[i:i + size])
        for i in range(0, len(words), step)
        if words[i:i + size]
    ]

def embed(texts):
    resp = client.embeddings.create(
        model="text-embedding-3-small", input=texts)
    return np.array([d.embedding for d in resp.data])

def answer(question, documents):
    q_vec = embed([question])[0]
    d_vec = embed(documents)
    sims = d_vec @ q_vec / (
        np.linalg.norm(d_vec, axis=1) * np.linalg.norm(q_vec))
    top = np.argsort(-sims)[: settings.top_k]
    context = "\n\n".join(documents[i] for i in top)

    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system",
             "content": "Answer using ONLY the provided context."},
            {"role": "user",
             "content": f"Context:\n{context}\n\nQuestion: {question}"},
        ],
    )
    return resp.choices[0].message.content, list(top)

12.6 API Layer

# src/docqa/main.py
from fastapi import FastAPI, Depends, HTTPException, Request
from sqlalchemy.orm import Session
from .config import settings
from .database import get_db, Base, engine
from .schemas import (
    IngestRequest, IngestResponse, QueryRequest, QueryResponse)
from .rag import chunk, embed, answer
from .metrics import REQUEST_COUNT, REQUEST_LATENCY
from .logging_config import logger, configure_logging
from .auth import current_user

Base.metadata.create_all(engine)
configure_logging()

app = FastAPI(title="Doc QA", version="1.0.0")

# In-memory doc store (in production: pgvector)
DOCS: dict[str, list[str]] = {}

@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    REQUEST_COUNT.labels(request.method, request.url.path).inc()
    with REQUEST_LATENCY.labels(request.url.path).time():
        response = await call_next(request)
    logger.info("http.request",
                extra={"method": request.method, "path": request.url.path,
                       "status": response.status_code})
    return response

@app.get("/health")
def health():
    return {"status": "ok"}

@app.post("/ingest", response_model=IngestResponse)
def ingest(payload: IngestRequest,
           user: str = Depends(current_user)):
    chunks = chunk(payload.text, settings.chunk_size, settings.chunk_overlap)
    DOCS[payload.doc_id] = chunks
    embed(chunks)         # persist embeddings in production
    return {"doc_id": payload.doc_id, "chunks": len(chunks)}

@app.post("/query", response_model=QueryResponse)
def query(payload: QueryRequest,
          user: str = Depends(current_user)):
    chunks = DOCS.get(payload.doc_id)
    if not chunks:
        raise HTTPException(404, "Document not found")
    text, indices = answer(payload.question, chunks)
    return {"answer": text, "source_chunks": indices}

12.7 Tests

# tests/test_api.py
from fastapi.testclient import TestClient
from unittest.mock import patch
from src.docqa.main import app

client = TestClient(app)

def auth_header(token="valid"):
    return {"Authorization": f"Bearer {token}"}

@patch("src.docqa.auth.current_user", return_value="test-user")
@patch("src.docqa.rag.embed")
def test_ingest(mock_embed, mock_user):
    mock_embed.return_value = [[0.0] * 8]
    r = client.post("/ingest",
                    json={"doc_id": "d1", "text": "Python is great. " * 20},
                    headers=auth_header())
    assert r.status_code == 200
    assert r.json()["doc_id"] == "d1"

@patch("src.docqa.auth.current_user", return_value="test-user")
@patch("src.docqa.rag.answer", return_value=("Python is great.", [0]))
def test_query(mock_answer, mock_user):
    r = client.post("/query",
                    json={"doc_id": "d1", "question": "What is Python?"},
                    headers=auth_header())
    assert r.status_code in (200, 404)

12.8 Docker Compose

version: "3.9"
services:
  api:
    build: .
    ports: ["8000:8000"]
    environment:
      DATABASE_URL: postgresql://user:pass@db:5432/docqa
      REDIS_URL: redis://cache:6379/0
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      JWT_SECRET: ${JWT_SECRET}
    depends_on: [db, cache]
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: docqa
    volumes: [pgdata:/var/lib/postgresql/data]
  cache:
    image: redis:7-alpine
volumes:
  pgdata:

12.9 Production Checklist

XIII. Summary & Quick Reference Sheet

13.1 Graph Algorithms Cheat Sheet

13.2 String Algorithms Cheat Sheet

13.3 Data Scale Cheat Sheet

13.4 Deep Learning Cheat Sheet

13.5 API Paradigm Cheat Sheet

13.6 System Design Cheat Sheet

13.7 Testing Cheat Sheet

13.8 Interview DSA Pattern Cheat Sheet

XIV. Exam Tips & Practice Questions

Top 12 Exam Tips

1. Graphs: always state whether the graph is weighted and directed before choosing BFS/DFS/Dijkstra.
2. Complexity first. Write the Big-O of every algorithm before you code it.
3. KMP vs Rabin–Karp: KMP for worst-case guarantee, Rabin–Karp when searching many patterns or when the rolling hash gives practical speed.
4. Scale questions: Polars for single-machine speed, Dask for pandas code outgrowing RAM, Spark for cluster-scale.
5. PyTorch: the training loop has six steps; state them in order if asked.
6. LLM APIs: mention temperature, max_tokens, streaming, and structured output (Pydantic).
7. REST vs GraphQL vs gRPC: never claim one is universally better — state the trade-off.
8. Property-based testing: name at least three properties — invariant, idempotence, roundtrip.
9. System design: clarify requirements, estimate scale, then draw boxes. State the bottleneck.
10. CQRS/Event Sourcing/Saga: only when the problem demands them. Explain why for smaller systems REST + RDBMS is simpler.
11. Cloud: name one service per compute model (EC2 = IaaS, ECS = CaaS, Lambda = FaaS).
12. Interviews: narrate your thinking. Communication is half the score.

Practice Questions

XV. Full Solutions to Practice Questions

Solution 1 — Shortest path with BFS

from collections import deque

def shortest_path(graph, start, target):
    if start == target:
        return [start]

    parent = {start: None}
    q = deque([start])

    while q:
        node = q.popleft()
        for neighbour in graph[node]:
            if neighbour not in parent:
                parent[neighbour] = node
                if neighbour == target:
                    path = []
                    while neighbour is not None:
                        path.append(neighbour)
                        neighbour = parent[neighbour]
                    return path[::-1]
                q.append(neighbour)
    return None

g = {"A": ["B", "C"], "B": ["D"], "C": ["D", "E"], "D": ["F"], "E": ["F"], "F": []}
print(shortest_path(g, "A", "F"))       # ['A', 'B', 'D', 'F']

Solution 2 — Cycle detection (two approaches)

def has_cycle_dfs(graph):
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {n: WHITE for n in graph}

    def dfs(n):
        color[n] = GRAY
        for m in graph[n]:
            if color[m] == GRAY:
                return True
            if color[m] == WHITE and dfs(m):
                return True
        color[n] = BLACK
        return False

    return any(color[n] == WHITE and dfs(n) for n in graph)

from collections import deque

def has_cycle_kahn(graph):
    indeg = {n: 0 for n in graph}
    for n in graph:
        for m in graph[n]:
            indeg[m] += 1
    q = deque([n for n, d in indeg.items() if d == 0])
    seen = 0
    while q:
        n = q.popleft(); seen += 1
        for m in graph[n]:
            indeg[m] -= 1
            if indeg[m] == 0:
                q.append(m)
    return seen != len(graph)

dag = {"A": ["B"], "B": ["C"], "C": []}
cyc = {"A": ["B"], "B": ["C"], "C": ["A"]}
print(has_cycle_dfs(dag),  has_cycle_kahn(dag))   # False False
print(has_cycle_dfs(cyc),  has_cycle_kahn(cyc))   # True True

Solution 3 — Dijkstra with path reconstruction

import heapq

def dijkstra_path(graph, start, target):
    dist = {n: float("inf") for n in graph}
    dist[start] = 0
    parent = {start: None}
    heap = [(0, start)]

    while heap:
        d, node = heapq.heappop(heap)
        if node == target:
            break
        if d > dist[node]:
            continue
        for nb, w in graph[node]:
            nd = d + w
            if nd < dist[nb]:
                dist[nb] = nd
                parent[nb] = node
                heapq.heappush(heap, (nd, nb))

    if dist[target] == float("inf"):
        return None, float("inf")

    path = []
    cur = target
    while cur is not None:
        path.append(cur)
        cur = parent[cur]
    return path[::-1], dist[target]

g = {
    "A": [("B", 4), ("C", 2)],
    "B": [("C", 5), ("D", 10)],
    "C": [("D", 3)],
    "D": [],
}
print(dijkstra_path(g, "A", "D"))    # (['A', 'C', 'D'], 5)

Solution 4 — Trie

class TrieNode:
    __slots__ = ("children", "is_end")
    def __init__(self):
        self.children = {}
        self.is_end = False

class Trie:
    def __init__(self):
        self.root = TrieNode()

    def insert(self, word):
        node = self.root
        for ch in word:
            node = node.children.setdefault(ch, TrieNode())
        node.is_end = True

    def search(self, word):
        node = self._find(word)
        return node is not None and node.is_end

    def _find(self, prefix):
        node = self.root
        for ch in prefix:
            if ch not in node.children:
                return None
            node = node.children[ch]
        return node

    def autocomplete(self, prefix):
        node = self._find(prefix)
        results = []
        if node:
            self._collect(node, prefix, results)
        return results

    def _collect(self, node, prefix, out):
        if node.is_end:
            out.append(prefix)
        for ch, child in node.children.items():
            self._collect(child, prefix + ch, out)

t = Trie()
for w in ["cat", "car", "card", "care", "dog"]:
    t.insert(w)
print(t.search("car"))                # True
print(t.autocomplete("car"))          # ['car', 'card', 'care']

Solution 5 — KMP

def build_lps(p):
    lps = [0] * len(p)
    length = 0; i = 1
    while i < len(p):
        if p[i] == p[length]:
            length += 1; lps[i] = length; i += 1
        elif length != 0:
            length = lps[length - 1]
        else:
            lps[i] = 0; i += 1
    return lps

def kmp(text, pattern):
    lps = build_lps(pattern)
    i = j = 0
    while i < len(text):
        if text[i] == pattern[j]:
            i += 1; j += 1
            if j == len(pattern):
                return i - j
        elif j != 0:
            j = lps[j - 1]
        else:
            i += 1
    return -1

print(build_lps("ABABCABAB"))
# [0, 0, 1, 2, 0, 1, 2, 3, 4]

Solution 6 — Polars lazy query

import polars as pl

result = (
    pl.scan_csv("sales.csv")
    .filter(pl.col("amount") > 100)
    .group_by("region")
    .agg([
        pl.col("amount").sum().alias("total"),
        pl.col("amount").mean().alias("avg"),
    ])
    .sort("total", descending=True)
    .collect()
)
print(result)

Solution 7 — Framework choice for 500 GB

Recommendation: PySpark, deployed on a managed cluster (Databricks, EMR), with partitioning on the date column. Use Dask if the workload is lighter and the team is Python-first.

Solution 8 — PyTorch training loop

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

class Net(nn.Module):
    def __init__(self, d_in, d_h, n_classes):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(d_in, d_h), nn.ReLU(),
            nn.Linear(d_h, d_h), nn.ReLU(),
            nn.Linear(d_h, n_classes))
    def forward(self, x):
        return self.net(x)

X = torch.randn(2000, 10)
y = torch.randint(0, 3, (2000,))
loader = DataLoader(TensorDataset(X, y), batch_size=32, shuffle=True)

model = Net(10, 64, 3)
opt   = optim.Adam(model.parameters(), lr=1e-3)
loss_fn = nn.CrossEntropyLoss()

for epoch in range(5):
    for xb, yb in loader:
        opt.zero_grad()                    # 1. zero gradients
        logits = model(xb)                 # 2. forward pass
        loss = loss_fn(logits, yb)         # 3. compute loss
        loss.backward()                    # 4. backward pass (autograd)
        opt.step()                         # 5. update parameters
    print(f"Epoch {epoch + 1}: loss = {loss.item():.4f}")   # 6. log

Solution 9 — LLM call with structured output and fallback

import os, time
from pydantic import BaseModel
from openai import OpenAI, APIError, RateLimitError

class Extracted(BaseModel):
    name: str
    roll: int
    cgpa: float

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def extract(text, retries=3):
    for attempt in range(1, retries + 1):
        try:
            resp = client.beta.chat.completions.parse(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": "Extract student info."},
                    {"role": "user",   "content": text},
                ],
                response_format=Extracted,
            )
            return resp.choices[0].message.parsed
        except RateLimitError:
            time.sleep(2 ** attempt)
        except APIError as e:
            print(f"API error: {e}")
            if attempt == retries:
                raise
    return None

Solution 10 — REST vs GraphQL vs gRPC scenarios

Solution 11 — Hypothesis roundtrip test

from hypothesis import given, strategies as st

def encode(s: str) -> bytes:
    return s.encode("utf-8")

def decode(b: bytes) -> str:
    return b.decode("utf-8")

@given(st.text())
def test_roundtrip(s):
    assert decode(encode(s)) == s

@given(st.text())
def test_encoded_is_bytes(s):
    assert isinstance(encode(s), bytes)

@given(st.text())
def test_length_grows_or_stays(s):
    # UTF-8 never shrinks a string's byte length below its character count
    assert len(encode(s)) >= len(s)

Solution 12 — URL shortener design

Functional requirements: create short URL, redirect, optional analytics.

Non-functional: 100M URLs total, 10k reads/s, 100 writes/s, sub-100 ms p99 reads.

Data model:

CREATE TABLE urls (
    short_code  VARCHAR(8) PRIMARY KEY,
    long_url    TEXT       NOT NULL,
    owner_id    BIGINT,
    created_at  TIMESTAMP  DEFAULT NOW(),
    clicks      BIGINT     DEFAULT 0
);
CREATE INDEX idx_urls_owner ON urls(owner_id);

ID generation: Snowflake-like ID (timestamp + machine + sequence) → Base62 encode → take first 7 characters.

Read path: check Redis; on hit, redirect; on miss, query Postgres, populate Redis with 24h TTL, redirect.

Write path: validate URL, generate code, insert with unique constraint, retry on collision.

Bottleneck: DB write throughput at scale — shard by hash of short code.

Trade-off: 301 (permanent) is cache-friendly but loses click data; 302 (temporary) tracks clicks but reduces CDN/browser caching.

Solution 13 — CQRS and Event Sourcing

from dataclasses import dataclass, field

@dataclass
class Event:
    type: str
    data: dict

class Account:
    def __init__(self, account_id):
        self.id = account_id
        self.balance = 0
        self.events: list[Event] = []

    def _apply(self, e):
        if e.type == "deposited":
            self.balance += e.data["amount"]
        elif e.type == "withdrawn":
            self.balance -= e.data["amount"]

    def deposit(self, amount):
        e = Event("deposited", {"amount": amount})
        self.events.append(e)
        self._apply(e)

    def withdraw(self, amount):
        if amount > self.balance:
            raise ValueError("Insufficient funds")
        e = Event("withdrawn", {"amount": amount})
        self.events.append(e)
        self._apply(e)

# --- Command side ---
acc = Account("A1")
acc.deposit(1000); acc.withdraw(250)

# --- Query side (projection) ---
def project(events):
    balance = 0
    for e in events:
        balance += e.data["amount"] if e.type == "deposited" else -e.data["amount"]
    return {"id": "A1", "balance": balance}

print(project(acc.events))     # {'id': 'A1', 'balance': 750}

CQRS solves: read and write workloads scale differently. Event sourcing solves: full audit trail and time travel, plus projection flexibility (multiple read models from the same events).

Solution 14 — EC2 vs ECS vs Lambda

Solution 15 — Two-sum in \(O(n)\)

def two_sum(nums, target):
    seen = {}
    for i, n in enumerate(nums):
        complement = target - n
        if complement in seen:
            return [seen[complement], i]
        seen[n] = i
    return []

print(two_sum([2, 7, 11, 15], 9))    # [0, 1]

Pattern: hash map for \(O(1)\) complement lookup. Time \(O(n)\), space \(O(n)\). This is the canonical "trade space for time" example.

Solution 16 — Mutable default argument bug

# BROKEN — the default list is created once and shared across calls
def add_item(item, target=[]):
    target.append(item)
    return target

print(add_item("a"))     # ['a']
print(add_item("b"))     # ['a', 'b']  ← surprise
print(add_item("c"))     # ['a', 'b', 'c']

# FIXED — use a sentinel, then create a fresh list inside
def add_item(item, target=None):
    if target is None:
        target = []
    target.append(item)
    return target

print(add_item("a"))     # ['a']
print(add_item("b"))     # ['b']

Why: default argument values are evaluated once at function definition time and stored on the function object (add_item.__defaults__). All calls without an explicit argument share that same list. Using None as a sentinel defers list creation to call time.

XVI. References, Key Takeaways & CO Mapping

16.1 Textbooks and References

16.2 Relevant Web Resources

16.3 Key Takeaways

1. Graphs: BFS for unweighted shortest paths, DFS for connectivity/cycles, Dijkstra for weighted, topological sort for dependency order, Union–Find for dynamic connectivity.
2. String algorithms: KMP guarantees linear time, Rabin–Karp uses rolling hashes, tries enable prefix search, and Aho–Corasick handles many patterns at once.
3. Large-scale data: Polars for single-machine speed, Dask for pandas-like parallel workloads, PySpark for cluster-scale. Lazy evaluation is the key optimisation mechanism.
4. Deep learning: PyTorch's training loop has six steps: zero_grad → forward → loss → backward → step → log. Always call model.eval() for inference.
5. LLM APIs: structured output with Pydantic, streaming for UX, RAG for grounding, and function calling for tool use.
6. API paradigms: REST for public, GraphQL for flexible clients, gRPC for high-performance internal services. Never universal superiority — always trade-offs.
7. Property-based testing with Hypothesis finds edge cases that example tests miss. Mutation testing verifies test quality.
8. System design is a structured conversation: clarify, estimate, design, deep-dive, discuss bottlenecks.
9. Advanced patterns (CQRS, event sourcing, saga, outbox) solve real problems but add complexity — use them only when the problem demands it.
10. Cloud deployment spans IaaS (EC2), CaaS (ECS, Cloud Run) and FaaS (Lambda). Match the compute model to the workload.
11. Interview preparation is a muscle: 100+ DSA problems, mock interviews, and clean communication beat raw talent alone.
12. Communication wins interviews. Narrate your thinking, clarify requirements, and state complexity explicitly.

16.4 CO Mapping

16.5 Recommended Learning Path

16.6 Course-Wide Integration Map