Python Backend Interview Handbook

# ❌ Mutable default argument — the #1 Python gotcha
def append_item(item, lst=[]):  # lst is created ONCE at definition time
    lst.append(item)
    return lst
 
append_item(1)  # [1]
append_item(2)  # [1, 2] — bug! shared across all calls
 
# ✅ Correct pattern
def append_item(item, lst=None):
    if lst is None:
        lst = []
    lst.append(item)
    return lst

# is vs ==
a = [1, 2, 3]
b = a           # same object
c = [1, 2, 3]   # equal but different object
 
a == c   # True  — equality
a is c   # False — identity
a is b   # True  — same object

a = 256; b = 256; a is b  # True  (interned)
a = 257; b = 257; a is b  # False (not interned) — watch out!

class Point:
    __slots__ = ['x', 'y']  # No __dict__ created per instance
    def __init__(self, x, y):
        self.x = x
        self.y = y
 
# 3–5x less memory than a regular class
# Use case: ORM row objects, graph nodes, DTOs at scale (1M+ instances)

from collections import deque, defaultdict, Counter
import heapq
 
# deque — O(1) at both ends (use instead of list for queues)
q = deque([1, 2, 3])
q.appendleft(0)  # O(1) — list.insert(0, x) is O(n)!
q.popleft()      # O(1)
 
# Counter — frequency counting
c = Counter(['apple', 'banana', 'apple'])
c.most_common(2)  # [('apple', 2), ('banana', 1)]
 
# defaultdict — clean grouping, no KeyError
graph = defaultdict(list)
graph['a'].append('b')
 
# heapq — min-heap (negate values for max-heap)
heap = []
heapq.heappush(heap, 3)
heapq.heappush(heap, 1)
heapq.heappop(heap)  # 1 (min)

# Use set not list for membership tests
VALID_STATUSES = {'active', 'inactive', 'pending'}  # O(1)
if user.status in VALID_STATUSES: ...               # vs list: O(n)
 
# LRU Cache using OrderedDict
from collections import OrderedDict
 
class LRUCache:
    def __init__(self, capacity: int):
        self.cache = OrderedDict()
        self.capacity = capacity
 
    def get(self, key: int) -> int:
        if key not in self.cache:
            return -1
        self.cache.move_to_end(key)  # Mark as recently used
        return self.cache[key]
 
    def put(self, key: int, value: int) -> None:
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)  # Evict LRU

# List comprehension — creates full list in memory
squares = [x**2 for x in range(10) if x % 2 == 0]
 
# Dict comprehension — transform dicts (common in APIs)
user_map = {user.id: user for user in users}
 
# Generator expression — lazy, constant memory
total = sum(x**2 for x in range(1_000_000))  # No list created!
 
# Rule: generator when iterating once over large data
#       list when you need random access, len(), or multiple passes

# Generator function — for streaming/pipeline patterns
def read_large_file(filepath):
    with open(filepath) as f:
        for line in f:
            yield line.strip()  # O(1) memory regardless of file size
 
# Process line by line — never loads full file
for line in read_large_file('large_log.txt'):
    process(line)

# Generator pipelines — Unix pipe style
def parse_logs(lines):
    for line in lines:
        yield json.loads(line)
 
def filter_errors(records):
    for record in records:
        if record['level'] == 'ERROR':
            yield record
 
# Compose — processes one item at a time, O(1) memory
lines   = read_large_file('app.log')
records = parse_logs(lines)
errors  = filter_errors(records)
 
# yield from — delegate to sub-generator
def flatten(nested):
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)
        else:
            yield item

import functools
import time
 
# Decorator with arguments — three levels of nesting
def retry(max_attempts=3, exceptions=(Exception,)):
    def decorator(func):
        @functools.wraps(func)  # Preserves __name__, __doc__
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(2 ** attempt)  # Exponential backoff
        return wrapper
    return decorator
 
@retry(max_attempts=3, exceptions=(ConnectionError, TimeoutError))
def call_external_api(url):
    return requests.get(url)

# Closure — function that captures outer scope variables
def make_counter(start=0):
    count = start  # closed-over variable
 
    def increment():
        nonlocal count  # must declare to modify outer var
        count += 1
        return count
 
    return increment
 
counter = make_counter(10)
counter()  # 11
counter()  # 12
 
# Classic closure bug in loops
funcs = [lambda: i for i in range(3)]
[f() for f in funcs]   # [2, 2, 2] — all capture same i!
 
# Fix: capture by value via default argument
funcs = [lambda i=i: i for i in range(3)]
[f() for f in funcs]   # [0, 1, 2] ✅

# Class-based decorator — for stateful decorators
class RateLimiter:
    def __init__(self, max_calls, period):
        self.max_calls = max_calls
        self.period = period
        self.calls = []
 
    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            self.calls = [c for c in self.calls if now - c < self.period]
            if len(self.calls) >= self.max_calls:
                raise Exception("Rate limit exceeded")
            self.calls.append(now)
            return func(*args, **kwargs)
        return wrapper
 
@RateLimiter(max_calls=100, period=60)
def api_endpoint():
    pass

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
 
# Abstract Base Class — enforce interface contracts
class Repository(ABC):
    @abstractmethod
    def get(self, id: int): pass
 
    @abstractmethod
    def save(self, entity): pass
 
