Python for ML Engineers, Not Just Programmers

The Two Kinds of Python Programmer

There is a version of Python knowledge that gets you through tutorials and classroom assignments. And there is a version that gets you through code review, take-home assessments, and production systems.

The gap between them is not algorithms or frameworks. It is habits: how you structure code, handle errors, think about memory, and organize projects so that other people - and future you - can work with them reliably.

This module covers the Python habits that hiring teams look for before they look at your model code.

Python Data Structures: The Right Tool Every Time

Choosing Between List, Dict, and Set

Most ML code processes collections. Choosing the wrong structure costs performance and clarity.

List: Ordered, indexed, allows duplicates. Use when order matters or you need integer indexing.

pythonCopy
batch_predictions = [0.92, 0.15, 0.77, 0.43]

Dictionary: Key → value mapping, O(1) average lookup. Use when you need fast lookup by identifier.

pythonCopy
feature_stats = {
    "age": {"mean": 35.2, "std": 12.1},
    "income": {"mean": 65000, "std": 28000},
}
# O(1): feature_stats["age"]["mean"]

Set: Unordered, unique values, O(1) membership test. Use to check membership or remove duplicates.

pythonCopy
valid_categories = {"click", "purchase", "view"}
if event_type not in valid_categories:
    raise ValueError(f"Unknown event type: {event_type}")

Dataclass: Lightweight record with named fields. Use instead of raw dicts for structured results.

pythonCopy
from dataclasses import dataclass

@dataclass
class ModelResult:
    prediction: float
    confidence: float
    model_version: str

defaultdict and Counter for Frequency Analysis

Two standard library containers that eliminate boilerplate in data preprocessing:

pythonCopy
from collections import defaultdict, Counter

# defaultdict: eliminates the "if key not in dict" pattern
label_counts = defaultdict(int)
for label in labels:
    label_counts[label] += 1

# Counter: frequency counting with useful methods built in
label_counts = Counter(labels)
most_common = label_counts.most_common(5)  # top-5 by count
print(label_counts["positive"] / len(labels))  # class frequency

Counter is the right tool for class imbalance checks, vocabulary building, and frequency analysis.

Comprehensions and Generators: Processing at Scale

List Comprehensions

Prefer comprehensions over explicit loops when building lists - they are faster and more readable.

pythonCopy
# Loop-style
normalized = []
for x in raw_values:
    normalized.append((x - mean) / std)

# Comprehension-style - same result
normalized = [(x - mean) / std for x in raw_values]

# With filtering
valid_samples = [x for x in raw_values if x is not None and x > 0]

# Nested: flatten a list of batches
all_preds = [p for batch in prediction_batches for p in batch]

Generator Expressions for Memory Efficiency

When you iterate over results once and do not need to store them, use a generator:

pythonCopy
# List comprehension: builds the entire list in memory (~80 MB for 10M items)
squares = [x**2 for x in range(10_000_000)]

# Generator: computes values on demand, constant memory
squares_gen = (x**2 for x in range(10_000_000))
total = sum(squares_gen)  # No large list ever created

Generators are the right choice for:

• Iterating over large datasets that do not fit in memory
• Streaming log files or prediction outputs
• Building data augmentation pipelines

Dictionary Comprehensions

pythonCopy
# Map feature names to their medians (only for columns with missing values)
feature_medians = {
    col: df[col].median()
    for col in df.columns
    if df[col].isna().any()
}

functools and itertools: The Underused Standard Library

functools.lru_cache for Expensive Computations

pythonCopy
from functools import lru_cache

@lru_cache(maxsize=1024)
def embed_text(text: str) -> tuple[float, ...]:
    """Cache embedding results - avoids redundant API calls for repeated inputs."""
    result = embedding_model.encode(text)
    return tuple(result)  # Must be hashable to serve as a cache key

If embed_text is called with the same string twice, the second call returns the cached result instantly. For embedding pipelines that process repeated queries, this eliminates redundant API cost.

functools.partial for Reusable Configurations

pythonCopy
from functools import partial
from sklearn.metrics import fbeta_score

# Pre-configure F-beta with beta=2 (recall-weighted) for this project
f2_score = partial(fbeta_score, beta=2, average="binary")

# Now call consistently without repeating the beta argument
score_a = f2_score(y_true, preds_a)
score_b = f2_score(y_true, preds_b)

itertools for Batch Processing

pythonCopy
import itertools

def batched(iterable, n: int):
    """Yield successive n-sized batches from an iterable."""
    it = iter(iterable)
    while True:
        batch = list(itertools.islice(it, n))
        if not batch:
            break
        yield batch

for batch in batched(all_records, batch_size=256):
    predictions = model.predict(batch)
    store(predictions)

pathlib: Modern File Handling

pathlib.Path replaces os.path with an object-oriented, composable interface.

pythonCopy
from pathlib import Path

