Python Fundamentals and Core Syntax Questions

Comprehensive knowledge of core Python language features and syntax, including primitive and composite data types such as integer numbers, floating point numbers, strings, booleans, lists, dictionaries, sets, and tuples. Candidates should understand variable assignment and naming, operators for arithmetic, logical, and comparison operations, and control flow constructs including conditional statements and loops. Expect familiarity with function definition, invocation, parameter passing, return values, and scope rules, as well as common built in functions and idioms such as iteration utilities, list comprehensions, generator expressions, and basic functional utilities like map and filter. Candidates should demonstrate error and exception handling techniques and best practices for writing readable and maintainable code with modularization and clear naming. Practical skills include file input and output, working with common data formats such as comma separated values and JavaScript Object Notation, selecting appropriate data structures with attention to performance and memory characteristics, and applying memory efficient patterns for processing large data sets using iterators and generators. Familiarity with the standard library and common utilities for parsing and transforming data, and the ability to write small code snippets to solve algorithmic and data manipulation tasks, are expected.

HardTechnical

62 practiced

You have expensive CPU-bound preprocessing functions that need to be parallelized in Python. Compare the trade-offs between using multithreading, multiprocessing, and asynchronous I/O for this workload. Then provide a sample implementation using concurrent.futures.ProcessPoolExecutor that parallelizes a CPU-bound function over chunks while minimizing memory copying (for example via memory-mapped files or numpy memmap).

Sample Answer

Multithreading vs multiprocessing vs async I/O (brief comparison)- Multithreading: Python threads share memory but are limited by the GIL for CPU-bound tasks, so threads won't speed up heavy CPU-bound preprocessing in pure Python (unless C extensions release GIL, e.g., numpy/BLAS ops). Good for I/O-bound work.- Multiprocessing: Spawns separate processes with independent Python interpreters; bypasses GIL and scales CPU-bound work across cores. Downsides: higher process startup cost and potential memory duplication when passing large arrays—unless you use shared memory (memmap, shared_memory) to avoid copies.- Async I/O: Cooperative concurrency for I/O-bound tasks; not useful for CPU-heavy preprocessing because it doesn't add parallel CPU execution.

Recommended: multiprocessing with ProcessPoolExecutor combined with memory-mapped files or multiprocessing.shared_memory to avoid copying large arrays.

Sample implementation- We create a numpy memmap for input and output on disk; processes open the same memmap and operate on assigned slices. Use ProcessPoolExecutor with an initializer to open memmap once per worker to reduce overhead.

python

import numpy as np
from concurrent.futures import ProcessPoolExecutor, as_completed
import os

INPUT_PATH = "input.dat"
OUTPUT_PATH = "output.dat"
DTYPE = np.float32
SHAPE = (10_000_000,)  # example large 1D array
CHUNK_SIZE = 500_000

# Worker-global references set by initializer to avoid reopening repeatedly
_worker_in = None
_worker_out = None

def _init_worker(input_path, output_path, shape, dtype):
    global _worker_in, _worker_out
    # open memmaps read-only / read-write in each worker
    _worker_in = np.memmap(input_path, dtype=dtype, mode='r', shape=shape)
    _worker_out = np.memmap(output_path, dtype=dtype, mode='r+', shape=shape)

def process_chunk(slice_tuple):
    """CPU-bound preprocessing applied to a slice of the memmap.
    slice_tuple = (start, end)
    """
    start, end = slice_tuple
    arr_view = _worker_in[start:end]          # view backed by memmap (no copy)
    # Example expensive operation (vectorized) - replace with actual preprocessing
    # Use operations that avoid creating many temporaries where possible
    out_view = np.sqrt(arr_view) * 2.0
    _worker_out[start:end] = out_view         # write directly to output memmap
    return (start, end)

