Python Programming Questions

Project layout

• mypackage/
  • src/mypackage/init.py # package implementation
  • src/mypackage/module.py
  • tests/ # unit tests
    • test_module.py
  • pyproject.toml / setup.cfg

Where to place tests
Keep tests outside package source (e.g., tests/ at repo root). This prevents accidental import of tests and mirrors CI test discovery.

init.py & public API
Use init.py to expose a stable public API with explicit imports and all. Keep implementation modules internal (e.g., mypackage._internal).

A concrete worked example of what that __init__.py and its module actually contain:

# src/mypackage/module.py
def process(data):
    return [x * 2 for x in data]

def _internal_helper():   # leading underscore: not part of the public API
    pass

# src/mypackage/__init__.py
from .module import process

__all__ = ["process"]

A caller then does from mypackage import process, process is the stable public name, while mypackage.module._internal_helper stays a private implementation detail nobody outside the package should import directly; __all__ also controls exactly what from mypackage import * pulls in.

Testing in CI

• Use pytest, run in CI (GitHub Actions) with matrix for Python versions. Run mypy (checks your type hints without actually running the code) and unit tests; fail on coverage thresholds.

A minimal CI job doing exactly that:

# .github/workflows/test.yml
jobs:
  test:
    strategy:
      matrix:
        python-version: ["3.11", "3.12"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
      - run: pip install -e .[dev]
      - run: pytest --cov=mypackage --cov-fail-under=80
      - run: mypy src/mypackage

Type stubs & docs

• Add type hints in code; supply .pyi stubs (a .pyi file is a type-hint-only file, holding no real code, for code Python itself cannot read type hints from directly, such as a compiled C extension) for compiled extensions.
• Use Sphinx or MkDocs (both turn your docstrings into a browsable documentation website) for docs; generate API reference from docstrings. Include CI job to build docs.

Which of these matter most for a first working setup versus are polish added later: the src/ layout, an explicit __init__.py public API, and pytest running in CI are the essentials, without them the package is neither testable nor safely importable by other code. mypy, coverage thresholds, .pyi stubs, and generated documentation sites are all valuable, genuinely worth doing, but they are follow-up hardening once the basic package structure and test suite already exist, not blockers to a first working, publishable package.

Best practices: pin dev deps, run linters, and keep public surface minimal and documented.

Solution (Pandas): use pivot_table taking last value for duplicates and use efficient grouping to limit memory.

import pandas as pd

df = pd.DataFrame({
    "user": ["u1", "u1", "u1", "u2", "u2"],
    "metric": ["score", "score", "clicks", "score", "clicks"],
    "value": [10, 15, 3, 20, 7],
})

# pivot using last observation
wide = df.pivot_table(index='user', columns='metric', values='value', aggfunc='last')
# optional: reset index
wide = wide.reset_index()

Worked example, verified with pandas on CPython 3.12, including a genuine duplicate (user, metric) pair so the tie-breaking behavior is visible:

df = pd.DataFrame({
    "user": ["u1", "u1", "u1", "u2", "u2"],
    "metric": ["score", "score", "clicks", "score", "clicks"],
    "value":  [10,     15,      3,       20,     7],
})
print(df)
wide = df.pivot_table(index='user', columns='metric', values='value', aggfunc='last').reset_index()
print(wide)

Input:

  user  metric  value
0   u1   score     10
1   u1   score     15
2   u1  clicks      3
3   u2   score     20
4   u2  clicks      7

Output:

metric user  clicks  score
0        u1       3      15
1        u2       7      20

u1 has two score rows (10 then 15); aggfunc='last' keeps whichever one appears LAST in df's row order, 15, and silently drops 10. This is the actual behavior the question asks about: the row order of the input DataFrame determines which duplicate value survives, so if "last" is meant to be "most recent by time" rather than "however the rows happened to arrive," the DataFrame must be sorted by a timestamp column first, pivot_table itself has no notion of time, only of row order.

Efficient for large data:

• If df is very large, pre-sort so last() works correctly: df.sort_values(['user','metric','timestamp'], inplace=True) then drop_duplicates keeping='last' then pivot.

# memory-friendly: deduplicate then pivot
df = pd.DataFrame({
    "user":      ["u1", "u1", "u1", "u2", "u2"],
    "metric":    ["score", "score", "clicks", "score", "clicks"],
    "value":     [10, 15, 3, 20, 7],
    "timestamp": pd.to_datetime([
        "2024-01-01", "2024-01-03", "2024-01-02", "2024-01-01", "2024-01-03"
    ]),
})
df2 = df.sort_values(['user','metric','timestamp']).drop_duplicates(['user','metric'], keep='last')
wide = df2.pivot(index='user', columns='metric', values='value').reset_index()

This second code block is an ALTERNATIVE to the pivot_table(aggfunc='last') call above for large data, not a required second step after it: pivot_table with aggfunc='last' already handles duplicates correctly on its own, this version exists purely because pre-deduplicating (drop_duplicates) before a plain pivot (which requires already-unique index/column pairs and raises otherwise) can be cheaper at scale than letting pivot_table do the deduplication and aggregation together internally.

Notes: drop_duplicates reduces rows before pivoting to lower memory. Use categorical for 'metric' to reduce memory. For extreme scale, use Dask DataFrame (a library that mirrors the Pandas API but splits data into partitions and runs the same operations, including pivot-style reshaping, across them in parallel) with same operations, once the table genuinely no longer fits on one machine; plain Pandas, as shown above, is enough below that point.

Approach: Use a min-heap (a min-heap is a tree-shaped data structure that always keeps its smallest element accessible at the top/root, retrievable in O(1), and restores that property in O(log n) after every push or pop; that is what lets this algorithm always grab the smallest available head-of-iterable next without rescanning every input from scratch) to pull the smallest next element from heads of each iterable. Support key by pushing (key(val), idx, val, iterator). Works with generators and streams and only holds one item per input plus heap overhead.

Implementation:

import heapq
import itertools

def merge_sorted_iterables(iters, key=None):
    key = (lambda x: x) if key is None else key
    heap = []
    for idx, it in enumerate(map(iter, iters)):
        try:
            val = next(it)
            heapq.heappush(heap, (key(val), idx, val, it))
        except StopIteration:
            continue
    while heap:
        _, idx, val, it = heapq.heappop(heap)
        yield val
        try:
            nxt = next(it)
            heapq.heappush(heap, (key(nxt), idx, nxt, it))
        except StopIteration:
            continue

Worked example, verified on CPython 3.12:

print(list(merge_sorted_iterables([[1, 4, 7], [2, 3, 9]])))
# [1, 2, 3, 4, 7, 9]

Tracing the heap: it starts holding one entry per input, (1, 0, 1, iter1) and (2, 1, 2, iter2), the smallest head of each list. Popping the smaller one yields 1 and immediately pushes that same iterator's next value, 4, back in, so the heap always holds exactly one "candidate next value" per iterable that still has values left. Popping the current minimum of that small set is always correct because every value not yet pulled from a given iterable is guaranteed to be >= the value currently sitting in the heap for it, since each input is already sorted; the full sequence of pops is 1, 2, 3, 4, 7, 9, matching the printed output exactly.

Key points:

• Memory: O(m) where m = number of input iterables (one element per iterator + heap metadata).
• Works with infinite streams and generators.
• Stable per iterator due to idx tie-breaker.

Edge cases: empty iterables, differing element types (ensure key handles them), expensive key (consider caching keys).

Approach: implement LRU with a dict for storage and a doubly-linked list for order; use threading.RLock (a lock that the SAME thread can safely acquire again without deadlocking itself, unlike a plain threading.Lock, in case the cache logic ever needs to re-enter the lock while already holding it) for concurrency. Decorator returns wrapper that locks around lookups and updates, minimizing lock hold time.

Implementation:

import threading
from functools import wraps

def lru_cache(maxsize=128):
    def deco(func):
        cache = {}
        head = tail = None
        lock = threading.RLock()
        class Node:
            __slots__=('key','val','prev','next')
            def __init__(self,k,v):
                self.key=k;self.val=v;self.prev=self.next=None
        def _move_to_front(node):
            nonlocal head, tail
            if node is head:
                return  # already the most-recently-used entry, nothing to do
            # unlink node from wherever it currently sits
            if node.prev:
                node.prev.next = node.next
            if node.next:
                node.next.prev = node.prev
            if node is tail:
                tail = node.prev
            # relink it at the head (the most-recently-used end)
            node.prev = None
            node.next = head
            if head:
                head.prev = node
            head = node
            if tail is None:
                tail = node
        @wraps(func)
        def wrapper(*args, **kwargs):
            nonlocal head, tail   # wrapper reassigns both below on eviction; without
                                   # this declaration LEGB makes them locals instead
            key=(args,tuple(sorted(kwargs.items())))
            with lock:
                node=cache.get(key)
                if node:
                    _move_to_front(node); return node.val
            val=func(*args, **kwargs)
            with lock:
                if key in cache: return cache[key].val
                node=Node(key,val); cache[key]=node; _move_to_front(node)
                if len(cache)>maxsize:
                    # cache is over capacity: evict the true tail, the
                    # least-recently-used entry, from both the linked list
                    # and the dict
                    lru_node = tail
                    tail = lru_node.prev
                    if tail:
                        tail.next = None
                    else:
                        head = None
                    del cache[lru_node.key]
            return val
        return wrapper
    return deco

Worked example: verified on CPython 3.12, calling the decorated function through a full eviction cycle so the policy can actually be watched working, not just taken on faith.

calls = []

@lru_cache(maxsize=2)
def square(n):
    calls.append(n)
    return n * n

print(square(1))   # 1  (miss: cache empty, computed and cached; order (MRU->LRU): [1])
print(square(2))   # 4  (miss: computed and cached; order: [2, 1])
print(square(1))   # 1  (hit: served from cache, 1 moves back to the front; order: [1, 2])
print(square(3))   # 9  (miss: cache was full at {1, 2}; since 1 was just reused, 2 is now
                    # the least-recently-used entry and gets evicted to make room for 3; order: [3, 1])
print(square(2))   # 4  (miss again: 2 was evicted in the previous step, so this recomputes
                    # instead of hitting the cache; order: [2, 3])
print(calls)        # [1, 2, 3, 2]  -- 2 appears twice: proof it was actually evicted and
                     # had to be recomputed, not just a claim

Complexity: O(1) average get/set.

Contention: a single global lock, one shared threading.RLock protecting the whole cache, serializes cache access: only one thread at a time can even check whether something is cached, regardless of which key it wants, which becomes a bottleneck under heavy concurrent traffic. Two ways to reduce that: a read-mostly optimistic check (read the dict for a hit without taking the lock first, since a plain dict read is safe to race on for a snapshot lookup, and only take the lock to confirm the hit and update the linked-list ordering, so the common cache-hit path spends less time holding the lock), or shard locks (split one cache into several smaller caches, each with its own separate lock, and route each key to one shard by hashing it, e.g. shard = hash(key) % num_shards; two threads reading keys that land in different shards no longer contend for the same lock at all, at the cost of maxsize now being enforced per shard rather than globally).

Go vocabulary, translated for a Python reader: a goroutine is Go's lightweight, concurrently-running function, similar in spirit to a Python thread but far cheaper to start and typically used in much greater numbers in real Go code; a channel is a typed, thread-safe queue you send values into and receive values out of, roughly like a queue.Queue shared between Python threads, except the compiler enforces the type of what flows through it; a WaitGroup is a counter that lets the caller block until every worker goroutine has signaled it is done, similar to calling .join() on a list of Python Thread objects; ctx (short for context) carries a shared cancellation signal through the call tree, and ctx.Done() returns a channel that closes the moment that signal fires, so any goroutine can cheaply check "has someone asked everything to stop?" without polling a shared boolean, comparable to checking a Python threading.Event.

Structure

I’d use a bounded worker pool. A feeder sends files with an index into a jobs channel. Each worker validates one file, sends {index, result, err} to a results channel, and watches ctx.Done() so context cancellation, meaning a cooperative stop signal, is fast.

Ordering
A single collector keeps nextIndex and a map of out-of-order results. If result 7 arrives before 6, store it until 6 is ready, then flush in submission order.

Fatal parse error
If a worker sees an unrecoverable parse error, it sends the error and calls cancel(). That stops the feeder, makes workers exit on ctx.Done(), and prevents new work from starting.

Shutdown

• close jobs after feeding stops
• WaitGroup waits for workers
• close results after workers finish
• collector drains until closed or canceled

Worked example: files 1, 2, 3, 4 arrive. If file 3 has a fatal parse error, 1 and 2 can still be emitted, 4 is never started, and the summary reports the error immediately.

Shape of the code (a sketch of the structure, not a full compiled program):

type job struct {
    index int
    path  string
}
type result struct {
    index  int
    output string
    err    error
}

func run(ctx context.Context, paths []string, numWorkers int) []result {
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    jobs := make(chan job)
    results := make(chan result)
    var wg sync.WaitGroup

    for w := 0; w < numWorkers; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := range jobs {
                out, err := validate(j.path)
                if err != nil && isFatal(err) {
                    cancel()
                }
                results <- result{index: j.index, output: out, err: err}
            }
        }()
    }

    go func() {
        for i, p := range paths {
            select {
            case jobs <- job{index: i, path: p}:
            case <-ctx.Done():
                close(jobs)
                return
            }
        }
        close(jobs)
    }()

    go func() {
        wg.Wait()
        close(results)
    }()

    return collectInOrder(results)
}

The jobs and results lines are the channel declarations; the three go func() { ... }() blocks are the goroutine launches, one pool of workers, one feeder, one closer. select { case jobs <- job{...}: ... case <-ctx.Done(): ... } is how the feeder stays responsive to cancellation even while trying to send a job that a worker isn't ready to receive yet, instead of blocking forever on a full channel after a fatal error.

This gives parallelism, ordered output, and fast failure without deadlocks.