Python Programming Questions

Direct answer

with expr as name: calls expr.__enter__() to acquire a resource and binds its return value to name, then guarantees expr.__exit__(exc_type, exc, tb) runs when the block ends, whether it ended normally or via an exception. That guarantee, not the syntax, is why context managers matter: cleanup code (closing a file, releasing a lock, rolling back a transaction) runs even when the code between acquisition and release raises, without the caller having to remember a try/finally.

Structured elaboration

What __enter__ and __exit__ are responsible for

• __enter__(self) acquires the resource and returns whatever the as clause should bind (often self, sometimes the underlying handle).
• __exit__(self, exc_type, exc, tb) always runs on the way out of the block. If the block raised, exc_type/exc/tb describe that exception; if it exited normally, all three are None.
• The return value of __exit__ controls exception suppression: returning a truthy value swallows the exception (it will not propagate further); returning False or None (the default if you don't return anything) lets it propagate. Suppressing exceptions silently is rarely what you want, so __exit__ should return False unless there is a specific, documented reason to swallow a particular exception type.

Why this beats manual try/finally

• A bare try:/finally: works too, but the context manager packages "how to acquire" and "how to release" into one reusable object, so callers cannot forget the finally and the acquire/release pairing is enforced by the protocol rather than by convention.

Worked example

A minimal custom context manager for a resource that must always be released, even if the block raises:

class ManagedResource:
    '''Stand-in for any resource that must be released: a file, socket,
    lock, or connection. __enter__ acquires it, __exit__ always releases it.
    '''
    def __init__(self, name):
        self.name = name
        self.closed = False

    def __enter__(self):
        print(f"acquired {self.name}")
        return self

    def __exit__(self, exc_type, exc, tb):
        self.closed = True
        print(f"released {self.name} (exc_type={exc_type})")
        return False  # never suppress the exception

r = ManagedResource("conn-1")
try:
    with r as res:
        print("using", res.name)
        raise ValueError("boom")
except ValueError as e:
    print("caught:", e)
print("closed?", r.closed)

Running this prints, in order:

acquired conn-1
using conn-1
released conn-1 (exc_type=<class 'ValueError'>)
caught: boom
closed? True

__exit__ runs and sets closed = True before the exception is allowed to propagate out of the with block; the caller's try/except then catches it exactly as if the with statement were not there at all, because __exit__ returned False.

Trade-offs & pitfalls

• Returning a truthy value from __exit__ to suppress an exception is occasionally the right call (the standard library's contextlib.suppress does exactly this), but doing it without saying so in the class's own name or docstring is a common source of silently-swallowed bugs: an exception vanishes with no trace of why.
• For a simple case like this, contextlib.contextmanager (a generator wrapped with a try/finally) is usually less boilerplate than a full class with __enter__/__exit__; reach for the class form when the object needs to expose state or methods beyond the with block itself (as ManagedResource does here by keeping closed inspectable afterward).
• A context manager only guarantees cleanup for the resource it explicitly manages; if __enter__ itself partially acquires state before raising, __exit__ is never called (the with statement only calls __exit__ once __enter__ has returned successfully), so __enter__ needs to clean up after itself on its own failure paths.

Direct answer

list.append(x) is $O(1)$ amortized; list.pop() (no argument, removing the last element) is $O(1)$; list.pop(0) and list.insert(0, x) are both $O(n)$, because removing or inserting at the front requires shifting every remaining element by one position; list.index(x) is $O(n)$, a linear scan from the start looking for a match. The two snippets below each hide an $O(n)$ operation inside a loop that runs $n$ times, so each function is $O(n^2)$ overall, not the $O(n)$ it might look like at a glance.

Structured elaboration

Why each complexity is what it is: Python's list is a dynamic array, backed by a contiguous block of memory holding references to the elements, plus some spare, over-allocated capacity at the end.

• append(x): writes into the spare capacity at the end and bumps a length counter; no shifting needed. When the spare capacity runs out, CPython reallocates a larger block (roughly 1.125x growth) and copies every existing element once, but this only happens occasionally, so the cost of those occasional $O(n)$ copies, averaged (amortized) over all the append calls, works out to $O(1)$ per call.
• pop(): removes the last element and decrements the length counter; nothing else moves, so it's a true $O(1)$, not just amortized.
• pop(0): removes the first element, then every one of the remaining $n-1$ elements has to shift left by one slot to close the gap, which is $O(n)$.
• insert(0, x): the mirror image; every existing element has to shift right by one slot to make room at the front, before x is written into the now-empty first slot, which is $O(n)$.
• index(x): there is no auxiliary structure telling you where a value lives, so the only way to find it is to check elements one at a time from the front until a match is found (or the end is reached), which is $O(n)$ in the worst case (element near the end, or absent).

Reading the two snippets:

def build_prefix_list(n):
    result = []
    for i in range(n):
        result.insert(0, i)     # O(n) shift, called n times
    return result

def contains_any(items, targets):
    hits = []
    for t in targets:
        if t in items:          # O(len(items)) scan, called len(targets) times
            hits.append(t)
    return hits

build_prefix_list calls insert(0, i) inside a loop that runs $n$ times. Each call's cost grows with however many elements are already in the list (0, then 1, then 2, ..., up to $n-1$), so the total work is $0 + 1 + 2 + \cdots + (n-1) = O(n^2)$, not $O(n)$.

contains_any does t in items inside a loop over targets; in on a list is a linear scan, $O(\text{len(items)})$ per check. If len(items) is roughly $n$ and len(targets) is also roughly $n$, the total work is $O(n) \cdot O(n) = O(n^2)$.

Worked example

print(build_prefix_list(5))                       # [4, 3, 2, 1, 0]
print(contains_any([1, 2, 3, 4, 5], [3, 9, 5]))     # [3, 5]

Fixed versions, same output, each $O(n)$ overall:

def build_prefix_list_fixed(n):
    result = list(range(n))   # build in natural order, O(n) total
    result.reverse()           # O(n) once, not O(n) per element inserted
    return result

def contains_any_fixed(items, targets):
    item_set = set(items)      # O(len(items)) once, upfront
    return [t for t in targets if t in item_set]   # O(1) average per membership check

print(build_prefix_list_fixed(5))                        # [4, 3, 2, 1, 0]
print(contains_any_fixed([1, 2, 3, 4, 5], [3, 9, 5]))      # [3, 5]

build_prefix_list_fixed builds the list in its natural (ascending) order with list(range(n)), which is $O(n)$, then reverses the whole list once with .reverse(), itself $O(n)$; total $O(n)$ instead of $O(n^2)$. contains_any_fixed pays the cost of building a set from items once ($O(n)$), then each membership check against that set is $O(1)$ average, for $O(n)$ total instead of $O(n^2)$.

Trade-offs & pitfalls

• The fix for "repeated insert(0, ...)" is almost always "build forwards, then reverse once" or, if you need frequent insertion and removal from both ends, switch the data structure entirely to collections.deque, which supports $O(1)$ append and pop from either end (at the cost of $O(n)$ random-access indexing, which list gives you for free).
• The fix for "repeated in against a list inside a loop" is almost always "build a set (or dict) once, outside the loop, and check membership against that instead." This is one of the single most common accidental-$O(n^2)$ patterns in data-processing code, and it is easy to miss because each individual line looks innocent.
• list.index(x) has the identical shape of risk: calling it repeatedly inside a loop over the same list is $O(n^2)$; if you need repeated lookups by value, build a dict mapping value to index once, upfront.
• Amortized $O(1)$ for append is a statement about the average cost across many calls, not a guarantee about any single call; an individual append that triggers a resize does real $O(n)$ work at that moment. This rarely matters in practice but is worth knowing when reasoning about worst-case latency for a single operation rather than total throughput.

Approach

Back the cache with a plain dict: keys are a canonical (args, sorted(kwargs.items())) tuple, values are (result, expires_at) pairs, and expires_at is measured with time.monotonic() rather than time.time() so the TTL is not affected by the system clock being adjusted (NTP sync, manual clock changes) mid-run. Eviction happens lazily: every access checks whether its own entry is stale and drops it if so, which means a key that is never looked up again just sits in memory until the janitor (below) or a fresh call for that key clears it. Adding thread-safety is a single threading.Lock guarding every read and write of the dict, since dict operations are not safe to interleave with a delete.

Code (Python 3.12)

import time
import threading
from functools import wraps

def ttl_cache(ttl_seconds):
    '''Decorator factory: cache each call's result for ttl_seconds.
    Thread-safe: one lock guards every read and write of the cache.
    '''
    def decorator(func):
        store = {}          # key -> (value, expires_at)
        lock = threading.Lock()

        @wraps(func)
        def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items())))
            now = time.monotonic()
            with lock:
                cached = store.get(key)
                if cached is not None:
                    value, expires_at = cached
                    if expires_at > now:
                        return value
                    del store[key]          # lazy eviction: stale, drop it
                value = func(*args, **kwargs)
                store[key] = (value, now + ttl_seconds)
                return value

        def cache_clear():
            with lock:
                store.clear()

        wrapper.cache_clear = cache_clear
        return wrapper
    return decorator