def parallel_preprocess():
    # create input memmap file if not exists (demo: fill with data)
    if not os.path.exists(INPUT_PATH):
        mm = np.memmap(INPUT_PATH, dtype=DTYPE, mode='w+', shape=SHAPE)
        mm[:] = np.random.rand(*SHAPE).astype(DTYPE)
        mm.flush()
        del mm

    # prepare output memmap of same shape
    out = np.memmap(OUTPUT_PATH, dtype=DTYPE, mode='w+', shape=SHAPE)
    out[:] = 0
    out.flush()
    del out

    # generate chunk ranges
    slices = []
    for i in range(0, SHAPE[0], CHUNK_SIZE):
        slices.append((i, min(i + CHUNK_SIZE, SHAPE[0])))

    # run parallel processing
    with ProcessPoolExecutor(max_workers=os.cpu_count(),
                             initializer=_init_worker,
                             initargs=(INPUT_PATH, OUTPUT_PATH, SHAPE, DTYPE)) as exe:
        futures = [exe.submit(process_chunk, s) for s in slices]
        for fut in as_completed(futures):
            start, end = fut.result()
            # optional: log progress
            # print(f"Processed {start}:{end}")

if __name__ == "__main__":
    parallel_preprocess()

Key points and trade-offs- Using numpy.memmap avoids copying large arrays between processes; each process maps the same file into its address space.- Keep chunk sizes large enough to amortize IPC and scheduling overhead; too-small chunks cause excessive overhead.- If writes overlap, coordinate locks; here slices are disjoint so no locking needed.- Alternatives: multiprocessing.shared_memory (zero-copy) for in-RAM shared buffers; use with care for lifecycle/cleanup.- Watch disk I/O: memmap backed by HDD/SSD may be slower than in-memory shared_memory; on fast NVMe the overhead is small.- Ensure your numpy operations are vectorized to minimize Python-level loops inside workers.

EasyTechnical

72 practiced

Describe the core built-in Python data types: int, float, bool, str, list, tuple, set, and dict. For each type give a concise example (in Python), state whether it is mutable or immutable, typical machine-learning pipeline use cases (for example feature vectors, configuration maps, unique id sets), and performance implications for membership testing, iteration, and memory footprint. Mention any gotchas an ML engineer should be aware of when choosing between these types.

Sample Answer

intExample:

python

x = 42

Mutable: immutableML use cases: counters, epochs, class labels (indices)Performance: membership not applicable; iteration trivial; small memory footprint per integer (platform-dependent). Gotcha: Python int is arbitrary precision (slow for large arrays) — prefer numpy dtypes for large numeric arrays.

floatExample:

python

f = 3.14

Mutable: immutableML use cases: model weights, losses, hyperparametersPerformance: iteration/ops fine but Python-level floats are slow for vector math — use numpy/pytorch tensors. Gotcha: floating-point precision and reproducibility issues.

boolExample:

python

b = True

Mutable: immutableML use cases: masks, flags, binary labelsPerformance: tiny footprint; subclass of int (True==1). Gotcha: can appear in arithmetic; use boolean arrays (numpy) for vectorized masks.

strExample:

python

s = "feature_name"

Mutable: immutableML use cases: feature names, file paths, config keysPerformance: membership (char in s) is O(n); iteration O(n); relatively compact but copies on modification. Gotcha: slicing creates new strings; for many tokens use intern() or categorical encodings.

listExample:

python

lst = [1.0, 2.0, 3.0]

Mutable: mutableML use cases: small sequences, pipelines steps, batch of heterogenous itemsPerformance: membership test O(n), iteration O(n) with low overhead; memory: array of pointers to Python objects (high overhead per element). Gotcha: not efficient for numeric heavy work — use numpy arrays.

tupleExample:

python

t = (1, 2, 3)

Mutable: immutableML use cases: fixed feature schemas, keys for composite identifiers, function returnsPerformance: same complexity as list for membership and iteration; slightly better memory and faster than list due to immutability. Gotcha: immutability makes them safe as dict keys.

setExample:

python

s = {"a","b","c"}

