Python Programming Questions

Direct answer

A bounded queue (fixed maxsize) applies backpressure automatically: once it is full, put() blocks (threading) or suspends (await put, asyncio) until a consumer frees a slot, so a fast producer is throttled to the consumers' pace without any extra code. queue.Queue gives you this with an internal threading.Condition; asyncio.Queue gives you the same shape with await instead of blocking waits. The wake-up semantics differ by mechanism: a raw condition variable needs you to choose notify() (wake exactly one waiter) versus notify_all() (wake everyone, most of whom will just recheck and go back to waiting), while asyncio.Queue hides that choice entirely, await put/await get suspend and resume the right coroutine without you managing wake-ups by hand.

Structured elaboration

Threading version: queue.Queue(maxsize=N) is backed by a Lock plus two Conditions internally (not-full, not-empty). put() blocks on the not-full condition when the queue is at capacity; get() blocks on not-empty when it is empty. You do not touch the condition variables yourself, Queue already calls notify() correctly.

asyncio version: a coroutine is a function defined with async def that can pause mid-execution, at an await, and let other coroutines run, then resume later exactly where it left off; the event loop is the scheduler that decides which paused coroutine gets to run next. asyncio.Queue(maxsize=N) has the same contract as the threading version, but await q.put(item) suspends the coroutine (yielding control to the event loop, not blocking an OS thread) when full, and await q.get() suspends when empty. Concurrency comes from running many producer/consumer coroutines as tasks on one event loop instead of many OS threads.

Graceful shutdown, both versions: signal producers to stop (an Event), let running producers finish their current item, then drain the queue (q.join(), which waits until every put item has had a matching task_done()), then push one sentinel value per consumer so each one exits its loop cleanly. q.join() hanging is almost always a sign that a consumer path skipped task_done(), most often via an exception on the processing line before it reached the task_done() call.

The harder variant, without queue.Queue/asyncio.Queue: you build the bounded buffer yourself from a Lock plus two Condition objects sharing that lock, one for "not full", one for "not empty." This is exactly what queue.Queue does internally, made explicit:

import threading, collections

class BoundedBuffer:
    def __init__(self, maxsize):
        self.maxsize = maxsize
        self.items = collections.deque()
        self.lock = threading.Lock()
        self.not_full = threading.Condition(self.lock)
        self.not_empty = threading.Condition(self.lock)

    def put(self, item):
        with self.not_full:
            while len(self.items) >= self.maxsize:
                self.not_full.wait()      # releases the lock, sleeps until notified
            self.items.append(item)
            self.not_empty.notify()       # wake exactly one waiting consumer

    def get(self):
        with self.not_empty:
            while not self.items:
                self.not_empty.wait()
            item = self.items.popleft()
            self.not_full.notify()        # wake exactly one waiting producer
            return item

notify() (one waiter) is correct here because exactly one slot opened up, waking everyone with notify_all() would just cause the other waiters to recheck their while condition and go back to sleep, which is harmless but wasteful (a "thundering herd" of wasted wake-ups). notify_all() earns its keep when a single event can satisfy many different waiters at once, which is not the case for a single-slot state change like this.

Worked example

asyncio.Queue-based pipeline with graceful shutdown, verified on CPython 3.12 (produced and consumed counts always match, since q.join() guarantees every item was drained before sentinels are sent; the exact counts vary run to run because they depend on real-time scheduling):

import asyncio, random

SENTINEL = object()

async def producer(pid, q, stop_event, rng):
    i = 0
    while not stop_event.is_set():
        await q.put((pid, i))          # blocks here once the queue is full
        i += 1
        await asyncio.sleep(rng.random() * 0.02)

async def consumer(cid, q, processed):
    while True:
        item = await q.get()
        if item is SENTINEL:
            q.task_done()
            break
        processed.append((cid, item))
        q.task_done()

async def run_pipeline(n_producers=2, n_consumers=3, maxsize=5, run_time=0.3, seed=7):
    rng = random.Random(seed)
    q = asyncio.Queue(maxsize=maxsize)
    stop_event = asyncio.Event()
    processed = []

    producers = [asyncio.create_task(producer(i, q, stop_event, rng)) for i in range(n_producers)]
    consumers = [asyncio.create_task(consumer(i, q, processed)) for i in range(n_consumers)]

    await asyncio.sleep(run_time)
    stop_event.set()
    await asyncio.gather(*producers)
    await q.join()                          # wait until every produced item is processed
    for _ in consumers:
        await q.put(SENTINEL)
    await asyncio.gather(*consumers)
    assert len(processed) == sum(1 for _ in processed)  # every produced item made it through

