Python Programming Questions

Context manager timed_block

import time
import logging

logger = logging.getLogger(__name__)

class timed_block:
    def __init__(self, name=None):
        self.name = name or 'timed_block'
    def __enter__(self):
        self.start = time.perf_counter()
        return self
    def __exit__(self, exc_type, exc, tb):
        elapsed = time.perf_counter() - self.start
        logger.info('%s elapsed: %.6f s', self.name, elapsed)

Python always calls __exit__(self, exc_type, exc, tb) on the way out of a with block, whether or not anything went wrong: if the block raised, exc_type is the exception class, exc is the actual exception instance, and tb is its traceback object; if the block finished normally, all three are simply None. This class does not use them (the timing log happens either way, and returning nothing from __exit__ means None, which is falsy, so any exception from inside the with block still propagates normally afterward), but a context manager that needed to react differently on failure, skipping the log, or logging a different message, would inspect exc_type to tell the two cases apart.

Usage:

with timed_block('heavy_task'):
    do_heavy_work()

Actually running it, with a concrete body and logging configured to print:

import logging, time

logging.basicConfig(level=logging.INFO)

def do_heavy_work():
    time.sleep(0.05)

with timed_block('heavy_task'):
    do_heavy_work()

This prints a line shaped like:

INFO:__main__:heavy_task elapsed: 0.0501xx s

The exact trailing digits vary run to run, since this is real wall-clock time and not something the code can pin down in advance, but the value is always at least 0.05 (the time.sleep(0.05) call blocks for at least that long) and the format itself, name elapsed: X.XXXXXX s, from the '%s elapsed: %.6f s' format string, is exactly what is shown every time.

This measures wall-clock time using perf_counter and logs at INFO. It returns the context object if callers want to inspect timing later.

Approach: stream the file line-by-line using an internal buffer of chunk_size, parse CSV incrementally, yield namedtuples and convert numeric-looking strings to int/float lazily.

Implementation:

import csv
from collections import namedtuple
from io import TextIOBase

def stream_csv(path_or_file, chunk_size=65536):
    f = open(path_or_file, 'r', newline='') if isinstance(path_or_file, str) else path_or_file
    reader = csv.reader(f)
    headers = next(reader)
    Row = namedtuple('Row', [h.replace(' ', '_') for h in headers])
    def conv(s):
        if s == '':
            return None
        try:
            i = int(s)
            return i
        except ValueError:
            try:
                return float(s)
            except ValueError:
                return s
    for raw in reader:
        yield Row(*[conv(field) for field in raw])
    if isinstance(path_or_file, str):
        f.close()

A namedtuple (created above via namedtuple('Row', [...])) is a lightweight, tuple-backed class whose fields you can access by name, row.age, instead of only by position, row[1], which is why it is used here rather than a plain tuple or a dict: it is as compact and fast as a tuple, but each yielded row reads like a normal object with named fields.

Worked example, verified on CPython 3.12:

import io

sample_csv = "name,age,score\nAlice,30,9.5\nBob,,7\n"
rows = list(stream_csv(io.StringIO(sample_csv)))
print(rows)
# [Row(name='Alice', age=30, score=9.5), Row(name='Bob', age=None, score=7)]
print(rows[0].name, rows[1].age)
# Alice None

conv turns '30' into the int 30 and '9.5' into the float 9.5; Bob's empty age field hits the s == '' check first and becomes None rather than being run through int()/float() at all, and rows[0].name shows the named-field access a plain tuple could not offer without remembering which index held which column.

Key points: single-pass, constant memory per-row, converts numbers eagerly per-field. For truly huge fields, use streaming CSV parsers (csv module is already streaming).

Complexity: O(n) time where n is total fields; O(1) extra space per row.

Edge cases: quoted newlines, malformed rows, large field sizes: consider using csv.DictReader and incremental file objects if needed.

Approach

I would stream the file line by line so memory stays bounded, meaning it stays roughly constant no matter how large the file is. For each line, trim whitespace, skip blanks, try to parse the fields, and update counters. Malformed lines go into a small issue count or log so the job can continue.

from dataclasses import dataclass


@dataclass
class Report:
    total: int = 0
    overdue: int = 0
    days_overdue_sum: int = 0
    malformed: int = 0


def process_tickets(path: str) -> Report:
    report = Report()

    with open(path, 'r', encoding='utf-8') as f:
        for line_no, line in enumerate(f, start=1):
            line = line.strip()
            if not line:
                continue

            parts = line.split(',')
            if len(parts) != 3:
                report.malformed += 1
                continue

            try:
                status = parts[1].strip().lower()
                days = int(parts[2].strip())
            except ValueError:
                report.malformed += 1
                continue

            if not parts[0].strip():
                report.malformed += 1
                continue

            report.total += 1
            if status == 'overdue':
                report.overdue += 1
                report.days_overdue_sum += days

    return report

Worked example: with A1,overdue,3, a blank line, and B2,open,0, the report counts 2 valid records, 1 overdue ticket, and 3 total overdue days.

Tradeoffs

• Fast and memory-safe for huge files
• One pass only, so you should add any needed counters up front
• If operators need examples of bad lines, log the first few rather than storing every bad line

Complexity: O(n) time and O(1) memory for the counters.

Possible Python/library causes

