Python Programming Questions

Approach

I’d normalize by lowercasing, removing punctuation, and collapsing repeated spaces. Then I’d use a set of normalized keys to keep only the first occurrence. A set is a data structure that gives fast membership checks, so this stays simple and efficient for a few thousand strings.

import re
from typing import List


def normalize_title(title: str) -> str:
    cleaned = re.sub(r'[\W_]+', ' ', title.lower())
    return ' '.join(cleaned.split())


def dedupe_titles(titles: List[str]) -> List[str]:
    seen = set()
    result = []

    for title in titles:
        norm = normalize_title(title)
        if norm in seen:
            continue
        seen.add(norm)
        result.append(norm)

    return result


titles = ['  Weekly Update!', 'weekly update', 'Budget Review', 'Budget review  ']
print(dedupe_titles(titles))

The regex pattern r'[\W_]+' is what actually strips punctuation: \W (capital W) matches any character that is NOT a letter, digit, or underscore, so it catches punctuation, symbols, and stray whitespace runs in one go; adding _ to the character class folds underscores into that same "replace with a space" rule too, since \W alone would NOT match an underscore (underscores count as word characters). The + means one-or-more, so any run of these unwanted characters collapses to a single space rather than leaving multiple spaces behind, and ' '.join(cleaned.split()) then trims leading/trailing spaces and collapses any remaining internal runs down to exactly one space each.

Example

Input: [' Weekly Update!', 'weekly update', 'Budget Review', 'Budget review ']

Output:

['weekly update', 'budget review']

That is the literal stdout of print(dedupe_titles(titles)) from the code block above.

Why this works: the first weekly update is kept, and the later duplicate is skipped because it normalizes to the same key.

Complexity: O(n * m) time, where m is string length, and O(n) extra space for the set.

If the team wants to keep original casing, I would store the first original string in result instead of the normalized version.

Plan: do out-of-core aggregation (out-of-core means: process data too big to fit in memory by working on it piece by piece, rather than loading the whole 50 GB file at once) by streaming CSV in chunks or using Dask. Maintain per-key aggregates (sum, count) and compute mean = sum/count.

Chunked Pandas approach:

from collections import defaultdict
import pandas as pd
sums = defaultdict(float)
counts = defaultdict(int)
for chunk in pd.read_csv('big.csv', chunksize=10_000_00):
    grp = chunk.groupby('key')['value'].agg(['sum','count'])
    for k, row in grp.iterrows():
        sums[k] += row['sum']; counts[k] += row['count']
# finalize
result = {k: (sums[k], counts[k], sums[k]/counts[k]) for k in sums}

Worked example (traced on a tiny 6-row file so the shapes are concrete)
Suppose big.csv held just:

key,value
a,10
b,20
a,30
c,5
b,15
a,50