asyncio.run(run_pipeline())

A representative run reported "total produced: 62, total consumed: 62"; the invariant produced == consumed holds on every run because q.join() will not return until all items are drained, the specific count itself is not reproducible (it depends on wall-clock scheduling) and is not the claim being made here.

Trade-offs & pitfalls

• Complexity: $O(1)$ per item for queue operations; space is bounded by maxsize for the queue itself, plus $O(k)$ for the $k$ in-flight producer/consumer tasks.
• Edge case: if a consumer raises an exception before calling task_done(), q.join() hangs forever; wrap the processing body in try/finally so task_done() always runs.
• Edge case: sending fewer sentinels than there are consumers leaves some consumers blocked on get() forever; one sentinel per consumer is required, not one total.
• Choosing notify() when the change actually affects multiple waiters (rare for a single bounded queue, more common in custom condition-based coordination) silently starves the other waiters; if you are not certain only one waiter's condition changed, notify_all() is the safer default even though it wakes more coroutines/threads than strictly necessary.
• The asyncio version scales to many more concurrent producers/consumers than the threading version for the same memory budget, since coroutines are far cheaper than OS threads, but it only helps if the "work" inside producer/consumer is itself non-blocking; a CPU-bound consumer inside an asyncio pipeline will stall the whole event loop exactly like any other CPU-bound coroutine.

Tests and benchmarks:

• Correctness tests: unit tests comparing NumPy reference outputs vs Numba outputs on varied inputs including edge cases (NaN, infinities, zero-length arrays, non-contiguous/strided arrays). Use property-based tests (the hypothesis library: instead of writing individual test inputs by hand, you describe a rule the output must always satisfy, and the library generates many random inputs trying to break that rule) for broader coverage.
• Numerical accuracy: assert tolerances (rtol/atol, relative and absolute tolerance) for floating ops, using np.testing.assert_allclose rather than exact equality, since compiled and interpreted floating-point code can differ in the last few bits even when both are correct.
• Performance benchmarks: microbenchmarks measuring time and compilation overhead (first call vs steady-state) using asv (airspeed velocity, a benchmarking tool that tracks a function's performance across commits, so a regression shows up tied to a specific commit rather than just "it got slower at some point") or pytest-benchmark.

A minimal correctness check, concretely (verified on CPython 3.12; the pattern below is the standard, documented Numba usage, whether or not numba happens to be installed in the environment reading this):

import numpy as np
from numba import njit

def f_numpy(x):
    return np.sqrt(x ** 2 + 1)

@njit
def f_numba(x):
    out = np.empty_like(x)
    for i in range(x.shape[0]):
        out[i] = (x[i] ** 2 + 1) ** 0.5
    return out

x = np.array([0.0, 1.0, 2.0, 3.0])
np.testing.assert_allclose(f_numpy(x), f_numba(x))
# passes silently if the two implementations agree within tolerance;
# assert_allclose raises AssertionError with the offending values if they don't

This is the shape every correctness test in this answer means: compute the same thing both ways, on real (including edge-case) input, and assert they agree, rather than eyeballing the code and assuming they do.

Gotchas & how to test:

• Compilation overhead: measure both cold (first call, which includes the one-time cost of compiling the function for that argument type) and hot runs (every call after, which reuse the already-compiled machine code); prefer caching compiled functions when appropriate.
• Unsupported Python features: ensure no reliance on Python objects in hot loops; test for object-mode fallback by enabling nopython=True in tests and catching TypingError. Numba's default @njit mode (short for nopython=True) compiles the whole function straight to machine code with no Python-object fallback; if some part of the function cannot be compiled that way, @njit raises TypingError immediately rather than silently degrading, which is exactly what you want in tests: a raised TypingError tells you precisely which line Numba could not compile, instead of the function quietly running slow through the interpreter (the older, more permissive default mode's "object-mode fallback").
• ABI/dtype differences: test with different dtypes and memory layouts (C/F contiguous, non-contiguous views). ABI (application binary interface) is the compiled-code calling convention two pieces of native code, here a NumPy build and a Numba-compiled function, must agree on to call each other correctly; a mismatch is rare but shows up as crashes or garbage values rather than a clean error. C-contiguous means an array's elements sit one after another in memory in row-major order (the default for a freshly created array); F-contiguous (Fortran-contiguous) means the same but column-major; a strided or non-contiguous view (e.g. arr[::2]) is neither, and a Numba function compiled assuming one layout can behave incorrectly or need an implicit, costly copy when handed the other, which is exactly why testing with a strided view as input matters, not just a fresh contiguous array.
• Thread-safety: test with multithreaded calls if using parallel=True; check for race conditions.