# Dataclass — use instead of plain dicts in APIs
@dataclass
class UserDTO:
    id: int
    name: str
    email: str
    roles: list[str] = field(default_factory=list)
    active: bool = True

# Strategy pattern — swap algorithms at runtime
class PricingStrategy(ABC):
    @abstractmethod
    def calculate(self, amount: float) -> float: ...
 
class RegularPricing(PricingStrategy):
    def calculate(self, amount): return amount
 
class DiscountPricing(PricingStrategy):
    def __init__(self, discount): self.discount = discount
    def calculate(self, amount): return amount * (1 - self.discount)
 
# Observer pattern — event-driven systems
class EventBus:
    def __init__(self):
        self._subscribers = defaultdict(list)
 
    def subscribe(self, event_type: str, handler):
        self._subscribers[event_type].append(handler)
 
    def publish(self, event_type: str, data):
        for handler in self._subscribers[event_type]:
            handler(data)
 
bus = EventBus()
bus.subscribe('user.created', send_welcome_email)
bus.subscribe('user.created', create_default_settings)
bus.publish('user.created', {'id': 1, 'email': 'a@b.com'})

# Repository pattern — decouple data access from business logic
class UserRepository(ABC):
    @abstractmethod
    async def get_by_id(self, user_id: int) -> Optional[User]: pass
 
class SQLUserRepository(UserRepository):
    def __init__(self, db: AsyncSession):
        self.db = db
 
    async def get_by_id(self, user_id: int) -> Optional[User]:
        result = await self.db.execute(select(User).where(User.id == user_id))
        return result.scalar_one_or_none()
 
class MockUserRepository(UserRepository):
    def __init__(self): self._users = {}
 
    async def get_by_id(self, user_id: int) -> Optional[User]:
        return self._users.get(user_id)
 
# In tests, inject MockUserRepository — no DB needed

# Custom exceptions — always define for your domain
class AppError(Exception):
    """Base exception"""
    pass
 
class NotFoundError(AppError):
    def __init__(self, resource: str, id):
        self.resource = resource
        self.id = id
        super().__init__(f"{resource} with id={id} not found")
 
class ValidationError(AppError):
    def __init__(self, field: str, message: str):
        self.field = field
        super().__init__(f"Validation failed on {field}: {message}")
 
# Context manager for resource management
from contextlib import contextmanager
 
@contextmanager
def transaction(db):
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise

# Exception chaining — preserve original cause
try:
    result = external_api.call()
except requests.RequestException as e:
    raise ServiceUnavailableError("Payment service down") from e
    # Original traceback preserved in __cause__

from typing import Optional, Union, TypeVar
 
T = TypeVar('T')
 
def get_user(user_id: int) -> Optional[User]: ...
 
def first(items: list[T]) -> Optional[T]:
    return items[0] if items else None

from typing import Protocol, TypedDict, Literal
 
# Protocol — structural subtyping (duck typing + type safety)
class Saveable(Protocol):
    def save(self) -> None: ...
 
# TypedDict — typed dicts for JSON structures
class UserResponse(TypedDict):
    id: int
    name: str
    email: str
 
# Literal — constrain to specific values
Status = Literal['active', 'inactive', 'banned']
def update_status(user_id: int, status: Status) -> None: ...

# Pattern 1: Two Sum — hashmap for O(n) complement lookup
def two_sum(nums: list[int], target: int) -> list[int]:
    seen = {}  # value -> index
    for i, num in enumerate(nums):
        complement = target - num
        if complement in seen:
            return [seen[complement], i]
        seen[num] = i
    return []
# Time: O(n)  Space: O(n)

# Pattern 2: Frequency map — top-K, duplicates, anagrams
from collections import Counter
 
def top_k_frequent(nums: list[int], k: int) -> list[int]:
    return [x for x, _ in Counter(nums).most_common(k)]
 
# Pattern 3: Group Anagrams — canonical key for grouping
def group_anagrams(strs: list[str]) -> list[list[str]]:
    groups = defaultdict(list)
    for s in strs:
        key = tuple(sorted(s))  # Canonical form
        groups[key].append(s)
    return list(groups.values())

# Fixed-size window — max sum of k elements
def max_sum_k(nums: list[int], k: int) -> int:
    window_sum = sum(nums[:k])
    max_sum = window_sum
 
    for i in range(k, len(nums)):
        window_sum += nums[i] - nums[i - k]  # Add new, remove old
        max_sum = max(max_sum, window_sum)
 
    return max_sum

# Variable-size window — longest substring without repeating chars
def length_of_longest_substring(s: str) -> int:
    seen = {}   # char -> last index
    left = 0
    max_len = 0
 
    for right, char in enumerate(s):
        if char in seen and seen[char] >= left:
            left = seen[char] + 1  # Shrink window
        seen[char] = right
        max_len = max(max_len, right - left + 1)
 
    return max_len
 
# Template: expand right → update state → shrink left while invalid → update result

# Sliding window rate limiter — DIRECT production usage
from collections import deque
import time
 
class SlidingWindowRateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests: dict[str, deque] = defaultdict(deque)
 
    def is_allowed(self, user_id: str) -> bool:
        now = time.time()
        q = self.requests[user_id]
        while q and q[0] < now - self.window:
            q.popleft()
        if len(q) < self.max_requests:
            q.append(now)
            return True
        return False