Mutable: mutableML use cases: unique id sets, vocabulary, fast deduplicationPerformance: membership O(1) average, iteration O(n) but unordered; memory: hash table overhead (larger). Gotcha: unhashable elements (lists) cannot be stored; order not preserved — use frozenset for immutability.

dictExample:

python

d = {"feature": 0.5, "bias": 1.0}

Mutable: mutableML use cases: configuration maps, feature->value lookups, label mapsPerformance: key lookup O(1) average, iteration over keys/values O(n); memory: significant overhead per entry for hash table. Gotcha: keys must be hashable and stable; insertion order preserved since Python 3.7 (useful), but large dicts are memory-heavy — consider arrays or structured arrays for dense numeric data.

General ML guidance / gotchas- For numeric vectors/matrices, use numpy, pandas, or tensors instead of Python lists/tuples for performance, memory efficiency, and vectorized ops.- Avoid storing millions of numeric values in Python containers; prefer typed arrays.- Choose immutable types as stable keys; beware of accidental mutation when caching.- Hash-based types give O(1) membership but cost memory; lists cost less memory for tiny collections but are O(n) for lookup.

MediumTechnical

64 practiced

Design and implement a Python 3 DatasetIterator class that accepts any iterable data source and yields minibatches of size batch_size. The iterator should support an optional shuffle flag and a reproducible random seed and handle the last incomplete batch by yielding it or dropping it based on a parameter. Describe how this iterator could integrate with a training loop.

Sample Answer

Approach: convert the incoming iterable to a sequence if it's not already (we need random access to shuffle reproducibly). Use an index list that we shuffle via a local random.Random(seed) to avoid global state. Yield slices of indices as minibatches; optionally drop the final incomplete batch.

python

import collections.abc
import random
from typing import Iterable, Iterator, List, Any, Optional

class DatasetIterator:
    def __init__(self,
                 data: Iterable,
                 batch_size: int,
                 shuffle: bool = False,
                 drop_last: bool = False,
                 seed: Optional[int] = None):
        if batch_size <= 0:
            raise ValueError("batch_size must be > 0")
        # Materialize into a sequence if needed (allows single-pass iterables)
        if isinstance(data, collections.abc.Sequence):
            self._data = data
        else:
            self._data = list(data)  # trade-off: one-time memory cost
        self.n = len(self._data)
        self.batch_size = batch_size
        self.shuffle = shuffle
        self.drop_last = drop_last
        self.seed = seed
        self._rand = random.Random(seed)
        self._reset_epoch()

    def _reset_epoch(self):
        # create index order for this epoch
        self._indices = list(range(self.n))
        if self.shuffle:
            self._rand.shuffle(self._indices)
        self._pos = 0

    def __iter__(self) -> Iterator[List[Any]]:
        self._reset_epoch()
        return self

    def __next__(self) -> List[Any]:
        if self._pos >= self.n:
            raise StopIteration
        end = self._pos + self.batch_size
        if end > self.n:
            if self.drop_last:
                raise StopIteration
            end = self.n
        batch_idx = self._indices[self._pos:end]
        self._pos = end
        # return actual items in the original order of the sampled indices
        return [self._data[i] for i in batch_idx]

    def __len__(self):
        # number of batches per epoch
        if self.drop_last:
            return self.n // self.batch_size
        return (self.n + self.batch_size - 1) // self.batch_size

Key concepts and reasoning:- Materialize non-sequence iterables so we can index and shuffle reproducibly. This is common in data loaders; streaming loaders would require a different design (buffering/epoch-aware IO).- Use a dedicated random.Random(seed) for reproducible shuffles without affecting global RNG.- We shuffle indices, not data, to avoid copying large samples repeatedly.

Time & space complexity:- Time: O(n) to materialize (if needed) + O(n) per epoch for shuffling; batch construction is O(batch_size) per batch.- Space: O(n) extra for materialized data and O(n) for index list.

Edge cases:- Empty dataset (iterator yields nothing).- batch_size > n (yields one batch unless drop_last True).- Non-reproducible behavior avoided by using local RNG.- For very large datasets, avoid list(data) and implement streaming+shuffle-on-disk or reservoir strategies.