Running the chunked-Pandas code above against this file (even split across multiple small chunks, since each chunk's partial sum/count is added into the running totals rather than overwriting them) produces:

print(result)
# {'a': (90.0, 3, 30.0), 'b': (35.0, 2, 17.5), 'c': (5.0, 1, 5.0)}

Each tuple is (sum, count, mean): key a appears three times with values 10, 30, 50 (sum 90, mean 30.0), key b twice with 20 and 15 (sum 35, mean 17.5), and key c once with 5 (sum 5, mean 5.0). Because the loop only ever adds each chunk's partial sum/count into the running totals, splitting this same 6-row file into 2 chunks of 3 rows each, or processing it as one chunk, produces the identical result dict; only how many times the loop body runs changes, not the answer.

Dask approach: Dask is a library that mirrors the Pandas API (dask.dataframe looks and behaves like pandas.DataFrame) but splits the data into partitions and runs the same operations across them in parallel, so the .groupby().agg() code you already know scales past what fits in one machine's RAM. Use dask.dataframe.read_csv and ddf.groupby('key').agg({'value':['sum','count']}).compute(): Dask handles partitioning and parallelism.

Trade-offs:

• Pure chunking: simple, low memory, single-threaded unless parallelized; needs dictionary sized by number of unique keys.
• Dask: parallel, scales across cores/machines, but adds scheduler overhead and cluster configuration.

Notes: If unique keys are huge, consider external grouping: this is a rare fallback, needed only once the number of distinct keys is itself too large for the sums/counts dictionaries to fit in memory (a separate problem from the original 50 GB of rows, which the chunking above already handles). Two concrete versions of it: sorting by key on disk (writing all rows out sorted so equal keys become adjacent, letting you aggregate in one streaming pass with no dict at all), or using a local key-hash partitioning to spill to disk (hashing each key to decide which of several on-disk partition files it belongs to, so each partition file can later be aggregated on its own, fully in memory, one at a time).

Requirements & idea: Provide a lazy, composable API that records string ops and applies them in a single pass to avoid temporaries. Offer a lightweight proxy object wrapping Series with an operation pipeline executed in-place or chunked.

API sketch, with a working _apply_pipeline (the earlier sketch left this function unimplemented; here it actually runs each recorded op vectorized, once, over the whole chunk, rather than materializing an intermediate Series between every step):

class StrChain:
    def __init__(self, series):
        self.series = series
        self.ops = []
    def lower(self):
        self.ops.append(('lower', None)); return self
    def replace(self, pat, repl):
        self.ops.append(('replace', (pat, repl))); return self
    def strip(self):
        self.ops.append(('strip', None)); return self
    def compute(self, chunk_size=10_000):
        # apply ops chunk-wise to avoid temporaries
        return _apply_pipeline(self.series, self.ops, chunk_size)


def _apply_pipeline(series, ops, chunk_size):
    parts = []
    for start in range(0, len(series), chunk_size):
        chunk = series.iloc[start:start + chunk_size]
        for op_name, arg in ops:
            if op_name == 'lower':
                chunk = chunk.str.lower()
            elif op_name == 'replace':
                pat, repl = arg
                chunk = chunk.str.replace(pat, repl, regex=False)
            elif op_name == 'strip':
                chunk = chunk.str.strip()
        parts.append(chunk)
    return type(series)(pd.concat(parts)) if parts else series

_apply_pipeline walks the recorded op list once per chunk (default: the whole Series in one chunk, for anything that fits in memory), calling the real vectorized Series.str method for each recorded op in sequence; "chunk-wise" here means each chunk only ever holds one intermediate Series at a time (reassigned to chunk on each step) rather than every stage's output existing simultaneously the way s.str.lower().str.replace(...).str.strip() chained directly would briefly do.

Worked example, verified with pandas on CPython 3.12:

import pandas as pd

s = pd.Series(['  Hello World  ', '  FOO-BAR  ', '  Already lower  '])
result = StrChain(s).lower().replace('-', ' ').strip().compute()
print(list(result))

Output:

['hello world', 'foo bar', 'already lower']

Each string is lowercased, has - replaced with a space, and is stripped of surrounding whitespace, in that recorded order, confirming the chain actually runs end to end and produces the same result s.str.lower().str.replace('-', ' ', regex=False).str.strip() would, just without materializing three separate full-Series temporaries to get there.

Implementation strategies:

• Represent ops as vectorized functions (use Series.str methods or numpy.char).
• Apply pipeline per chunk: read chunk, apply all ops in sequence in-place (reuse buffers), write out to result array or new Series with preallocated dtype.
• For unicode/regex heavy ops, compile regex ahead.
• Use numba (compiles a plain Python function to machine code the first time it runs) or cython (compiles Python-like code all the way down to C, with explicit type declarations, for the most control and the largest potential speedup) for the hot path once profiling shows plain vectorized Series.str ops are the actual bottleneck; for most chains, the vectorized ops above are already enough and neither is needed by default.

Minimize allocations:

• Reuse a single buffer per chunk; preallocate numpy object/bytes arrays when possible.
• Fuse operations into one pass (e.g., lower+strip -> single routine, implemented as one numpy.char or Cython function operating character-by-character instead of two separate full passes over the data).

Trade-offs: chunking adds overhead but limits peak memory; fusing ops increases implementation complexity.

Reproducible environment for data-science project:

• Virtual environment: use venv or conda env per project.
• Dependency pinning: record exact packages in requirements.txt via pip freeze or use Pipfile.lock/poetry.lock (a lock file records the exact version, and a hash, of every package actually installed, including indirect dependencies pulled in by your direct ones, so a second install reproduces the identical environment instead of picking up newer, still-technically-compatible versions) for deterministic installs.
• Packaging: provide pyproject.toml for build metadata; publish internal wheel (a wheel is a pre-built, ready-to-install package file, so pip install just unpacks it with no compiler needed) if needed.
• Reproducible runs: fix random seeds for numpy/torch/tf; record environment metadata (python version, OS, package hashes).
• Data & compute: version datasets (DVC, Data Version Control, a tool that tracks large dataset files the way git tracks code, or checksums), containerize with Docker (packages the OS plus the application together so it runs identically on any machine) for identical runtime, and provide examples to reproduce results.

A minimal concrete example of the actual commands and files, tying the bullet points above to something runnable:

python -m venv .venv
source .venv/bin/activate
pip install pandas==2.2.2 numpy==1.26.4
pip freeze > requirements.txt

requirements.txt now contains exact, pinned lines like:

numpy==1.26.4
pandas==2.2.2

A second machine reproduces the identical environment with pip install -r requirements.txt, since every version is already pinned rather than left to "whatever the latest compatible release is today."

Which of these matter most versus are nice-to-have polish: a virtual environment plus a pinned requirements.txt (or an equivalent lock file from Poetry/Pipenv) is the non-negotiable baseline, without it, nothing else on this list is reproducible. Fixed random seeds and recorded environment metadata are the next tier, cheap to add and directly explain "why did I get a different number this time." DVC and full Docker containerization are the heaviest tools here and are worth reaching for once the dataset itself is large/versioned or the runtime has native, non-Python dependencies that a plain virtualenv cannot pin; a small project with a stable Python-only dependency set does not need either to be reproducible.

Include CI step that installs from lockfile and runs a smoke test to validate reproducibility.

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.