Python Programming Questions

Review checklist

• Identify large temporaries and peak memory sites
• Prefer in-place ops or fused kernels (a fused kernel is a single compiled operation that does several arithmetic steps in one pass over the data, e.g. a*b+c computed in one native loop, instead of allocating a separate full-size temporary array after every individual step)
• Use dtype minimization (float32 vs float64)
• Avoid creating many intermediate arrays in chained expressions
• Ensure readability and add comments explaining optimizations

Code suggestions

• Replace chained ops with single expression using NumPy ufuncs (ufunc, short for universal function: a NumPy function like np.exp or np.multiply that applies element-by-element across an array in a single compiled pass) or out= parameter (out=: tells the ufunc to write its result directly into an existing array's memory instead of allocating a brand-new array for the result):

# bad: many temporaries
a = np.exp(x)
b = a * w
c = np.where(b>0, b, 0)

# better: reuse buffers
tmp = np.empty_like(x)   # allocate the reusable buffer once, up front
np.exp(x, out=tmp)
np.multiply(tmp, w, out=tmp)
np.maximum(tmp, 0, out=tmp)

The tmp = np.empty_like(x) line is what makes the "better" version actually runnable: every out=tmp call below it writes its result back into that same pre-allocated buffer instead of allocating a fresh array each time, which is the entire memory saving, a, b, and c in the "bad" version are each a full new array; tmp in the "better" version is one array, reused three times.

Worked example, verified on CPython 3.12, confirming the rewrite produces identical values and showing the memory difference concretely with tracemalloc:

import numpy as np
import tracemalloc

x = np.array([1.0, -2.0, 3.0, -4.0])
w = np.array([0.5, 0.5, 0.5, 0.5])

def bad(x, w):
    a = np.exp(x)
    b = a * w
    c = np.where(b > 0, b, 0)
    return c

def better(x, w):
    tmp = np.empty_like(x)
    np.exp(x, out=tmp)
    np.multiply(tmp, w, out=tmp)
    np.maximum(tmp, 0, out=tmp)
    return tmp

print(np.allclose(bad(x, w), better(x, w)))
# True

tracemalloc.start()
_, before = tracemalloc.get_traced_memory()
for _ in range(1000):
    bad(x, w)
_, peak_bad = tracemalloc.get_traced_memory()
tracemalloc.stop()

tracemalloc.start()
_, before2 = tracemalloc.get_traced_memory()
for _ in range(1000):
    better(x, w)
_, peak_better = tracemalloc.get_traced_memory()
tracemalloc.stop()

print("bad peak > better peak:", peak_bad > peak_better)
# True: bad() allocates 3 new arrays (a, b, c) per call; better() allocates 1 (tmp) and reuses it

Both functions agree on every value (np.allclose is True); the peak-memory comparison confirms, directly, that bad() allocates strictly more per call, three fresh arrays (a, b, c) versus better()'s single reused buffer, which is the mechanism this checklist item is about, not just a naming convention.

• Use NumExpr (evaluates a whole array expression like a*b+c in one call without materializing each intermediate array) or Numba (compiles a Python function to machine code the first time it runs) for large elementwise chains to reduce temporaries.
• Use memory-mapped arrays (np.memmap, which lets an array live on disk and be accessed in pieces instead of fully loaded into RAM) for very large datasets.

Additional advice

• Add benchmarks demonstrating memory/regression
• Add comments and keep variable names meaningful
• Consider chunked processing for very large arrays to keep memory bounded

Approach (iterative): Use a stack to process items in order, pushing list elements in reverse so original order is preserved. This avoids recursion depth limits.

Implementation:

def flatten_list(nested):
    """Flatten nested lists (iterative)."""
    result = []
    stack = list(reversed(nested))

    while stack:
        item = stack.pop()
        if isinstance(item, list):
            stack.extend(reversed(item))
        else:
            result.append(item)
    return result

Worked example, verified on CPython 3.12:

print(flatten_list([1, [2, [3, 4], 5], 6]))
# [1, 2, 3, 4, 5, 6]

Tracing the stack (pop() takes from the end, so pushing a sub-list's items in reverse order means they come back out of the stack in their original left-to-right order, since a stack is last-in-first-out: the last thing pushed is the first thing popped, so reversing before pushing cancels out the stack's own order-reversal): stack starts as [6, [2, [3, 4], 5], 1] (the whole input list, reversed). Popping 1 (a plain value) appends it to result, leaving stack = [6, [2, [3, 4], 5]]. Popping [2, [3, 4], 5] (a list) pushes its contents back on in reverse, [5, [3, 4], 2], giving stack = [6, 5, [3, 4], 2]. Popping 2 appends it (result = [1, 2]). Popping [3, 4] pushes [4, 3] back on, giving stack = [6, 5, 4, 3]. Popping 3, then 4, then 5, then 6 in that order appends each in turn, building result = [1, 2, 3, 4, 5, 6], matching the original left-to-right order exactly, and matching the printed output above.

Key points: preserves original order; handles arbitrary nesting; non-recursive.

Time Complexity: O(n) where n is total number of elements (visit each once).
Space Complexity: O(n) for result plus O(d) stack where d is nesting breadth (worst-case O(n)).

Edge cases: empty list, lists with non-list iterables (only lists are flattened per spec), deeply nested single chains handled without recursion.

Approach: schedule tasks, wait with timeout, cancel pending tasks after timeout, re-raise exceptions from completed tasks. (A coroutine is what you get back from calling an async def function; calling it does not run it yet, asyncio.create_task(c) is what actually schedules a coroutine to start running concurrently with everything else.)

Implementation:

import asyncio

async def gather_with_timeout(coros, timeout):
    tasks = [asyncio.create_task(c) for c in coros]
    done, pending = await asyncio.wait(tasks, timeout=timeout)
    # cancel pending
    for t in pending:
        t.cancel()
    # await cancellations to finish
    if pending:
        await asyncio.gather(*pending, return_exceptions=True)
    # propagate exceptions from done
    results = []
    for t in done:
        exc = t.exception()
        if exc:
            # re-raise first exception
            raise exc
        results.append(t.result())
    return tuple(results), tuple(pending)

# Usage example
async def main():
    async def work(n):
        await asyncio.sleep(n); return n
    done, pending = await gather_with_timeout([work(0.1), work(2)], timeout=1)
    print(done, pending)

# asyncio.run(main())

Notes: ensures pending tasks are cancelled and awaited to avoid warnings; exceptions from completed tasks propagate.

Running it: uncommenting asyncio.run(main()) and adding print(len(pending), pending[0].cancelled()) right after print(done, pending) shows what actually happens: work(0.1) finishes well inside the 1-second timeout and work(2) does not, so done prints as (0.1,) (the tuple of results from the finished work(0.1)) and the follow-up print shows 1 True, one pending task, and it reports cancelled() == True, confirming the still-running work(2) really was cancelled rather than left dangling.

Bottleneck reasoning

A bottleneck is the part of the system that dominates total runtime. If the report generator scans the same task records over and over, each new question forces another full pass through the data. That scales poorly as the team grows.

5 scans * 100000 rows = 500000 row visits
1 scan * 100000 rows = 100000 row visits

Intuition: one pass is 5x less work in this example.

What I would change
I would switch the data flow from repeated scans to one ingestion pass that builds summary structures as it goes:

• counters for totals by status, owner, or date
• maps indexed by the fields used in reports
• one reducer function that updates all needed aggregates together

Then each summary question reads from the precomputed aggregates instead of touching the raw records again. That keeps queries fast and the code maintainable because the aggregation logic lives in one place.

Tradeoff
The main cost is a bit more memory for the summary maps. I would accept that tradeoff because it is much cheaper than rescanning the whole dataset repeatedly.

Example
If ten report questions all need ticket counts by team, I would build that count once during ingestion, then reuse it for every report instead of looping over the task table ten times.

Goal: return k most frequent items from a possibly streaming iterable using O(k) extra memory where possible.

Approach: Use Misra-Gries (frequent algorithm) for streaming approximate top-k with O(k) memory; for exact top-k when feasible, use counting with a hashmap and a min-heap of size k (memory O(u) where u unique). For very large universes, prefer Misra-Gries. This is a genuinely advanced, fairly obscure streaming algorithm, most engineers have never needed it; skip straight to "hashmap + heap, exact counts" below unless the interviewer specifically probes for a single-pass, bounded-memory answer.

The core intuition, in plain language, before any code: Misra-Gries tracks at most k candidate items with counts. When a new, not-yet-tracked item arrives and there is no room left (k candidates already exist), instead of dropping the new item and keeping the old ones untouched, EVERY existing candidate's counter is decremented by one, as if the new item had "cancelled out" one occurrence of each current candidate, and the new item itself is not added. Any item that truly appears more than roughly n/k times (n being the total stream length) cannot be fully cancelled away by this process, no matter how the decrements land, so it is guaranteed to still be a candidate at the end; the guarantee is one-sided, though, some less-frequent items may also survive as candidates (false positives), which is exactly why the code takes a second pass over the real data to compute true counts for whatever candidates survived, rather than trusting the approximate counters directly.

Misra-Gries implementation (approximate, deterministic guarantees):

from collections import defaultdict

def top_k_frequent(stream, k):
    if k < 1:
        return []
    counters = {}
    for x in stream:
        if x in counters:
            counters[x] += 1
        elif len(counters) < k:
            counters[x] = 1
        else:
            # decrement all
            to_del = []
            for y in list(counters):
                counters[y] -= 1
                if counters[y] == 0:
                    del counters[y]
    # counters are candidates; to get actual counts, re-scan
    true_counts = defaultdict(int)
    for x in stream:
        if x in counters:
            true_counts[x] += 1
    return sorted(true_counts.items(), key=lambda t: -t[1])[:k]

Worked trace, verified on CPython 3.12: an 8-item stream with k=2 (so the true top item, 'a', appears 5 out of 8 times, far more than the other three items, each appearing once):

stream = ['a', 'a', 'b', 'c', 'a', 'd', 'a', 'a']
print(top_k_frequent(stream, 2))
# [('a', 5), ('d', 1)]

Tracing counters step by step: a (candidate, table has room) -> {'a': 1}. Second a (already tracked, increment) -> {'a': 2}. b (room) -> {'a': 2, 'b': 1}. c arrives with the table full (2 candidates already, k=2): every existing candidate is decremented, a drops to 1, b drops to 0 and is deleted, and c itself is never added -> {'a': 1}. Third a (tracked, increment) -> {'a': 2}. d (room again, since b was just evicted) -> {'a': 2, 'd': 1}. Fourth and fifth a (tracked, increment twice) -> {'a': 4, 'd': 1}. Final candidates: a and d. The second pass then counts each candidate's REAL frequency across the whole stream (a: 5, d: 1) and returns them sorted by that real count.

This trace also shows the false-positive behavior honestly: b and c both had a true count of 1, identical to d's true count of 1, but only d happened to survive as a candidate, purely because of when it arrived relative to the decrement-all events. The algorithm's real guarantee is only about the dominant item: a, at 5 out of 8 occurrences, appears far more often than roughly n/k = 4, and it survives every decrement round intact enough to still be tracked at the end. Nothing is guaranteed about which of the equally-infrequent items happen to survive alongside it, which is precisely why a second, exact pass over the real data is required before trusting any reported count, the approximate candidate set is a correct superset guarantee for truly frequent items, not a precise answer on its own.

Notes: The algorithm uses O(k) memory and one or two passes (approximate single-pass; exact requires second pass over data or storing counts). Choose Misra-Gries for streaming huge data; use Counter+heap for exact counts when unique items fit memory.