Integration with a training loop (example):

python

# pseudo-training loop
iterator = DatasetIterator(dataset, batch_size=64, shuffle=True, seed=42)
for epoch in range(epochs):
    for batch in iterator:
        X_batch, y_batch = collate(batch)  # convert raw samples to tensors
        preds = model(X_batch)
        loss = loss_fn(preds, y_batch)
        loss.backward()
        optimizer.step()
        optimizer.zero_grad()
    # optionally re-seed or rely on iterator.__iter__ reset to shuffle next epoch

This iterator is simple, reproducible, and fits typical ML training patterns. For production/large-scale, extend to lazy loading, multiprocessing, and async prefetching.

HardTechnical

70 practiced

Create a Python 3 decorator 'timed_lru_cache' that both times function execution and caches results. The decorator should handle unhashable arguments by serializing them deterministically (for example via json.dumps with sort_keys) before using as cache keys. Discuss the limitations and security considerations of using serialization for cache keys and show example usage.

Sample Answer

Approach: implement a decorator factory timed_lru_cache(maxsize=128) that:- Builds deterministic cache keys by serializing args + kwargs via json.dumps(..., sort_keys=True).- Falls back to a safe, explicit fallback for non-JSON-serializable objects (with clear security caveats).- Tracks LRU eviction with collections.OrderedDict.- Measures execution time (time.perf_counter) and stores last runtime on the cached entry.

python

import json
import time
import functools
from collections import OrderedDict

def _make_key(args, kwargs):
    """
    Deterministic key: try JSON with sort_keys. If that fails, use repr fallback.
    Using repr is less deterministic across runs/versions but avoids insecure pickle.
    """
    try:
        key = json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True, default=lambda o: repr(o), separators=(",", ":"))
        return key
    except (TypeError, ValueError):
        # As a last resort; note: repr may vary and isn't collision-proof
        return repr((args, tuple(sorted(kwargs.items()))))

def timed_lru_cache(maxsize=128):
    def decorator(func):
        cache = OrderedDict()  # key -> (result, elapsed_seconds)
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = _make_key(args, kwargs)
            if key in cache:
                # Move to end => most recently used
                result, elapsed = cache.pop(key)
                cache[key] = (result, elapsed)
                wrapper.hits += 1
                return result
            wrapper.misses += 1
            start = time.perf_counter()
            result = func(*args, **kwargs)
            elapsed = time.perf_counter() - start
            cache[key] = (result, elapsed)
            # enforce maxsize
            if len(cache) > maxsize:
                cache.popitem(last=False)  # evict least recently used
            return result
        # attach stats
        wrapper.hits = 0
        wrapper.misses = 0
        wrapper.cache = cache  # expose for inspection
        wrapper.get_last_time = lambda *a, **k: cache.get(_make_key(a, k), (None, None))[1]
        return wrapper
    return decorator

Key points:- Uses deterministic JSON serialization where possible (sort_keys=True) so equivalent dicts/lists map to same key.- Falls back to repr for objects that aren't JSON-serializable; this is less robust.

Time & space complexity:- Cache lookup/insertion: amortized O(1) using OrderedDict.- Serialization cost: O(size of args) per call; can dominate for large inputs.

Edge cases and limitations:- Mutable arguments: if caller mutates a list after caching, cached result might become inconsistent. Prefer immutables or deep-copy inputs before using cache.- Non-deterministic repr: repr-based fallbacks can change across versions or produce collisions.- Serialization cost: heavy for large tensors/arrays; for ML models prefer hashing based on ids/checksums (e.g., stable hash of numpy arrays) instead of full serialization.- Thread-safety: this implementation is not thread-safe; use threading.Lock around cache accesses in concurrent contexts.