project_root = Path(__file__).parent.parent
data_dir = project_root / "data"
raw_dir = data_dir / "raw"
processed_dir = data_dir / "processed"

# Create directories
processed_dir.mkdir(parents=True, exist_ok=True)

# List all CSV files
csv_files = list(raw_dir.glob("*.csv"))

# Read and write files
config = (project_root / "configs" / "train.yaml").read_text()
(processed_dir / "features_v2.parquet").write_bytes(df.to_parquet())

# Existence check
if not (raw_dir / "events.parquet").exists():
    raise FileNotFoundError("Raw data not found. Run download_data.py first.")

Using pathlib throughout eliminates string concatenation bugs and makes path construction self-documenting.

Type Hints: Code That Explains Itself

Type hints are not enforced at runtime, but they serve as documentation, enable IDE autocomplete, and let mypy catch type errors before they reach production.

pythonCopy
import numpy as np
import pandas as pd
from sklearn.base import BaseEstimator
from collections import Counter

def compute_class_weights(y: np.ndarray) -> dict[int, float]:
    """Return weights inversely proportional to class frequency."""
    counts = Counter(y)
    total = len(y)
    return {cls: total / (len(counts) * count) for cls, count in counts.items()}

def fit_and_evaluate(
    model: BaseEstimator,
    X_train: np.ndarray,
    y_train: np.ndarray,
    X_val: np.ndarray,
    y_val: np.ndarray,
) -> dict[str, float]:
    model.fit(X_train, y_train)
    val_pred = model.predict(X_val)
    return {"accuracy": float((val_pred == y_val).mean()), "n_val": len(y_val)}

When a teammate sees fit_and_evaluate, they know the expected types without reading the body.

Decorators for ML Code

Three practical decorator uses in ML engineering:

pythonCopy
import time
import logging
import functools

logger = logging.getLogger(__name__)

def timer(func):
    """Log how long a function takes."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        logger.info(f"{func.__name__} completed in {elapsed:.2f}s")
        return result
    return wrapper

def retry(max_attempts: int = 3):
    """Retry a function on failure with exponential backoff."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(2 ** attempt)
        return wrapper
    return decorator

@timer
@retry(max_attempts=3)
def call_embedding_api(text: str) -> list[float]:
    # API calls benefit from both timing and retry
    return client.embed(text)

@functools.wraps(func) preserves the original function's name and docstring - always include it in decorators.

Structuring a Reusable ML Package

The deliverable for this module is a small Python package. The minimum viable structure:

src/
└── myproject/
    ├── __init__.py           # Public API: what callers import
    ├── data/
    │   ├── __init__.py
    │   ├── loader.py         # Data loading → returns DataFrames
    │   └── validator.py      # Schema and quality checks
    ├── features/
    │   ├── __init__.py
    │   └── pipeline.py       # sklearn Pipeline construction
    └── utils/
        ├── __init__.py
        ├── config.py         # TrainConfig dataclass
        └── logging.py        # Logging setup

Single responsibility per module: loader.py loads data, validator.py checks it, pipeline.py transforms it. Nothing reaches across boundaries unnecessarily.

pythonCopy
# src/myproject/__init__.py - the public API
from myproject.data.loader import load_training_data
from myproject.features.pipeline import build_preprocessor
from myproject.utils.config import TrainConfig

__all__ = ["load_training_data", "build_preprocessor", "TrainConfig"]

A simple CLI entry point:

pythonCopy
# scripts/train.py
import argparse
from myproject import load_training_data, build_preprocessor, TrainConfig

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", required=True)
    parser.add_argument("--output-dir", default="models/")
    args = parser.parse_args()

    config = TrainConfig.from_yaml(args.config)
    df = load_training_data(config.data_path)
    # ... training logic ...

if __name__ == "__main__":
    main()

Common Anti-Patterns in ML Python

Mutable default arguments:

pythonCopy
# Wrong - the list is shared across all calls
def add_feature(features, result=[]):
    result.append(features)
    return result

# Correct
def add_feature(features, result=None):
    if result is None:
        result = []
    result.append(features)
    return result

Bare except that swallows errors:

pythonCopy
# Wrong - hides whether it was a data error, memory error, or bug
try:
    prediction = model.predict(features)
except:
    prediction = default_value

# Correct - catch only what you can meaningfully handle
try:
    prediction = model.predict(features)
except ValueError as e:
    logger.warning(f"Invalid features for prediction: {e}")
    prediction = default_value

Modifying a collection while iterating over it:

pythonCopy
# Wrong - skips elements unpredictably
for sample in samples:
    if sample.is_invalid():
        samples.remove(sample)

# Correct - filter to a new list
samples = [s for s in samples if not s.is_invalid()]

Where to Go Next

The deliverable for this module: a small reusable package with a data loader, a preprocessing function, and a CLI entry point - organized using the structure above, with type hints, logging, and at least one unit test.

Module 2 builds on this by adding version control, environment management, and the terminal habits that make ML project workflows reproducible and auditable.