• dict ordering: dict insertion order IS deterministic in CPython 3.7+ (this is a language guarantee, not an implementation quirk), but the actual non-determinism seen in production comes from something upstream feeding the dict in a different order between runs, for example a filesystem glob() returning filenames in a different order, or multiple workers finishing and inserting their results in whatever order they happen to complete, not the dict itself misbehaving.
• set iteration: unordered, causes non-deterministic traversal.
• floating-point reductions (sum/mean) non-associative ordering leads to tiny differences.
• parallelism: thread/process scheduling, race conditions, non-deterministic map ordering.
• use of RNG without fixed seeds (random, numpy.random, torch).
• reliance on OS-level nondeterminism (filesystem glob order, multiprocessing.Queue timing).

Two of these causes, made concrete, verified on CPython 3.12:

Non-associative floating-point reduction, a naive accumulator loop genuinely disagreeing with itself depending on order (built-in sum() on recent CPython uses a more accurate internal algorithm and happens to avoid this specific example, which is exactly why the naive loop below, closer to what a hand-rolled or per-worker partial-sum reduction actually does, is the more honest demonstration):

def manual_sum(values):
    total = 0.0
    for v in values:
        total += v
    return total

vals = [0.1, 0.2, 0.3]
print(manual_sum(vals))
# 0.6000000000000001
print(manual_sum(list(reversed(vals))))
# 0.6

import math
print(math.fsum(vals) == math.fsum(list(reversed(vals))))
# True

The same three floats, summed in two different orders by a plain accumulator, produce two DIFFERENT bit patterns, 0.6000000000000001 versus exactly 0.6, purely from the order of addition (this is exactly what happens when parallel workers each compute a partial sum and the main process combines them in whatever order they finish); math.fsum (a compensated-summation algorithm that tracks and corrects rounding error as it goes) gives the identical, order-independent result either way.

Unseeded RNG producing different output on repeated runs, and the fix:

import random

random.seed(42)
print(random.random())
# 0.6394267984578837
random.seed(42)
print(random.random())
# 0.6394267984578837 -- identical, since the seed was reset

Calling random.seed(42) again before the second random.random() call reproduces the exact same float; omitting the seed (or seeding from the current time, which is random's default) would make this genuinely different on every run, which is exactly the failure mode the fix below addresses.

Concrete fixes

• Sort keys/lists before iteration (e.g., sorted(dict.items())).
• Use deterministic data structures: collections.OrderedDict where explicit order required.
• Fix RNG seeds at pipeline start for all libs (random.seed, np.random.default_rng with seed passed).
• Use stable reduction algorithms: math.fsum or pairwise summation (numpy.sum with dtype and stable algorithm libs).
• Avoid unordered set for grouping; use sorted containers or deterministic grouping keys.
• Control parallelism: fix number of workers, use deterministic partitioning, or run single-threaded for critical steps.
• Pin library versions and document environment (requirements, Docker).

These steps produce reproducible outputs suitable for production ETL runs and debugging.

1) Multiprocessing (multiprocessing module)

• Benefit: sidesteps GIL (Global Interpreter Lock: a lock inside CPython that only lets one thread execute Python bytecode at a time, which is why adding more threads alone does not speed up pure-Python CPU-bound work) by using multiple processes; good for CPU-bound tasks that can be partitioned.
• Limitations: IPC and data serialization overhead, higher memory usage per process.
• When: embarrassingly parallel workloads (map-style), batch processing across cores.

2) C-accelerated libraries (NumPy / vectorization)

• Benefit: move heavy numeric loops into C (SIMD, Single Instruction Multiple Data, a CPU feature applying one operation to many values in one step; contiguous memory), drastically faster for array math.
• Limitations: requires expressing work in array form; not helpful for complex control flow.
• When: numerical computations over arrays, matrix ops.

3) Just-in-time / ahead-of-time compilation (Numba or Cython)

• Numba: JIT-compiles (just-in-time compiles: instead of compiling ahead of time before the program ever runs, the function is compiled straight to machine code the first time it is actually called) Python functions to machine code with little code changes; excellent for loops on numeric data.
  • Benefit: low development cost, big speedups.
  • Limitation: not all Python features supported; first-call compilation overhead.
• Cython: compile Python to C for maximum control and speed; can produce greatest gains but requires typing and build step.
  • When: algorithmic hotspots that remain after vectorization or need fine-grained control.

A concrete contrast (real values, not timing, since exact speed depends on hardware and is not something to hardcode here):

import numpy as np

def squares_loop(n):
    return [i * i for i in range(n)]

def squares_vectorized(n):
    return np.arange(n) ** 2

print(squares_loop(5))
# [0, 1, 4, 9, 16]
print(squares_vectorized(5))
# [ 0  1  4  9 16]

Both produce the identical five values; squares_loop pays Python's per-element interpreter overhead five separate times, while squares_vectorized dispatches once into a single compiled C loop, the same mechanism that makes strategy 2 (vectorization) and strategy 3 (Numba/Cython compilation) both faster than a plain interpreted loop: removing per-element interpreter overhead, not changing the underlying arithmetic.

Choose based on workload: use vectorization first, then Numba for loop-heavy numeric code, and multiprocessing when parallelism across cores is needed and data can be partitioned.