calls = {"n": 0}

@ttl_cache(ttl_seconds=0.05)
def slow_square(x):
    calls["n"] += 1
    return x * x

print(slow_square(4))   # 16
print(slow_square(4))   # 16, served from cache
print(calls["n"])       # 1
time.sleep(0.06)
print(slow_square(4))   # 16, recomputed: previous entry had expired
print(calls["n"])       # 2

Key points

• A dict gives average $O(1)$ lookup by (args, kwargs), which is the right structure whenever the key space is a finite, hashable set of call signatures; if an argument is unhashable (a list, a dict), building the key itself raises TypeError before the cache is even consulted.
• Lazy eviction (check-and-drop on access) costs nothing extra for keys that keep getting called, but a key that is called once and never again just occupies memory forever, since nothing ever revisits it to notice it went stale. A proactive sweep fixes that: a daemon=True background thread that wakes up on an interval, takes the lock, and removes every entry whose expires_at has passed, independent of whether anyone accesses it. The two are complementary, not either/or: lazy handles the common case cheaply, proactive bounds worst-case memory for cold keys.
• Thread-safety changes exactly one thing structurally: every dict read-then-maybe-write sequence in wrapper has to happen atomically with respect to other threads, or two threads can both see a miss, both call func, and one result silently overwrites the other (wasted work, not corruption, since both results are equally valid, but still wrong for a function with side effects). A single lock around the whole "check, maybe compute, store" block is sufficient here because the cached function's own body is not being timed for lock duration; if func itself is slow, holding the lock across the call to func serializes unrelated cache misses on different keys too, which is the real cost of this simple approach: it trades throughput under contention for a small, easy-to-reason-about critical section. A production version would narrow the lock to just the dict operations and use a per-key lock (or an if key not in store: store[key] = SENTINEL double-checked pattern) to let different keys compute concurrently.