# Valid parentheses
def is_valid(s: str) -> bool:
    stack = []
    pairs = {')': '(', '}': '{', ']': '['}
    for char in s:
        if char in '({[':
            stack.append(char)
        elif not stack or stack[-1] != pairs[char]:
            return False
        else:
            stack.pop()
    return len(stack) == 0
 
# Monotonic stack — next greater element
def next_greater_element(nums: list[int]) -> list[int]:
    result = [-1] * len(nums)
    stack = []  # Indices waiting for their next greater
 
    for i, num in enumerate(nums):
        while stack and nums[stack[-1]] < num:
            result[stack.pop()] = num
        stack.append(i)
 
    return result

from collections import deque
 
# BFS template
def bfs(graph: dict, start: str) -> list[str]:
    visited = {start}
    queue = deque([start])
    order = []
 
    while queue:
        node = queue.popleft()
        order.append(node)
        for neighbor in graph.get(node, []):
            if neighbor not in visited:
                visited.add(neighbor)
                queue.append(neighbor)
 
    return order
 
# Level-by-level BFS — when depth matters
def bfs_levels(graph: dict, start: str) -> list[list[str]]:
    visited = {start}
    queue = deque([start])
    levels = []
 
    while queue:
        level = []
        for _ in range(len(queue)):  # Process one level at a time
            node = queue.popleft()
            level.append(node)
            for neighbor in graph.get(node, []):
                if neighbor not in visited:
                    visited.add(neighbor)
                    queue.append(neighbor)
        levels.append(level)
 
    return levels

import heapq
 
# Top-K largest — min-heap of size k, O(n log k)
def top_k_largest(nums: list[int], k: int) -> list[int]:
    heap = []
    for num in nums:
        heapq.heappush(heap, num)
        if len(heap) > k:
            heapq.heappop(heap)  # Remove smallest
    return sorted(heap, reverse=True)
 
# K-way merge — merge K sorted lists
def merge_k_sorted(lists: list[list[int]]) -> list[int]:
    result = []
    heap = [(lst[0], i, 0) for i, lst in enumerate(lists) if lst]
    heapq.heapify(heap)
 
    while heap:
        val, li, ei = heapq.heappop(heap)
        result.append(val)
        if ei + 1 < len(lists[li]):
            heapq.heappush(heap, (lists[li][ei + 1], li, ei + 1))
 
    return result
# Backend usage: merge paginated results from multiple DB shards
 
# Median of stream
class MedianFinder:
    def __init__(self):
        self.lo = []  # max-heap (negate) for lower half
        self.hi = []  # min-heap for upper half
 
    def add_num(self, num: int):
        heapq.heappush(self.lo, -num)
        heapq.heappush(self.hi, -heapq.heappop(self.lo))
        if len(self.hi) > len(self.lo):
            heapq.heappush(self.lo, -heapq.heappop(self.hi))
 
    def find_median(self) -> float:
        if len(self.lo) > len(self.hi):
            return -self.lo[0]
        return (-self.lo[0] + self.hi[0]) / 2

# Standard binary search
def binary_search(nums: list[int], target: int) -> int:
    left, right = 0, len(nums) - 1
    while left <= right:
        mid = left + (right - left) // 2
        if nums[mid] == target: return mid
        elif nums[mid] < target: left = mid + 1
        else: right = mid - 1
    return -1
 
# Binary search on answer — "minimize X such that condition holds"
def minimize(feasible_fn, lo: int, hi: int) -> int:
    while lo < hi:
        mid = (lo + hi) // 2
        if feasible_fn(mid):
            hi = mid        # mid works, try smaller
        else:
            lo = mid + 1    # mid fails, try larger
    return lo
 
# Find first position (left boundary)
def find_first(nums: list[int], target: int) -> int:
    left, right = 0, len(nums)
    result = -1
    while left < right:
        mid = (left + right) // 2
        if nums[mid] == target:
            result = mid
            right = mid     # Keep searching left
        elif nums[mid] < target:
            left = mid + 1
        else:
            right = mid
    return result

# DFS — iterative (avoids stack overflow on deep graphs)
def dfs_iterative(graph, start):
    visited = set(); stack = [start]; result = []
    while stack:
        node = stack.pop()
        if node in visited: continue
        visited.add(node); result.append(node)
        stack.extend(reversed(graph.get(node, [])))
    return result
 
# Backtracking template — combinations/subsets
def combination_sum(candidates: list[int], target: int) -> list[list[int]]:
    result = []
 
    def dfs(start: int, current: list[int], remaining: int):
        if remaining == 0:
            result.append(current[:]); return
        if remaining < 0:
            return  # Prune
 
        for i in range(start, len(candidates)):
            current.append(candidates[i])
            dfs(i + 1, current, remaining - candidates[i])
            current.pop()  # Undo choice
 
    dfs(0, [], target)
    return result
# Pattern: choose → explore → unchoose

# Build: O(n)  Query: O(1)
def build_prefix(nums: list[int]) -> list[int]:
    prefix = [0] * (len(nums) + 1)
    for i, num in enumerate(nums):
        prefix[i + 1] = prefix[i] + num
    return prefix
 