Security considerations:- Avoid pickle for key serialization when inputs come from untrusted sources—pickle can execute arbitrary code. JSON is safer but won't handle all types.- Be cautious exposing repr strings or serialized forms in logs (may leak sensitive data).- If you must support complex objects, implement explicit safe serializers (e.g., for numpy arrays use .tobytes() + a hash) and whitelist types.

Example usage:

python

@timed_lru_cache(maxsize=64)
def expensive_transform(data):
    # data can be a list/dict; serialization will create the cache key
    time.sleep(0.1)  # simulate work
    return sum(data)

print(expensive_transform([1,2,3]))  # computes, caches
print(expensive_transform([1,2,3]))  # cache hit, faster
print("hits", expensive_transform.hits, "misses", expensive_transform.misses)
print("last call time (cached):", expensive_transform.get_last_time([1,2,3]))

Alternatives:- For ML payloads (numpy, tensors), compute a deterministic hash (e.g., SHA256 of bytes + shape + dtype) rather than serializing full content to reduce cost and avoid collisions/security pitfalls.- Use functools.lru_cache when arguments are hashable; combine with custom wrappers that map non-hashable args to hashable tokens.

MediumTechnical

81 practiced

A Python-based model serving process leaks memory gradually in production. Describe the debugging steps you would take to find and fix the leak using standard Python tools (tracemalloc, gc, objgraph, logging). Include a prioritized list of quick mitigations you might apply to reduce the business impact while you find a permanent fix.

Sample Answer

Situation: A Python model-serving process in production shows gradual memory growth causing OOMs.

Approach / Steps (prioritized):1. Reproduce & baseline- Confirm leak pattern (RSS growth over time) with monitoring (Prometheus/Grafana), note process PID and timeline.

2. Add lightweight instrumentation- Enable tracemalloc snapshots at intervals to see growing allocations.- Force GC and log object counts periodically.

Example instrumentation:

python

import tracemalloc, gc, logging
tracemalloc.start()
logging.basicConfig(level=logging.INFO)

def snapshot(tag):
    gc.collect()
    snapshot = tracemalloc.take_snapshot()
    top = snapshot.statistics('lineno')[:10]
    logging.info("%s top allocations: %s", tag, top)

3. Narrow suspects with gc and objgraph- Use gc.get_objects(), gc.get_referrers() to find large groups.- Use objgraph to show growth of specific types (e.g., dict, list, torch.Tensor, numpy.ndarray, Response objects).

python

import objgraph
objgraph.show_growth(limit=20)
objgraph.show_backrefs([obj], max_depth=5)

4. Inspect ML-specific resources- Check for detached tensors retained via closure, caching of inference results, global caches, thread-local storage, connection/session pools, model warmup data.- For PyTorch: ensure .detach() and .cpu() before storing, call torch.cuda.empty_cache() when appropriate and avoid storing tensors in lists/dicts.

5. Use tracemalloc diffing- Take snapshots at t0 and t1, use snapshot.compare_to() to find leaking sources (files/lines).

6. Validate hypothesis & fix- Patch code: remove unintended references, clear caches after use, close sessions, use weakrefs for caches, explicit del + gc.collect in controlled places, fix circular refs.- Add unit/integration test that exercises long-running inference and asserts stable memory.

Quick mitigations (prioritized to reduce business impact):1. Process recycling: implement worker max requests/restart (e.g., gunicorn --max-requests) to bound memory.2. Reduce concurrency/batch size to lower peak memory.3. Scale horizontally: add more instances while restarting problematic ones.4. Enable memory limits in container (cgroups/Kubernetes) with graceful restart on threshold.5. Turn off or limit nonessential caching/logging that stores objects.6. Fall back to a simpler model version (smaller memory footprint) temporarily.

Why this works:- tracemalloc pinpoints allocation sites; gc + objgraph reveal retained references; ML-specific checks target common pitfalls (tensor retention, GPU cache). Quick mitigations reduce customer impact while you isolate and patch the root cause.

Unlock Full Question Bank

Get access to hundreds of Python Fundamentals and Core Syntax interview questions and detailed answers.

Join thousands of developers preparing for their dream job.