Coding problems — idiomatic Python under pressure — Python Mastery

These are NOT LeetCode-style algorithms. They're "show me you can write production Python" problems — the kind senior interviewers actually use to evaluate code quality.

For each:

Try the problem yourself in 25–40 min, with type hints + tests.
Compare with the model solution.
Note what idioms / stdlib features you missed.

Use pytest to verify your solutions. Treat every problem as if you're being watched coding live.

Problem 1 — Word frequency from log file (warm-up)

Prompt: Given a path to a text file, return the 10 most common words (case-insensitive, words = [a-zA-Z']+). Should handle a 1 GB file without loading it all into memory.

Time: 15 min.

Model solution

python

from __future__ import annotations
import re
from collections import Counter
from pathlib import Path

WORD = re.compile(r"[a-zA-Z']+")

def top_words(path: Path | str, k: int = 10) -> list[tuple[str, int]]:
    counts: Counter[str] = Counter()
    with open(path, encoding="utf-8", errors="replace") as f:
        for line in f:                       # streaming
            counts.update(w.lower() for w in WORD.findall(line))
    return counts.most_common(k)

What interviewers look for

Streaming (don't f.read()).
Counter not manual dict.
Generator expression in update (no intermediate list).
Type hints + Path | str.
errors="replace" for messy files.
Compiled regex (module-level).

Tests

python

def test_top_words(tmp_path):
    p = tmp_path / "x.txt"
    p.write_text("Hello hello world\nWorld of code", encoding="utf-8")
    assert top_words(p, 2) == [("hello", 2), ("world", 2)]

Problem 2 — Rate limiter (in-memory)

Prompt: Implement a sliding-window rate limiter as a class. Method allow(key) returns True if the key is under the limit (e.g., 10 calls / 60 seconds), else False. Must be thread-safe.

Time: 25 min.

Model solution

python

from collections import deque
from threading import Lock
from time import monotonic

class RateLimiter:
    def __init__(self, *, limit: int, window: float) -> None:
        self.limit = limit
        self.window = window
        self._buckets: dict[str, deque[float]] = {}
        self._lock = Lock()

    def allow(self, key: str) -> bool:
        now = monotonic()
        cutoff = now - self.window
        with self._lock:
            q = self._buckets.setdefault(key, deque())
            while q and q[0] < cutoff:
                q.popleft()
            if len(q) >= self.limit:
                return False
            q.append(now)
            return True

Discussion points

Why monotonic not time.time (immune to clock changes).
Memory leak for keys never seen again — periodic cleanup or TTL dict.
For multi-process: use Redis with ZADD/ZREMRANGEBYSCORE.
Token-bucket vs sliding-window vs leaky-bucket — be ready to discuss.

Stretch: implement as async with asyncio.Lock.

Problem 3 — Retry decorator

Prompt: Write a retry decorator with: times, exceptions=(Exception,), delay, backoff (multiplier), and optional on_failure callback. Works for sync and async functions.

Time: 30 min.

Model solution

python

from __future__ import annotations
import asyncio, functools, inspect, time
from collections.abc import Callable
from typing import Any, ParamSpec, TypeVar, cast

P = ParamSpec("P")
R = TypeVar("R")

def retry(
    *,
    times: int = 3,
    exceptions: tuple[type[BaseException], ...] = (Exception,),
    delay: float = 0.0,
    backoff: float = 2.0,
    on_failure: Callable[[BaseException, int], None] | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    def decorator(fn: Callable[P, R]) -> Callable[P, R]:
        if inspect.iscoroutinefunction(fn):
            @functools.wraps(fn)
            async def aw(*args: P.args, **kwargs: P.kwargs) -> R:
                wait = delay
                for attempt in range(1, times + 1):
                    try:
                        return await cast(Any, fn)(*args, **kwargs)
                    except exceptions as e:
                        if on_failure:
                            on_failure(e, attempt)
                        if attempt == times:
                            raise
                        if wait:
                            await asyncio.sleep(wait)
                            wait *= backoff
                raise AssertionError("unreachable")
            return cast(Callable[P, R], aw)

        @functools.wraps(fn)
        def sw(*args: P.args, **kwargs: P.kwargs) -> R:
            wait = delay
            for attempt in range(1, times + 1):
                try:
                    return fn(*args, **kwargs)
                except exceptions as e:
                    if on_failure:
                        on_failure(e, attempt)
                    if attempt == times:
                        raise
                    if wait:
                        time.sleep(wait)
                        wait *= backoff
            raise AssertionError("unreachable")
        return sw
    return decorator

Look-fors

functools.wraps (preserves metadata).
ParamSpec for typing.
Sync/async handling via iscoroutinefunction.
Re-raise on last attempt.
Configurable backoff.
on_failure hook for logging / metrics.

Note that you should prefer tenacity in real code; this is for the interview.

Problem 4 — LRU cache from scratch

Prompt: Implement an LRUCache class with get(key) and put(key, value), both O(1). Capacity in constructor. (Don't import functools.lru_cache.)

Time: 25 min.

Model solution

python

from collections import OrderedDict
from typing import Generic, TypeVar

K = TypeVar("K"); V = TypeVar("V")

class LRUCache(Generic[K, V]):
    def __init__(self, capacity: int) -> None:
        if capacity <= 0:
            raise ValueError("capacity must be > 0")
        self.capacity = capacity
        self._data: OrderedDict[K, V] = OrderedDict()

    def get(self, key: K) -> V | None:
        if key not in self._data:
            return None
        self._data.move_to_end(key)
        return self._data[key]

    def put(self, key: K, value: V) -> None:
        if key in self._data:
            self._data.move_to_end(key)
        self._data[key] = value
        if len(self._data) > self.capacity:
            self._data.popitem(last=False)

Idiom highlights

Use stdlib OrderedDict (O(1) move_to_end).
Generic over types.
Reject invalid capacity early.

Stretch: implement without OrderedDict (dict + doubly-linked list) — useful exercise even though stdlib does it for you.

Problem 5 — Concurrent URL downloader (async)

Prompt: Given a list of URLs, fetch all and return dict[url, body]. Concurrency must be capped (e.g., 20 at a time). Failed URLs return None. Total timeout per request: 10 s.

Time: 30 min.

Model solution

python

import asyncio
import httpx

async def fetch_all(urls: list[str], *, concurrency: int = 20) -> dict[str, str | None]:
    sem = asyncio.Semaphore(concurrency)

    async def one(client: httpx.AsyncClient, url: str) -> tuple[str, str | None]:
        async with sem:
            try:
                r = await client.get(url, timeout=10)
                r.raise_for_status()
                return url, r.text
            except (httpx.HTTPError, httpx.TimeoutException):
                return url, None

    async with httpx.AsyncClient(http2=True, follow_redirects=True) as client:
        results = await asyncio.gather(*(one(client, u) for u in urls))
    return dict(results)

Senior idioms

Shared AsyncClient (pooling).
Semaphore for backpressure.
Per-request timeout in the get().
Narrow exception list — don't swallow Exception.
gather (or TaskGroup if interviewer wants structured concurrency).

Variation: stream results with asyncio.as_completed to process them ASAP.

Problem 6 — Group-by + aggregate (no Pandas)

Prompt: Given list[dict] of records like {"user_id": int, "amount": float, "currency": str}, return total amount per currency. Ignore records missing keys. ~1M records, do it in pure Python.

Time: 15 min.

Model solution

python

from collections import defaultdict
from typing import Iterable, Mapping, Any

def totals_by_currency(records: Iterable[Mapping[str, Any]]) -> dict[str, float]:
    out: defaultdict[str, float] = defaultdict(float)
    for r in records:
        amt = r.get("amount")
        cur = r.get("currency")
        if amt is None or cur is None:
            continue
        out[cur] += amt
    return dict(out)

What they're checking: defaultdict instead of setdefault boilerplate, defensive .get, narrow type hints, accept any Iterable[Mapping].

Bonus: ask "what if records is a 10 GB JSONL stream?" — answer: iterate file line by line, parse with json.loads, same function works (yay generator-friendly design).

Problem 7 — Top-K with a heap

Prompt: Given a stream of (score: float, item: str) tuples (possibly billions), return the K items with highest scores. Constant memory ≈ K.

Time: 20 min.

Model solution

python

import heapq
from typing import Iterable

def top_k(stream: Iterable[tuple[float, str]], k: int) -> list[tuple[float, str]]:
    heap: list[tuple[float, str]] = []      # min-heap of size k
    for score, item in stream:
        if len(heap) < k:
            heapq.heappush(heap, (score, item))
        elif score > heap[0][0]:
            heapq.heapreplace(heap, (score, item))
    return sorted(heap, reverse=True)

Look-fors: heapq (min-heap by default), use heapreplace (one op) not push+pop, return sorted result.

Problem 8 — Context manager for a temp dir + cleanup

Prompt: Write a context manager working_dir(path) that:

cds into path (creating it if needed).
Restores the original cwd on exit.
Deletes path if delete=True.
Reraises any exception properly.

Time: 20 min.

Model solution

python

import os, shutil
from contextlib import contextmanager
from pathlib import Path
from collections.abc import Iterator

@contextmanager
def working_dir(path: str | Path, *, create: bool = True, delete: bool = False) -> Iterator[Path]:
    path = Path(path)
    original = Path.cwd()
    if create:
        path.mkdir(parents=True, exist_ok=True)
    os.chdir(path)
    try:
        yield path
    finally:
        os.chdir(original)
        if delete:
            shutil.rmtree(path, ignore_errors=True)

Idioms: @contextmanager, try/finally for cleanup (not except — let exceptions propagate), keyword-only options.

Problem 9 — Memoise (with TTL)

Prompt: Write @memoize(ttl=60) decorator. Calls within ttl seconds for the same args return cached. Different args → cached separately.

Time: 20 min.

Model solution

python

import functools, time
from collections.abc import Callable
from typing import Any

def memoize(*, ttl: float):
    def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
        cache: dict[tuple, tuple[float, Any]] = {}
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items())))
            now = time.monotonic()
            cached = cache.get(key)
            if cached and now - cached[0] < ttl:
                return cached[1]
            result = fn(*args, **kwargs)
            cache[key] = (now, result)
            return result
        return wrapper
    return decorator

Discussion: kwargs sorting for stable keys, no thread-safety (add Lock for multi-threaded), no eviction (memory grows).

Problem 10 — Validate + parse a config file

Prompt: Read a config.toml with sections like [server], [database], [features]. Validate using Pydantic and raise readable errors. Support env-var overrides (e.g., DATABASE_URL overrides database.url).

Time: 30 min.

Model solution

python

from pathlib import Path
import tomllib
from pydantic import BaseModel, Field, HttpUrl
from pydantic_settings import BaseSettings, SettingsConfigDict

class ServerCfg(BaseModel):
    host: str = "127.0.0.1"
    port: int = Field(default=8000, ge=1, le=65535)

class DatabaseCfg(BaseModel):
    url: str
    pool_size: int = Field(default=10, ge=1, le=200)

class FeatureFlags(BaseModel):
    new_dashboard: bool = False
    rate_limit: bool = True

class AppCfg(BaseSettings):
    server: ServerCfg = ServerCfg()
    database: DatabaseCfg
    features: FeatureFlags = FeatureFlags()
    model_config = SettingsConfigDict(env_nested_delimiter="__", env_prefix="APP_")

def load(path: str | Path) -> AppCfg:
    data = tomllib.loads(Path(path).read_text(encoding="utf-8"))
    return AppCfg.model_validate(data)

Use:

python

cfg = load("config.toml")
# Override via env: APP_DATABASE__URL=postgresql://...

Quality points: nested models, sensible defaults, env override via pydantic-settings, Field constraints with friendly errors.

Problem 11 — SQL-injection-free query builder

Prompt: Without an ORM, write a function that executes a SELECT against SQLite with optional where filters and order_by. Must be safe from SQL injection.

Time: 25 min.

Model solution

python

import sqlite3
from typing import Any, Iterable

ALLOWED_ORDERS = {"asc", "desc"}

def query(
    db: sqlite3.Connection,
    table: str,
    *,
    columns: list[str],
    where: dict[str, Any] | None = None,
    order_by: tuple[str, str] | None = None,
    limit: int | None = None,
) -> list[tuple]:
    # Validate identifiers (we CANNOT parameterise them)
    if not table.isidentifier():
        raise ValueError("invalid table name")
    for c in columns:
        if not c.isidentifier():
            raise ValueError(f"invalid column: {c}")

    sql_parts = [f"SELECT {', '.join(columns)} FROM {table}"]
    params: list[Any] = []

    if where:
        clauses = []
        for col, val in where.items():
            if not col.isidentifier():
                raise ValueError(f"invalid where column: {col}")
            clauses.append(f"{col} = ?")
            params.append(val)
        sql_parts.append("WHERE " + " AND ".join(clauses))

    if order_by:
        col, direction = order_by
        if not col.isidentifier() or direction.lower() not in ALLOWED_ORDERS:
            raise ValueError("invalid order_by")
        sql_parts.append(f"ORDER BY {col} {direction.upper()}")

    if limit is not None:
        sql_parts.append("LIMIT ?")
        params.append(int(limit))

    sql = " ".join(sql_parts)
    return db.execute(sql, params).fetchall()

Senior insight: parameter substitution can't be used for identifiers (table/column names). Validate them against an allow-list or str.isidentifier().

Problem 12 — Producer/Consumer with bounded queue (async)

Prompt: One producer generates items at variable rate. Three consumers process them (each takes 100–500 ms). Queue must apply backpressure (max 100 items). Handle graceful shutdown on Ctrl+C.

Time: 30 min.

Model solution

python

import asyncio, random, signal

SENTINEL: object = object()

async def producer(q: asyncio.Queue, n: int) -> None:
    for i in range(n):
        await q.put(i)        # blocks when full → backpressure
    for _ in range(3):
        await q.put(SENTINEL) # one per consumer

async def consumer(q: asyncio.Queue, name: str) -> None:
    while True:
        item = await q.get()
        try:
            if item is SENTINEL:
                return
            await asyncio.sleep(random.uniform(0.1, 0.5))
            print(f"{name} processed {item}")
        finally:
            q.task_done()

async def main() -> None:
    q: asyncio.Queue = asyncio.Queue(maxsize=100)
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(q, 1000))
        for i in range(3):
            tg.create_task(consumer(q, f"c{i}"))

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("shutting down")

Senior idioms: TaskGroup (structured concurrency), Queue(maxsize) for backpressure, sentinel objects (or None if not a valid item), task_done() if using Queue.join().

Problem 13 — Diff two large dicts

Prompt: Two dicts (potentially nested arbitrarily). Return added / removed / changed paths. Path = tuple of keys (("user", "address", "city")).

Time: 25 min.

Model solution

python

from typing import Any

Path = tuple[str, ...]

def diff(a: dict, b: dict, path: Path = ()) -> dict[str, list]:
    added, removed, changed = [], [], []

    def walk(x: Any, y: Any, p: Path) -> None:
        if isinstance(x, dict) and isinstance(y, dict):
            for k in x.keys() - y.keys():
                removed.append(p + (k,))
            for k in y.keys() - x.keys():
                added.append(p + (k,))
            for k in x.keys() & y.keys():
                walk(x[k], y[k], p + (k,))
        elif x != y:
            changed.append((p, x, y))

    walk(a, b, path)
    return {"added": added, "removed": removed, "changed": changed}

Idioms: dict view set ops (a.keys() - b.keys()), recursive walk, no mutation of inputs, typed return.

Problem 14 — Streaming JSON aggregator

Prompt: Read a JSON Lines file (each line is a JSON object). Compute average of field latency_ms per service. File is too large to fit in memory.

Time: 20 min.

Model solution

python

import json
from collections import defaultdict
from pathlib import Path
from typing import Iterable

def stream_lines(path: Path) -> Iterable[dict]:
    with open(path, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                yield json.loads(line)

def avg_latency(path: Path) -> dict[str, float]:
    sums: defaultdict[str, float] = defaultdict(float)
    counts: defaultdict[str, int] = defaultdict(int)
    for rec in stream_lines(path):
        svc = rec.get("service")
        lat = rec.get("latency_ms")
        if svc and isinstance(lat, (int, float)):
            sums[svc] += lat
            counts[svc] += 1
    return {svc: sums[svc] / counts[svc] for svc in sums}

Bonus: ask about exact statistics vs streaming approximations (percentiles need tdigest/hdrhistogram).

Problem 15 — Hot-reloading config

Prompt: A class Settings reads a config.json on init. Add a start_watching() method that re-reads it whenever the file changes (use watchfiles). Other code holds a reference to Settings and should see updates.

Time: 30 min.

Model solution

python

from __future__ import annotations
import asyncio, json, threading
from pathlib import Path
from watchfiles import awatch

class Settings:
    def __init__(self, path: Path) -> None:
        self.path = path
        self._lock = threading.RLock()
        self._data: dict = {}
        self._load()

    def _load(self) -> None:
        with self._lock:
            self._data = json.loads(self.path.read_text(encoding="utf-8"))

    def __getitem__(self, key: str):
        with self._lock:
            return self._data[key]

    async def start_watching(self) -> None:
        async for _ in awatch(self.path):
            try:
                self._load()
            except Exception as e:
                print(f"reload failed: {e}")

Discussion: thread-safety (RLock), failure handling (don't crash on bad file), how to integrate with FastAPI lifespan, sending a signal (callback) on reload for downstream consumers.

Tips for tackling these in the interview

Clarify first: ask about scale, latency, what error semantics, whether tests are expected.
Talk while coding: narrate decisions; if you go quiet for 2 min, you look stuck.
Type hints from the start: shows seniority.
Stdlib first: Counter, defaultdict, OrderedDict, heapq, itertools. Don't write a class when a function works.
Tests: even one assert shows you think tests-first.
Idiom over cleverness: a 5-line solution that uses the stdlib well > 20 lines of clever one-liners.
Discuss trade-offs at the end: what about thread safety? memory? edge cases? "If we had more time…"

These 15 problems cover most "intermediate-Python coding" rounds. Practice each twice — once cold, then again referencing only the standard library docs.