def range_sum(prefix: list[int], left: int, right: int) -> int:
    return prefix[right + 1] - prefix[left]
 
# Subarray sum equals K — prefix + hashmap
def subarray_sum(nums: list[int], k: int) -> int:
    count = 0; prefix = 0
    seen = {0: 1}  # prefix_sum -> frequency
 
    for num in nums:
        prefix += num
        count += seen.get(prefix - k, 0)
        seen[prefix] = seen.get(prefix, 0) + 1
 
    return count

# Bottom-up DP template
def dp_template(n: int) -> int:
    dp = [0] * (n + 1)
    dp[0] = base_case
    for i in range(1, n + 1):
        dp[i] = transition(dp, i)
    return dp[n]
 
# 0/1 Knapsack — resource allocation
def knapsack(weights: list[int], values: list[int], capacity: int) -> int:
    n = len(weights)
    dp = [[0] * (capacity + 1) for _ in range(n + 1)]
    for i in range(1, n + 1):
        for w in range(capacity + 1):
            dp[i][w] = dp[i-1][w]  # Don't take item
            if weights[i-1] <= w:
                dp[i][w] = max(dp[i][w], dp[i-1][w - weights[i-1]] + values[i-1])
    return dp[n][capacity]
# Backend usage: feature flag allocation, resource scheduling

# Detect cycle in directed graph (3-color DFS)
def has_cycle(graph: dict) -> bool:
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {node: WHITE for node in graph}
 
    def dfs(node) -> bool:
        color[node] = GRAY
        for neighbor in graph.get(node, []):
            if color[neighbor] == GRAY: return True  # Back edge = cycle
            if color[neighbor] == WHITE and dfs(neighbor): return True
        color[node] = BLACK
        return False
 
    return any(dfs(n) for n in graph if color[n] == WHITE)
 
# Dijkstra's — weighted shortest path
def dijkstra(graph: dict, start: str) -> dict[str, float]:
    distances = {node: float('inf') for node in graph}
    distances[start] = 0
    heap = [(0, start)]
 
    while heap:
        dist, node = heapq.heappop(heap)
        if dist > distances[node]: continue
        for neighbor, weight in graph.get(node, []):
            new_dist = dist + weight
            if new_dist < distances[neighbor]:
                distances[neighbor] = new_dist
                heapq.heappush(heap, (new_dist, neighbor))
 
    return distances
 
# Union-Find — connected components
class UnionFind:
    def __init__(self, n: int):
        self.parent = list(range(n))
        self.rank = [0] * n
 
    def find(self, x: int) -> int:
        if self.parent[x] != x:
            self.parent[x] = self.find(self.parent[x])  # Path compression
        return self.parent[x]
 
    def union(self, x: int, y: int) -> bool:
        px, py = self.find(x), self.find(y)
        if px == py: return False
        if self.rank[px] < self.rank[py]: px, py = py, px
        self.parent[py] = px
        if self.rank[px] == self.rank[py]: self.rank[px] += 1
        return True

import asyncio
 
# Sequential — 0.2s total
async def sequential():
    a = await fetch_user(1)  # suspend, wait 0.1s
    b = await fetch_user(2)  # suspend, wait 0.1s
    return a, b
 
# Concurrent — ~0.1s total
async def concurrent():
    a, b = await asyncio.gather(fetch_user(1), fetch_user(2))
    return a, b
 
# With error handling
results = await asyncio.gather(*tasks, return_exceptions=True)
successes = [r for r in results if not isinstance(r, Exception)]

# Semaphore — limit concurrency (protect downstream services)
async def fetch_all(urls: list[str]) -> list[dict]:
    semaphore = asyncio.Semaphore(10)  # Max 10 concurrent
 
    async def fetch_one(url: str) -> dict:
        async with semaphore:
            return await fetch(url)
 
    return await asyncio.gather(*[fetch_one(u) for u in urls])
 
# Timeout
result = await asyncio.wait_for(fetch(url), timeout=5.0)
 
# Task group (Python 3.11+)
async with asyncio.TaskGroup() as tg:
    task1 = tg.create_task(fetch_user(1))
    task2 = tg.create_task(fetch_user(2))
# Both done here, any exception propagates

# CPU-bound in async context — WRONG vs CORRECT
# ❌ WRONG — blocks event loop
async def compute(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()
 
# ✅ CORRECT — offload to thread pool
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)
 
async def compute(data: bytes) -> str:
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        executor, lambda: hashlib.sha256(data).hexdigest()
    )

from fastapi import FastAPI, Depends, HTTPException, status, Query, Path
from pydantic import BaseModel, EmailStr, validator
 
app = FastAPI(title="User API", version="1.0.0")
 
class UserCreate(BaseModel):
    name: str
    email: EmailStr
    age: int
 
    @validator('age')
    def age_valid(cls, v):
        if not 0 <= v <= 150:
            raise ValueError('age must be 0–150')
        return v
 
class UserResponse(BaseModel):
    id: int; name: str; email: str
    class Config: from_attributes = True
 
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(payload: UserCreate, db: AsyncSession = Depends(get_db)):
    user = User(**payload.dict())
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user
 
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int = Path(..., gt=0),
    db: AsyncSession = Depends(get_db)
):
    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# Dependency injection — the heart of FastAPI
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
 
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> User:
    payload = verify_token(token)  # raises if invalid
    user = await db.get(User, payload['sub'])
    if not user:
        raise HTTPException(401, "User not found")
    return user
 