Complexity

Per call: $O(1)$ average for the dict lookup/insert (hashing the key), plus whatever func itself costs on a miss. Space: $O(u)$ where $u$ is the number of distinct, not-yet-expired argument combinations currently cached.

Edge cases

• Unhashable arguments raise TypeError when the key tuple is built, before func ever runs; this is the same failure mode as any dict-keyed cache and is not TTL-specific.
• A function with side effects (writes to a file, mutates a global) should generally not be cached at all, since a cache hit silently skips the side effect on the second call.
• Two related, absorbed variants of this same design:
  • "Time the call and cache it": extend the stored tuple to (result, expires_at, duration), timing func with time.perf_counter() around the call on a miss, so callers can inspect how expensive an entry was to produce (useful for deciding TTL length empirically) without changing the cache's core structure.
  • "Cache to disk via pickle": persisting store across process restarts with pickle.dump/pickle.load trades in-memory-only simplicity for durability, at the cost of needing every cached value (and every key, since tuples of primitives pickle fine but arbitrary objects may not) to be picklable, plus a decision about whether a pickled entry's expires_at should still be honored after the process was down for a while (it should: monotonic time does not survive a restart, so persisted entries need to be re-validated against wall-clock-derived expiry, or simply treated as expired on load).

Direct answer

A tuple can be a dict key or set member because it is immutable and, as long as everything inside it is itself immutable, hashable; a list cannot, because lists are mutable and Python refuses to hash mutable containers. The invariant that makes any object safely usable as a hash-table key is: if two objects compare equal (a == b), they must produce the same hash (hash(a) == hash(b)); the reverse is not required (unequal objects may share a hash, that is just a collision to be resolved). Hashability in CPython means "defines __hash__", and that hash value must never change for the lifetime of the object while it is a key or set member, which is why mutable types either omit __hash__ or, for user classes, have it silently disabled the moment you define __eq__ without also defining __hash__.