Automation: integrate benchmarks in CI (nightly) and gate correctness on PRs; document performance expectations and maintain regression alerts.

Debug & remediation plan:

1. Reproduce locally

• Run the test repeatedly: pytest -k name --maxfail=1 -q --count=100 (pytest-repeat plugin) to reproduce flakiness.
• Run under same env as CI (Python version, env vars).

2. Collect traces

• Add logging with timestamps or use pytest -s; capture stack traces on timeout by increasing timeout to inspect where it hangs.
• Use faulthandler: in test setup call faulthandler.dump_traceback_later and dump on timeout.
• Use pytest --durations and --showlocals to find slow parts.

3. Isolate cause

• Check for test order dependency: run single test and with -k to see isolation.
• Mock external resources (network, DB, filesystem) and use deterministic fixtures.
• Check for race conditions: add sleeps, use thread/async synchronization, run under race detectors or sanitizers if available.

4. Fixes

• Replace real I/O with fixtures/mocks or use ephemeral resources.
• Make async tests use event loop proper awaits and timeouts; use pytest-timeout to fail faster.
• Ensure teardown always cleans up (use tmp_path, monkeypatch).

5. Enforce stability in CI

• Add time budget and stricter assertions; mark inherently flaky tests with @pytest.mark.flaky only after triage.
• Run flaky-test detection in CI: rerun once and fail if consistently flaky.
• Add metrics and alerting if test durations increase.

Result: deterministic, fast tests and fewer CI interruptions.

Worked walkthrough on one concrete flaky test: say test_worker_processes_queue starts a background thread that pushes a result onto a queue.Queue, then the test does time.sleep(0.05) and asserts the queue has an item, assuming 50ms is always enough for the worker thread to finish. On a fast, quiet machine this passes almost every time; on a loaded CI runner the worker thread sometimes has not finished by the time the fixed sleep ends, and the assertion fails, only sometimes, which is exactly what "flaky" means.