# Chain dependencies
async def get_active_user(user: User = Depends(get_current_user)) -> User:
    if not user.is_active:
        raise HTTPException(403, "Inactive user")
    return user
 
# Middleware — cross-cutting concerns
@app.middleware("http")
async def request_id_middleware(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response
 
# Background tasks — fire and forget after response
@app.post("/users")
async def create_user(payload: UserCreate, background_tasks: BackgroundTasks, ...):
    user = await create_user_in_db(db, payload)
    background_tasks.add_task(send_welcome_email, user.email)  # After response
    background_tasks.add_task(notify_analytics, user.id)
    return user

import jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
 
pwd_context = CryptContext(schemes=["bcrypt"])
SECRET_KEY = "your-256-bit-secret"  # From env in production
 
def create_access_token(user_id: int, roles: list[str]) -> str:
    payload = {
        "sub": str(user_id),
        "roles": roles,
        "exp": datetime.utcnow() + timedelta(minutes=30),
        "iat": datetime.utcnow(),
        "type": "access"
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")
 
def verify_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        if payload.get("type") != "access":
            raise InvalidTokenError("Wrong token type")
        return payload
    except jwt.ExpiredSignatureError:
        raise TokenExpiredError("Token has expired")
    except jwt.InvalidTokenError as e:
        raise InvalidTokenError(f"Invalid token: {e}")
 
# Password hashing — bcrypt is intentionally slow (~250ms at cost=12)
def hash_password(password: str) -> str:
    return pwd_context.hash(password)
 
def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

# Refresh token rotation — production pattern
async def refresh_tokens(refresh_token: str, db) -> tuple[str, str]:
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=["HS256"])
    except jwt.InvalidTokenError:
        raise InvalidTokenError("Invalid refresh token")
 
    # Check not revoked
    record = await db.execute(
        select(RefreshToken).where(
            RefreshToken.token == refresh_token,
            RefreshToken.revoked == False
        )
    )
    if not record.scalar_one_or_none():
        raise InvalidTokenError("Token already used or revoked")
 
    # Revoke old, issue new (rotation)
    await db.execute(
        update(RefreshToken)
        .where(RefreshToken.token == refresh_token)
        .values(revoked=True)
    )
    new_access = create_access_token(int(payload["sub"]), payload["roles"])
    new_refresh = create_refresh_token(int(payload["sub"]))
    await db.commit()
    return new_access, new_refresh

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
 
class Base(DeclarativeBase): pass
 
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    name: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    orders: Mapped[list["Order"]] = relationship(back_populates="user")
 
# The N+1 Problem — most common DB performance bug
# ❌ WRONG: N+1 queries
users = (await db.execute(select(User))).scalars().all()
for user in users:
    orders = await db.execute(select(Order).where(Order.user_id == user.id))  # N extra!
 
# ✅ CORRECT: eager loading
from sqlalchemy.orm import selectinload, joinedload
 
users = await db.execute(
    select(User).options(selectinload(User.orders))  # One extra query, not N
)
# joinedload — for one-to-one/few (JOIN instead of separate query)

# Cursor pagination — O(1) seek vs offset O(n) scan
async def paginate_cursor(db, cursor: Optional[int] = None, limit: int = 20):
    query = select(User).order_by(User.id).limit(limit + 1)
    if cursor:
        query = query.where(User.id > cursor)
    users = (await db.execute(query)).scalars().all()
    has_next = len(users) > limit
    return users[:limit], users[-1].id if has_next else None
 
# Optimistic locking — handle concurrent updates
async def transfer(db, from_id: int, to_id: int, amount: int):
    result = await db.execute(select(Account).where(Account.id == from_id))
    account = result.scalar_one()
 
    updated = await db.execute(
        update(Account)
        .where(Account.id == from_id, Account.version == account.version)
        .values(balance=Account.balance - amount, version=Account.version + 1)
    )
    if updated.rowcount == 0:
        raise ConcurrentModificationError("Account modified concurrently")

# Cache-aside pattern — standard approach
async def get_user_cached(user_id: int) -> Optional[dict]:
    cache_key = f"user:{user_id}"
    cached = await r.get(cache_key)
    if cached:
        return json.loads(cached)
 
    user = await db.get(User, user_id)
    if user:
        await r.set(cache_key, json.dumps(user.to_dict()), ex=300)
    return user.to_dict() if user else None
 
# Distributed lock with Lua (atomic release)
@asynccontextmanager
async def redis_lock(r, lock_name: str, timeout: int = 10):
    lock_value = str(uuid.uuid4())
    lock_key = f"lock:{lock_name}"
    acquired = await r.set(lock_key, lock_value, nx=True, ex=timeout)
    if not acquired:
        raise LockNotAcquiredError(lock_name)
    try:
        yield
    finally:
        release_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        end return 0
        """
        await r.eval(release_script, 1, lock_key, lock_value)

from celery import Celery
 
celery = Celery(
    "myapp",
    broker="redis://localhost:6379/0",   # Task queue
    backend="redis://localhost:6379/1",  # Result store
)
 
celery.conf.update(
    task_soft_time_limit=300,       # 5 min — raises SoftTimeLimitExceeded
    task_time_limit=360,            # 6 min — kills worker
    worker_max_tasks_per_child=1000 # Restart to prevent memory leaks
)
 
@celery.task(bind=True, max_retries=3)
def send_email(self, user_id: int, template: str):
    try:
        user = get_user(user_id)
        email_service.send(user.email, template)
    except EmailServiceError as e:
        raise self.retry(exc=e, countdown=2 ** self.request.retries)
 
# Call tasks
send_email.delay(user_id=1, template="welcome")            # Async now
send_email.apply_async(args=[1, "welcome"], countdown=60)  # Delayed

# Structured JSON logging — always use in production
import logging, json
from contextvars import ContextVar
 
request_id_var: ContextVar[str] = ContextVar('request_id', default='')
 
class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        return json.dumps({
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "request_id": request_id_var.get(),
            "module": record.module,
            "line": record.lineno,
            **({"exception": self.formatException(record.exc_info)}
               if record.exc_info else {})
        })
 
# Health check endpoint — required for load balancers, k8s
@app.get("/health")
async def health_check():
    checks = {}
    status_code = 200
    try:
        await db.execute(text("SELECT 1")); checks["database"] = "ok"
    except Exception as e:
        checks["database"] = f"error: {e}"; status_code = 503
    try:
        await redis.ping(); checks["redis"] = "ok"
    except Exception as e:
        checks["redis"] = f"error: {e}"; status_code = 503
    return JSONResponse(
        content={"status": "ok" if status_code == 200 else "degraded", "checks": checks},
        status_code=status_code
    )
# /health/live  — is process running? (don't restart if DB is down)
# /health/ready — ready to serve traffic? (remove from LB if degraded)

import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
 
@pytest.fixture
async def db():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with async_sessionmaker(engine)() as session:
        yield session
    await engine.dispose()
 
@pytest.fixture
async def client(db):
    app.dependency_overrides[get_db] = lambda: db
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()
 
@pytest.mark.asyncio
async def test_create_user(client):
    response = await client.post("/users", json={"name": "Alice", "email": "a@b.com", "age": 25})
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Alice"
    assert "id" in data
 
# Mocking external services
from unittest.mock import AsyncMock, patch
 
@pytest.mark.asyncio
async def test_email_sent_on_register(client):
    with patch('myapp.services.email.send_email', new_callable=AsyncMock) as mock_email:
        await client.post("/users", json={"name": "Alice", "email": "a@b.com"})
        mock_email.assert_called_once_with("a@b.com", "welcome")

-- EXPLAIN ANALYZE — your debugging superpower
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT u.id, u.name, COUNT(o.id) as order_count
FROM users u
LEFT JOIN orders o ON o.user_id = u.id
WHERE u.created_at > NOW() - INTERVAL '30 days'
GROUP BY u.id, u.name
ORDER BY order_count DESC LIMIT 20;
 
-- What to look for:
-- ❌ Seq Scan on large table → add index
-- ❌ High rows estimate mismatch → run ANALYZE (stale statistics)
-- ✅ Index Scan / Index Only Scan
 
-- Indexing strategy
CREATE INDEX CONCURRENTLY idx_users_email ON users(email);
 
-- Composite — equality conditions FIRST
CREATE INDEX idx_orders_user_status ON orders(user_id, status);
-- Useful for: WHERE user_id = ? AND status = ?
-- NOT useful for: WHERE status = ? (no user_id prefix)
 
-- Partial — only index a subset (smaller, faster)
CREATE INDEX idx_active_users ON users(email) WHERE is_active = true;
 
-- Covering — include extra cols to avoid heap fetch
CREATE INDEX idx_orders_covering ON orders(user_id)
  INCLUDE (id, status, created_at);

-- SELECT FOR UPDATE — pessimistic locking for bank transfers
BEGIN;
SELECT * FROM accounts WHERE id = 1 FOR UPDATE;  -- Lock this row
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
COMMIT;
 
-- SKIP LOCKED — queue processing pattern (no blocking)
SELECT * FROM jobs WHERE status = 'pending'
LIMIT 1 FOR UPDATE SKIP LOCKED;
 
-- Window functions — analytics without multiple queries
SELECT user_id, amount,
    SUM(amount) OVER (PARTITION BY user_id ORDER BY order_date) AS running_total,
    ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY order_date DESC) AS recency_rank
FROM orders;
 
-- JSONB for flexible schemas
CREATE TABLE events (
    id SERIAL PRIMARY KEY,
    type VARCHAR(50),
    payload JSONB,
    created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_events_payload ON events USING gin(payload);
SELECT * FROM events WHERE payload @> '{"action": "login"}';

# Production Dockerfile — multi-stage build
FROM python:3.12-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
FROM python:3.12-slim AS runtime
WORKDIR /app
RUN useradd -m -u 1000 appuser  # Non-root for security
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY --chown=appuser:appuser . .
USER appuser
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# docker-compose.yml — local dev
version: '3.8'
services:
  api:
    build: .
    ports: ["8000:8000"]
    environment:
      - DATABASE_URL=postgresql+asyncpg://postgres:password@db:5432/mydb
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      db: { condition: service_healthy }
 
  db:
    image: postgres:15
    environment: { POSTGRES_DB: mydb, POSTGRES_USER: postgres, POSTGRES_PASSWORD: password }
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s; retries: 5
 
  worker:
    build: .
    command: celery -A celery_app worker --loglevel=info
    depends_on: [redis, db]
 
  redis:
    image: redis:7-alpine

# Best practices:
# 1. Never delete a column in the same migration as removing code
# 2. Always test downgrade() works
# 3. Use CONCURRENTLY for index creation in production
# 4. Data migrations go in separate scripts, not schema migrations
 
# Zero-downtime deployment order:
# Phase 1: Deploy migration (add nullable column)
# Phase 2: Deploy app code (reads new column, writes both)
# Phase 3: Backfill data (background job)
# Phase 4: Add NOT NULL constraint, remove old column
 
def upgrade():
    # Step 1: Add nullable (instant, no lock)
    op.add_column('users', sa.Column('phone', sa.String(20), nullable=True))
 
def downgrade():
    op.drop_column('users', 'phone')

# .github/workflows/ci.yml
name: CI
on: [push, pull_request]
 
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:15
        env: { POSTGRES_DB: testdb, POSTGRES_USER: postgres, POSTGRES_PASSWORD: password }
        options: --health-cmd pg_isready --health-interval 10s --health-retries 5
      redis:
        image: redis:7
        options: --health-cmd "redis-cli ping" --health-interval 10s
 
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.12', cache: 'pip' }
      - run: pip install -r requirements.txt
      - name: Run migrations
        run: alembic upgrade head
        env: { DATABASE_URL: postgresql+asyncpg://postgres:password@localhost:5432/testdb }
      - name: Run tests
        run: pytest tests/ -v --cov=myapp --cov-report=xml
      - name: Lint
        run: ruff check . && mypy myapp/

from pydantic_settings import BaseSettings
from functools import lru_cache
 
class Settings(BaseSettings):
    DATABASE_URL: str
    JWT_SECRET_KEY: str
    REDIS_URL: str = "redis://localhost:6379/0"
    ENV: str = "development"
    DEBUG: bool = False
    LOG_LEVEL: str = "INFO"
 
    class Config:
        env_file = ".env"
 
@lru_cache()  # Read env once — singleton pattern
def get_settings() -> Settings:
    return Settings()
 
settings = get_settings()

class ShortenedURL(Base):
    __tablename__ = "shortened_urls"
    id = Column(Integer, primary_key=True)
    original_url = Column(String(2048), nullable=False)
    short_code = Column(String(10), unique=True, index=True)
    click_count = Column(Integer, default=0)
    expires_at = Column(DateTime, nullable=True)
    created_at = Column(DateTime, server_default=func.now())
    is_active = Column(Boolean, default=True)

ALPHABET = string.ascii_letters + string.digits  # 62 chars
 
# Base62 encode DB ID — guaranteed unique, no collision
def id_to_code(id: int) -> str:
    code = []
    while id > 0:
        code.append(ALPHABET[id % 62])
        id //= 62
    return ''.join(reversed(code)) or ALPHABET[0]
 
class ShortenRequest(BaseModel):
    url: HttpUrl
    custom_alias: Optional[str] = None
    ttl_days: Optional[int] = None
 
@app.post("/shorten", response_model=ShortenResponse)
async def shorten_url(request: ShortenRequest, db=Depends(get_db)):
    url_obj = ShortenedURL(
        original_url=str(request.url),
        expires_at=datetime.utcnow() + timedelta(days=request.ttl_days)
                   if request.ttl_days else None
    )
    db.add(url_obj)
    await db.flush()  # Get auto-increment ID
    url_obj.short_code = request.custom_alias or id_to_code(url_obj.id)
    await db.commit()
    return ShortenResponse(short_url=f"https://short.ly/{url_obj.short_code}")
 
@app.get("/{short_code}")
async def redirect(short_code: str, background_tasks: BackgroundTasks, db=Depends(get_db)):
    # Check Redis cache first
    cached = await redis.get(f"url:{short_code}")
    if cached:
        background_tasks.add_task(increment_click, short_code)
        return RedirectResponse(cached, status_code=301)
 
    url_obj = (await db.execute(
        select(ShortenedURL).where(
            ShortenedURL.short_code == short_code,
            ShortenedURL.is_active == True
        )
    )).scalar_one_or_none()
 
    if not url_obj: raise HTTPException(404, "Not found")
    if url_obj.expires_at and url_obj.expires_at < datetime.utcnow():
        raise HTTPException(410, "URL expired")
 
    await redis.set(f"url:{short_code}", url_obj.original_url, ex=3600)
    background_tasks.add_task(increment_click, short_code)
    return RedirectResponse(url_obj.original_url, status_code=301)

# Sliding Window (Redis ZSET)
async def is_rate_limited(user_id: str, limit: int = 100, window: int = 60) -> bool:
    key = f"rl:{user_id}"
    now = time.time()
    pipe = redis.pipeline()
    await pipe.zremrangebyscore(key, 0, now - window)  # Remove old
    await pipe.zadd(key, {str(now): now})
    await pipe.zcard(key)
    await pipe.expire(key, window)
    results = await pipe.execute()
    return results[2] > limit
 
# Middleware with response headers
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    user_id = getattr(request.state, 'user_id', request.client.host)
    limited = await is_rate_limited(user_id)
    if limited:
        return JSONResponse(
            status_code=429,
            content={"error": "RATE_LIMIT_EXCEEDED"},
            headers={"Retry-After": "60"}
        )
    response = await call_next(request)
    return response

class TaskStatus(str, Enum):
    TODO = "todo"; IN_PROGRESS = "in_progress"
    DONE = "done"; CANCELLED = "cancelled"
 
class TaskPriority(str, Enum):
    LOW = "low"; MEDIUM = "medium"; HIGH = "high"; CRITICAL = "critical"
 
@app.get("/tasks", response_model=list[TaskResponse])
async def list_tasks(
    status: Optional[TaskStatus] = None,
    assignee_id: Optional[int] = None,
    page: int = Query(1, ge=1),
    page_size: int = Query(20, le=100),
    db=Depends(get_db), user=Depends(get_active_user)
):
    filters = []
    if status: filters.append(Task.status == status)
    if assignee_id: filters.append(Task.assignee_id == assignee_id)
    query = select(Task)
    if filters: query = query.where(and_(*filters))
    query = query.offset((page - 1) * page_size).limit(page_size)
    return (await db.execute(query)).scalars().all()
 
@app.patch("/tasks/{task_id}", response_model=TaskResponse)
async def update_task(
    task_id: int, payload: TaskUpdate,
    background_tasks: BackgroundTasks,
    db=Depends(get_db), user=Depends(get_active_user)
):
    task = await db.get(Task, task_id)
    if not task: raise HTTPException(404)
    if task.created_by != user.id and not user.is_admin:
        raise HTTPException(403, "Not authorized")
 
    old_assignee = task.assignee_id
    for field, val in payload.dict(exclude_unset=True).items():
        setattr(task, field, val)
    await db.commit()
 
    if payload.assignee_id and payload.assignee_id != old_assignee:
        background_tasks.add_task(notify_user, payload.assignee_id, task.title)
    return task

# Clean architecture: decouple logic from delivery mechanism
class NotificationChannel(str, Enum):
    EMAIL = "email"; SMS = "sms"; PUSH = "push"
 
class NotificationSender(ABC):
    @abstractmethod
    async def send(self, notification) -> bool: pass
 
class EmailSender(NotificationSender):
    async def send(self, notification) -> bool:
        template = await get_template(notification.template_id)
        rendered = template.render(**notification.context)
        try:
            await sendgrid.send_email(
                to=notification.context['email'],
                subject=rendered.subject,
                body=rendered.body
            )
            return True
        except Exception as e:
            logger.error(f"Email failed: {e}")
            return False
 
class NotificationService:
    def __init__(self):
        self.senders = {
            NotificationChannel.EMAIL: EmailSender(),
            NotificationChannel.SMS: SMSSender(),
        }
 
    async def send_bulk(self, notifications: list) -> dict:
        semaphore = asyncio.Semaphore(50)
 
        async def send_one(n):
            async with semaphore:
                return await self.senders[n.channel].send(n)
 
        results = await asyncio.gather(*[send_one(n) for n in notifications], return_exceptions=True)
        return {"sent": sum(1 for r in results if r is True),
                "failed": sum(1 for r in results if r is not True)}
 
# Celery task — async delivery with retries
@celery.task(bind=True, max_retries=3, default_retry_delay=60)
def deliver_notification(self, notification_dict: dict):
    try:
        success = asyncio.run(NotificationService().send(Notification(**notification_dict)))
        if not success: raise DeliveryError()
        update_status(notification_dict['id'], 'delivered')
    except Exception as e:
        update_status(notification_dict['id'], 'failed')
        raise self.retry(exc=e)

@app.post("/auth/register", response_model=TokenResponse, status_code=201)
async def register(payload: RegisterRequest, db=Depends(get_db)):
    if await get_user_by_email(db, payload.email):
        raise HTTPException(409, "Email already registered")
    user = User(
        email=payload.email,
        name=payload.name,
        password_hash=pwd_context.hash(payload.password)
    )
    db.add(user); await db.commit()
    send_welcome_email.delay(user.id)
    return create_token_response(user)
 
@app.post("/auth/login", response_model=TokenResponse)
async def login(form=Depends(OAuth2PasswordRequestForm), db=Depends(get_db)):
    user = await get_user_by_email(db, form.username)
    # Constant-time comparison — prevents timing attacks
    if not user or not pwd_context.verify(form.password, user.password_hash):
        raise HTTPException(401, "Invalid credentials",
                            headers={"WWW-Authenticate": "Bearer"})
    if not user.is_active:
        raise HTTPException(403, "Account disabled")
    return create_token_response(user)
 
@app.post("/auth/password-reset/request")
async def request_reset(email: EmailStr, db=Depends(get_db)):
    user = await get_user_by_email(db, email)
    if user:  # Always return 200 — prevent email enumeration!
        send_reset_email.delay(user.email, create_reset_token(user.id))
    return {"message": "If an account exists, you'll receive a reset link"}
 
@app.post("/auth/logout")
async def logout(token=Depends(oauth2_scheme), user=Depends(get_current_user)):
    payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    jti = payload.get("jti")
    if jti:
        await redis.set(f"blocklist:{jti}", "1", ex=1800)  # Blocklist until expiry
    return {"message": "Logged out successfully"}