Structured elaboration

Why mutability breaks hashing. A dict finds a key by computing hash(key) once at insertion time and using it to pick a slot. If the key's value (and therefore its hash) could change after insertion, the entry would sit in the wrong slot for its new hash, and a later lookup with the same object would compute a different hash and look in the wrong place, silently failing to find it, or corrupting the table depending on how the collision chain unwound. Python avoids this whole class of bug by making list, dict, and set unhashable, and making tuple hashable only when its contents are hashable (a tuple is a fixed-size, ordered container, but it is only as hashable as what's inside it).

The __hash__/__eq__ contract. For a type to be safely usable as a dict key or set member:

1. a == b implies hash(a) == hash(b). This is the invariant that actually matters: if you break it, two "equal" objects can land in different hash-table slots, and a lookup by one will fail to find the entry stored under the other, even though the container logically already has that key.
2. hash(a) must be stable for the object's lifetime as a key (an object should not become externally observably different, from the hash table's perspective, after insertion).
3. Python's default behavior for user classes reflects this: if you define __eq__ without defining __hash__, Python sets __hash__ to None for that class (making instances unhashable), because the compiler cannot verify your custom __eq__ still satisfies rule 1 against the default identity-based hash.

Where this shows up with tuples.

• (1, 2) is hashable: both elements are immutable ints.
• (1, [2, 3]) is NOT hashable: it contains a list, so hash() raises TypeError, even though the outer tuple itself cannot be mutated.
• frozenset is the immutable, hashable counterpart to set, useful as a key when you need "a set of things" as the key itself.
• collections.namedtuple (and typing.NamedTuple) are still plain tuples under the hood, so they remain hashable under the same rule and add named-field readability with no extra runtime cost over a positional tuple.

Worked example

t = (1, 2)
print(hash(t))
# some integer, e.g. -3550055125485641917

try:
    hash([1, 2])
except TypeError as e:
    print("TypeError:", e)
# TypeError: unhashable type: 'list'

try:
    hash((1, [2, 3]))   # tuple containing a list is still unhashable
except TypeError as e:
    print("TypeError:", e)
# TypeError: unhashable type: 'list'

Defining __eq__ without __hash__ makes a class unhashable, demonstrating the contract directly:

class Point:
    def __init__(self, x):
        self.x = x
    def __eq__(self, other):
        return self.x == other.x
    # no __hash__ defined -> Python sets __hash__ = None here

try:
    hash(Point(1))
except TypeError as e:
    print("TypeError:", e)
# TypeError: unhashable type: 'Point'

Using a tuple as a composite grouping key (a common real use):

from collections import defaultdict

rows = [
    {"user_id": 1, "date": "2025-11-01", "value": 10},
    {"user_id": 1, "date": "2025-11-01", "value": 5},
    {"user_id": 2, "date": "2025-11-02", "value": 7},
]

totals = defaultdict(int)
for r in rows:
    key = (r["user_id"], r["date"])   # tuple: immutable and hashable
    totals[key] += r["value"]
# totals == {(1, '2025-11-01'): 15, (2, '2025-11-02'): 7}

Trade-offs & pitfalls

• A tuple is only hashable if all of its elements are; a tuple of lists, dicts, or sets is not, and you will find this out at runtime with a TypeError, not at write time.
• If you define __eq__ on a class and want instances to remain usable as dict keys or set members, you must also define a compatible __hash__ (or explicitly reuse the default with __hash__ = object.__hash__ if identity equality is acceptable); forgetting this is a common way custom value objects silently become unhashable.
• Violating the "equal implies equal hash" invariant (defining __eq__ and a __hash__ that disagrees with it) does not raise an error; it produces "impossible" bugs where key in some_dict returns False for a key that logically should match, because the two objects hashed into different slots.
• namedtuple and frozenset are the two common upgrades once a plain tuple's positional-only access, or a plain tuple's ordering, respectively, stops being convenient, both without giving up hashability.

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.