1. Reproduce locally: pytest -k test_worker_processes_queue --count=100 (--count comes from the pytest-repeat plugin, which adds that option to run the same test that many times in one pytest invocation) runs the test repeatedly in one process; a genuinely flaky test now fails on some of those runs and passes on others, in the same environment, which rules out "it only fails in CI" as an explanation and confirms the flakiness is reproducible on demand rather than a one-off fluke.
2. Collect traces: pytest --durations=0 --showlocals (--durations=0 prints every test's wall-clock duration sorted slowest-first instead of only the slowest few; --showlocals prints each local variable's value at the point of failure) shows, on a failing run, that the assertion failed with the queue still empty: the worker thread genuinely had not produced a result yet, not that the assertion logic itself was wrong. faulthandler.dump_traceback_later (a standard-library facility you arm with a countdown; if the process is still alive when the countdown expires, it dumps every thread's current stack trace, which is exactly what you need for a hang rather than a fast failure) would show the worker thread still inside its own processing function at the moment the assertion already gave up, confirming a genuine race between the fixed sleep and the thread's actual completion time, not a logic bug in the worker itself.
3. Isolate cause: running the test alone (not the full suite) still reproduces it sometimes, which rules out cross-test state leakage as the cause; the pattern (passes when the machine is quiet, fails more often when it is busy) is the signature of a race condition between a fixed-duration sleep and actual, variable-duration background work, not a logic bug.
4. Fix: replace the fixed time.sleep(0.05) with an actual wait on the real completion signal, queue.get(timeout=5), which blocks only as long as it actually needs to (returning the moment the worker pushes its result, and only failing the test if 5 real seconds pass with nothing produced), removing the race entirely instead of tuning the sleep duration and hoping it is long enough.
5. Enforce in CI: add pytest-timeout (a plugin that fails any individual test that runs longer than a configured limit, rather than letting a hang stall the whole CI job indefinitely) as a backstop for any future test that hangs outright, and only reach for @pytest.mark.flaky (a marker that tells a plugin to automatically retry a failing test before reporting it failed) after a test like this has actually been triaged and understood, never as a first response to red CI, since it hides the exact race just diagnosed instead of fixing it.

A race detector or sanitizer (a class of tool, e.g. a language's -race flag or ThreadSanitizer, that instruments a program at build or run time to flag unsynchronized concurrent access to shared memory as it happens, rather than waiting for it to occasionally produce a wrong answer) is the equivalent tool one level down, in compiled languages with real memory-level data races. Python's GIL rules out that specific class of memory corruption, so the direct equivalent here is exactly the timing/synchronization walkthrough above, not a separate tool to reach for.

Direct answer

Immutable built-in types cannot be changed after creation; any operation that looks like a change actually builds a new object. That group includes int, float, str, tuple, frozenset, and bytes. Mutable built-in types can be changed in place, keeping the same identity; that group includes list, dict, set, and bytearray. The practical consequence that trips people up: because Python passes object references into functions, mutating a mutable argument in place affects the caller's object, while "mutating" an immutable argument inside a function just rebinds a local name and leaves the caller untouched.

Structured elaboration

The distinction is about the object, not the variable: a variable is just a name bound to an object, and reassigning a name (x = x + 1) never mutates anything, it just points the name at a new (or cached) object. Mutation means calling a method or operation that changes the object's internal state while its identity (id()) stays the same, such as list.append, dict.__setitem__, or set.add.

Function-argument consequence:

• Pass a mutable object into a function and mutate it in place inside the function (a_list.append(1)), and the caller sees the change, because the parameter name inside the function is bound to the same object the caller holds.
• Pass an immutable object and try to "modify" it (a_str += "x"), and inside the function you are actually creating a brand-new string and rebinding the local parameter name to it; the caller's original string is untouched, because rebinding a local name never reaches back to the caller's variable.

Hashability follows directly from mutability: a value used as a dict key or set member must have a stable hash for as long as it lives in that structure, because the hash table uses the hash to pick a storage slot. A list can mutate after being inserted, which would silently invalidate its slot, so Python refuses to hash it at all (TypeError: unhashable type: 'list'). A tuple is hashable exactly when everything inside it is hashable (a tuple of ints and strings is fine as a key; a tuple containing a list is not).

Copying semantics matter here too: list.copy(), dict.copy(), and copy.copy() all do a shallow copy, meaning the new container is a distinct object but its elements are the same nested objects as the original. If an element is itself mutable, mutating that nested element through either the original or the copy is visible from both, because they share the same nested object. copy.deepcopy() recursively copies everything, giving full independence, at the cost of more time and memory.

Worked example

def f(a_list):     # mutable argument
    a_list.append(1)

def g(a_str):       # immutable argument
    a_str += "x"

L = []
f(L)
print(L)             # [1]      caller's list was mutated in place

s = "hi"
g(s)
print(s)             # hi       caller's string is untouched

A tuple's immutability is shallow, which is a common surprise: the tuple itself cannot be reassigned to hold a different object, but if one of its elements is a mutable object, that element can still be mutated in place.

t1 = (1, [2, 3])
t2 = t1
t2[1].append(4)
print(t1)            # (1, [2, 3, 4])   the tuple's identity didn't change, but its list element did

Shallow copy versus deep copy, showing the shared-nested-object bug:

import copy

original = {"tags": ["a", "b"]}
shallow = original.copy()
shallow["tags"].append("c")
print(original)      # {'tags': ['a', 'b', 'c']}   mutation through shallow leaked back

deep = copy.deepcopy(original)
deep["tags"].append("d")
print(original)      # {'tags': ['a', 'b', 'c']}   unaffected by the deep copy's mutation
print(deep)           # {'tags': ['a', 'b', 'c', 'd']}

Trade-offs & pitfalls

• The classic real-world bug this causes: passing a shared mutable object (a config dict, a buffer list) into several pipeline stages that each assume they own it, and one stage's in-place mutation silently corrupts what the next stage sees. Prefer passing immutable snapshots (a tuple, a frozenset, or a dataclass(frozen=True)) between stages that should not affect each other.
• Shallow-copy bugs are common precisely because .copy() looks like it should fully isolate the copy; it only isolates the top level. Reach for copy.deepcopy() when the contents are nested and mutable, but remember it costs real time and memory proportional to the whole structure, so don't reach for it by default on large nested data.
• Immutable types are the safer default for anything shared across threads or processes, since there is no in-place state to race on; this is part of why frameworks encourage immutable message/record types for concurrent code.
• A common wrong instinct: assuming tuple is "just a faster, read-only list" in every sense. It is read-only at the top level only; a tuple containing mutable elements offers no deep immutability guarantee.

Approach

Since the elements are guaranteed hashable (hashable means it can be used as a dict key or set member: numbers, strings, and tuples made entirely of hashable values all qualify; lists, dicts, and sets themselves do not, since they can change after creation), a set gives $O(1)$ average membership checks, so the whole pass stays linear: walk the input once, keep a set of everything already seen, and only append an item to the output the first time it appears.

Code (Python 3.12)

def unique_preserve_order(seq):
    """Return items from seq in first-seen order with duplicates removed.
    Assumes every element of seq is hashable.
    """
    seen = set()
    out = []
    for item in seq:
        if item not in seen:
            seen.add(item)
            out.append(item)
    return out

print(unique_preserve_order([3, 1, 3, 2, 1, 4]))
# [3, 1, 2, 4]
print(unique_preserve_order(["b", "a", "b", "c"]))
# ['b', 'a', 'c']
print(unique_preserve_order([]))
# []

The identical logic also dedupes an already-sorted input (a "unique_sorted" ask is the same function; sortedness only changes what the output looks like, not the algorithm) and works the same way on any hashable payload, such as deployment IDs:

deployment_ids = ["d-9", "d-3", "d-3", "d-11", "d-9", "d-2"]
print(unique_preserve_order(deployment_ids))
# ['d-9', 'd-3', 'd-11', 'd-2']

Key points

• The set membership check is what keeps this linear; checking membership against the growing output list instead (if item not in out) would silently work but degrade to $O(n^2)$, since list membership is a linear scan.
• Order is preserved because out is only ever appended to, never reordered; the set is used purely for the O(1) "have I seen this" check and never influences output order.

Complexity

Time: $O(n)$ average, one hash-table lookup and possible insert per element. Space: $O(n)$ worst case (all elements unique), for both seen and out.

Edge cases

• Empty input returns [].
• All-duplicate input returns a single-element list.
• Unhashable elements (a list or dict inside seq) raise TypeError from item not in seen / seen.add(item); per the problem statement this function is only contracted to handle hashable elements, so that is expected and correct behavior, not a bug to work around here.

Adapting for input too large to hold a full result list in memory

The seen set still has to track every distinct value you've encountered (there is no way to know something is a duplicate without remembering it), so the real constraint to relax is holding the full result in memory at once, not the full seen-set. Turn the function into a generator that yields as it goes, so a caller can stream results to a sink (a file, a socket, a downstream queue) instead of building out as a list:

def unique_preserve_order_stream(iterable):
    seen = set()
    for item in iterable:
        if item not in seen:
            seen.add(item)
            yield item

print(list(unique_preserve_order_stream([3, 1, 3, 2, 1, 4])))
# [3, 1, 2, 4]

This removes the $O(n)$ out list (memory now scales only with the number of distinct items seen so far, in seen, plus whatever the consumer of the generator chooses to retain), and it lets a downstream writer start consuming the first unique item immediately instead of waiting for the whole input to be processed. If even seen itself cannot fit in memory (billions of distinct values), the exact-uniqueness guarantee has to be traded for an approximate one (a probabilistic structure such as a Bloom filter, which can have false positives), which is a genuinely different problem from what is asked here and worth naming as the next escalation rather than silently downgrading correctness.