Python Interview Practice

Group orders by customer, returning total spend and order count per customer.

defaultdict with a lambda accumulator. Skips the 'if key not in dict' dance. The standard-library answer to SQL GROUP BY with SUM and COUNT. If you can write this from a blank file in under three minutes, the dictionary section of a Python round is solved.

from collections import defaultdict

def group_orders(orders):
    agg = defaultdict(lambda: {"total": 0, "count": 0})
    for row in orders:
        cid = row["customer_id"]
        agg[cid]["total"] += row["amount"]
        agg[cid]["count"] += 1
    return dict(agg)

Invert a dict; when values collide, collect the original keys into a list.

defaultdict(list) keyed on the value. Append the original key. O(n). The trap is reaching for a dict comprehension and silently losing collisions when two keys had the same value.

from collections import defaultdict

def invert(d):
    out = defaultdict(list)
    for k, v in d.items():
        out[v].append(k)
    return dict(out)

Dedupe a list of dicts by a composite key, keeping the latest record per key.

Two reasonable answers. A dict keyed on the composite tuple, updating when newer ts arrives: O(n) time and memory. Or sort by (key, ts DESC) and take the first occurrence per key: O(n log n), friendlier on streamed input. The dict approach is the Python version of the SQL ROW_NUMBER dedup pattern.

def dedup_latest(records, key_fn, ts_fn):
    latest = {}
    for r in records:
        k = key_fn(r)
        if k not in latest or ts_fn(r) > ts_fn(latest[k]):
            latest[k] = r
    return list(latest.values())

Compute the average of a numeric column in a 10GB CSV without loading the whole file.

Running sum and count, one row at a time. csv.DictReader iterates lazily; pandas.read_csv without chunksize crashes the kernel. The interviewer's follow-up is what changes when the file is gzipped (open via gzip.open in text mode) or distributed across S3 prefixes.

import csv

def avg_column(path, col):
    total, count = 0.0, 0
    with open(path) as f:
        for row in csv.DictReader(f):
            try:
                total += float(row[col]); count += 1
            except (ValueError, TypeError):
                pass
    return total / count if count else None

Flatten a nested JSON into a flat dict with dot-separated keys.

Recursive walk. For each key, if the value is a dict, recurse with prefix + key + '.'. Otherwise, add to result. Decide upfront how to handle lists: explode into multiple rows or serialize. The interviewer always asks; have an answer ready.

def flatten(d, prefix=""):
    out = {}
    for k, v in d.items():
        key = f"{prefix}{k}" if not prefix else f"{prefix}.{k}"
        if isinstance(v, dict):
            out.update(flatten(v, key))
        else:
            out[key] = v
    return out

Sessionize an event stream with a 30-minute inactivity gap.

Sort by (user_id, ts). Walk the list. Start a new session whenever the gap exceeds 30 minutes or the user changes. Assign per-user session counters. The mistake to avoid: comparing against the session start instead of the immediately previous event.

def sessionize(events, gap=1800):
    events.sort(key=lambda e: (e["user_id"], e["ts"]))
    sid, prev_user, prev_ts = 0, None, None
    for e in events:
        if e["user_id"] != prev_user or e["ts"] - prev_ts > gap:
            sid += 1
        e["session_id"] = sid
        prev_user, prev_ts = e["user_id"], e["ts"]
    return events

Generator that reads a large file and yields batches of N lines.

yield from a list that resets every N. Memory stays constant. Pandas has chunksize for the DataFrame version. The interviewer's follow-up is what happens at the last partial batch: yield the partial or skip it. Yield it.

def batches(path, n):
    batch = []
    with open(path) as f:
        for line in f:
            batch.append(line.rstrip("\n"))
            if len(batch) == n:
                yield batch
                batch = []
    if batch:
        yield batch

Decorator that retries a function up to 3 times with exponential backoff and jitter.

Inner function catches a narrow set of exceptions (requests.RequestException, not bare Exception), sleeps 2**attempt seconds plus a random jitter, re-raises after the last attempt. Jitter prevents the self-inflicted DDoS when every worker retries at the same later time. Mention tenacity as the production library.

import time, random
from functools import wraps

def retry(max_attempts=3):
    def deco(fn):
        @wraps(fn)
        def wrapped(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return fn(*args, **kwargs)
                except Exception:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(2 ** attempt + random.uniform(0, 1))
        return wrapped
    return deco

Validate incoming records against a schema; route invalid records to a dead-letter queue.

Define expected fields and types. Iterate records, collect errors per record. Valid records go to output; invalid go to DLQ list with the error attached. The senior signal is returning both lists and tagging the failure mode (missing field, wrong type, out-of-range value) rather than just rejecting.

def validate(records, schema):
    valid, dlq = [], []
    for r in records:
        errs = []
        for field, typ in schema.items():
            if field not in r:
                errs.append(f"missing:{field}")
            elif not isinstance(r[field], typ):
                errs.append(f"type:{field}")
        if errs:
            dlq.append({"record": r, "errors": errs})
        else:
            valid.append(r)
    return valid, dlq

Find the top 10 rows by a numeric column in a 50GB CSV.

heapq.nsmallest or nlargest with a min-heap of size 10. Read line by line. Never call pd.read_csv without chunksize on a 50GB file. The interviewer is filtering for candidates who know that pandas isn't a magic solution and that streaming + small-state is the standard pattern.

import csv, heapq

def top_n(path, col, n=10):
    heap = []
    with open(path) as f:
        for row in csv.DictReader(f):
            try:
                v = float(row[col])
            except (ValueError, TypeError):
                continue
            if len(heap) < n:
                heapq.heappush(heap, (v, row))
            elif v > heap[0][0]:
                heapq.heappushpop(heap, (v, row))
    return [r for _, r in sorted(heap, reverse=True)]

Concurrent fetch with rate limit using asyncio.

asyncio.Semaphore(N) bounds in-flight requests. The interviewer's follow-up is what happens when one request hangs: asyncio.wait_for with a timeout per call. The senior signal is naming why rate limiting matters for downstream services even when your local quota allows more.

Stream-merge sorted iterators into a single sorted iterator.

heapq.merge yields lazily. O(n log k). Beats materialize-then-sort for k iterators. The classic external-sort merge step. Don't reinvent it; name heapq.merge and explain why.

import heapq

def merge_sorted(*iters, key=None):
    return heapq.merge(*iters, key=key)

SCD Type 2 merge in pandas: expire current rows where the source differs and insert new rows.

Identify rows where source differs from target. Expire current (set valid_to, is_current=False), insert new with valid_from=now and is_current=True. Use pd.merge with indicator=True to spot the diff. The interview signal is naming why the surrogate key changes and the natural key stays.

Parse a 2GB Apache access log; output the top 100 URLs by request count, grouped by HTTP status code.

Read line by line. Parse with regex (compile outside the loop). collections.defaultdict(Counter) keyed by status code. After processing, .most_common(100) per status. Mention that on truly large logs the right answer is Spark or DuckDB, not single-machine Python.

Write a property-based test confirming a transformation function never drops rows.

hypothesis.@given with strategies that produce realistic input. Assert len(output) == len(input). The senior signal is volunteering this kind of test for data quality functions where the input space is too large for example-based tests to cover. Most candidates have never used hypothesis; mentioning it lands well.