Ch 3 — Python Engineering Practices¶

# src/my_ai/__init__.py
"""my_ai — Applied AI library."""

from importlib.metadata import version, PackageNotFoundError

try:
    __version__: str = version("my-ai")
except PackageNotFoundError:  # running from source without install
    __version__ = "0.0.0+dev"

# Explicit public API — only what callers should import directly
from my_ai.models.transformer import TransformerModel
from my_ai.data.loader import DataLoader

__all__ = ["TransformerModel", "DataLoader", "__version__"]

# tests/test_metrics.py
import numpy as np
import pytest

from my_ai.utils.metrics import cosine_similarity, top_k_accuracy


class TestCosineSimilarity:
    def test_identical_vectors_score_one(self) -> None:
        v = np.array([1.0, 2.0, 3.0])
        assert cosine_similarity(v, v) == pytest.approx(1.0, abs=1e-6)

    def test_orthogonal_vectors_score_zero(self) -> None:
        a = np.array([1.0, 0.0])
        b = np.array([0.0, 1.0])
        assert cosine_similarity(a, b) == pytest.approx(0.0, abs=1e-6)

    def test_opposite_vectors_score_negative_one(self) -> None:
        a = np.array([1.0, 0.0])
        b = np.array([-1.0, 0.0])
        assert cosine_similarity(a, b) == pytest.approx(-1.0, abs=1e-6)

    def test_zero_vector_raises(self) -> None:
        zero = np.zeros(3)
        with pytest.raises(ValueError, match="zero vector"):
            cosine_similarity(zero, np.array([1.0, 2.0, 3.0]))

# tests/conftest.py
import numpy as np
import pytest


@pytest.fixture(scope="session")
def rng() -> np.random.Generator:
    """Deterministic RNG shared across the entire test session."""
    return np.random.default_rng(seed=42)


@pytest.fixture
def small_embedding_table(rng: np.random.Generator) -> np.ndarray:
    """256-row, 64-dim embedding table — fast to create, large enough to test."""
    return rng.standard_normal((256, 64)).astype(np.float32)


@pytest.fixture
def sample_texts() -> list[str]:
    return [
        "the model converged after 10 epochs",
        "gradient clipping prevents exploding gradients",
        "attention is all you need",
    ]

# tests/test_loader.py
import numpy as np

from my_ai.data.loader import embed_texts


def test_embed_texts_shape(
    small_embedding_table: np.ndarray,
    sample_texts: list[str],
) -> None:
    result = embed_texts(sample_texts, small_embedding_table)
    assert result.shape == (len(sample_texts), 64)

# tests/test_transforms.py
import numpy as np
import pytest

from my_ai.data.transforms import l2_normalise


@pytest.mark.parametrize(
    "shape",
    [(1, 8), (16, 256), (3, 1, 512)],
    ids=["1x8", "16x256", "3x1x512"],
)
def test_l2_normalise_unit_norm(shape: tuple[int, ...]) -> None:
    rng = np.random.default_rng(0)
    X = rng.standard_normal(shape).astype(np.float32)
    X_norm = l2_normalise(X, axis=-1)
    norms = np.linalg.norm(X_norm, axis=-1)
    np.testing.assert_allclose(norms, np.ones_like(norms), atol=1e-5)


@pytest.mark.parametrize(
    "bad_input, error_type",
    [
        (np.array([]), ValueError),
        (np.zeros((4, 8)), ValueError),  # zero-norm rows
    ],
)
def test_l2_normalise_invalid_inputs(
    bad_input: np.ndarray, error_type: type[Exception]
) -> None:
    with pytest.raises(error_type):
        l2_normalise(bad_input, axis=-1)

# tests/test_api_client.py
from unittest.mock import MagicMock, patch

import pytest

from my_ai.data.api_client import EmbeddingAPIClient


def test_client_retries_on_timeout() -> None:
    client = EmbeddingAPIClient(base_url="https://api.example.com", max_retries=3)

    call_count = 0

    def flaky_post(*args, **kwargs):  # type: ignore[no-untyped-def]
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise TimeoutError("request timed out")
        mock_resp = MagicMock()
        mock_resp.json.return_value = {"embedding": [0.1, 0.2, 0.3]}
        mock_resp.status_code = 200
        return mock_resp

    with patch("requests.Session.post", side_effect=flaky_post):
        result = client.embed("hello world")

    assert call_count == 3
    assert len(result) == 3


def test_client_raises_after_max_retries() -> None:
    client = EmbeddingAPIClient(base_url="https://api.example.com", max_retries=2)

    with patch("requests.Session.post", side_effect=TimeoutError):
        with pytest.raises(TimeoutError):
            client.embed("hello")

from typing import Any
import numpy as np


# ERROR: Returning Any from typed function
# def get_embedding(model: Any, text: str):    ← mypy: missing return type
#     return model.encode(text)

# FIX: annotate the return
def get_embedding(model: Any, text: str) -> np.ndarray:
    result: np.ndarray = model.encode(text)
    return result


# ERROR: Item "None" of "Optional[str]" has no attribute "upper"
# def shout(s: str | None) -> str:
#     return s.upper()           ← mypy: value is not always str

# FIX: guard before use
def shout(s: str | None) -> str:
    if s is None:
        return ""
    return s.upper()


# ERROR: Dict[str, int] not compatible with Dict[str, Any]
# def process(cfg: dict[str, Any]) -> None: ...
# int_cfg: dict[str, int] = {"lr": 1}
# process(int_cfg)   ← mypy: dict is invariant in value type

# FIX: use Mapping (covariant) for read-only parameters
from collections.abc import Mapping


def process(cfg: Mapping[str, int]) -> None:
    lr = cfg["lr"]
    print(f"lr={lr}")

from typing import Protocol
import numpy as np


class TokenizerProtocol(Protocol):
    """Structural type that any tokenizer must satisfy."""

    def encode(self, text: str) -> list[int]: ...
    def decode(self, ids: list[int]) -> str: ...
    @property
    def vocab_size(self) -> int: ...


def build_vocab_embeddings(
    tokenizer: TokenizerProtocol,
    dim: int,
) -> np.ndarray:
    """Works with any object satisfying TokenizerProtocol — no inheritance required."""
    rng = np.random.default_rng(0)
    return rng.standard_normal((tokenizer.vocab_size, dim)).astype(np.float32)

import torch  # no bundled stubs — mypy complains without ignore

# Single-line suppression with error code (preferred over blanket ignore)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")  # type: ignore[attr-defined]

# Document WHY you are suppressing — future maintainers will thank you
model = torch.nn.Linear(128, 64)  # type: ignore[no-untyped-call]  # torch stubs incomplete

# src/my_ai/utils/log.py
import logging
import sys


def get_logger(name: str, level: int = logging.INFO) -> logging.Logger:
    logger = logging.getLogger(name)
    if logger.handlers:
        return logger  # already configured — avoid duplicate handlers

    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(
        logging.Formatter(
            fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
            datefmt="%Y-%m-%dT%H:%M:%S",
        )
    )
    logger.addHandler(handler)
    logger.setLevel(level)
    logger.propagate = False
    return logger


# Usage in library code — never call logging.basicConfig() inside a library
log = get_logger(__name__)

def train_step(loss: float, step: int) -> None:
    log.info("step=%d loss=%.4f", step, loss)  # use % formatting — lazily evaluated
    if loss > 10.0:
        log.warning("Unusually high loss at step %d: %.4f", step, loss)

# src/my_ai/utils/structured_log.py
import logging
import sys

import structlog


def configure_logging(*, json_output: bool = False) -> None:
    """Call once at application startup — not inside library code."""
    shared_processors: list[structlog.types.Processor] = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_log_level,
        structlog.stdlib.add_logger_name,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
    ]

    if json_output:
        renderer: structlog.types.Processor = structlog.processors.JSONRenderer()
    else:
        renderer = structlog.dev.ConsoleRenderer(colors=True)

    structlog.configure(
        processors=[*shared_processors, renderer],
        wrapper_class=structlog.stdlib.BoundLogger,
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )
    logging.basicConfig(format="%(message)s", stream=sys.stdout, level=logging.INFO)


# Usage
import structlog as _structlog

log = _structlog.get_logger(__name__)


def load_dataset(path: str, split: str) -> None:
    log = _structlog.get_logger(__name__).bind(split=split, path=path)
    log.info("loading_dataset")
    # ... load ...
    log.info("dataset_loaded", n_samples=10_000, dtype="float32")

# src/my_ai/config.py
from __future__ import annotations

from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        env_prefix="MY_AI_",       # MY_AI_DATABASE_URL, MY_AI_API_KEY, etc.
        case_sensitive=False,
    )

    # Required — no default; raises ValidationError if missing
    api_key: str = Field(description="OpenAI-compatible API key")

    # Optional with defaults
    model_name: str = Field(default="gpt-4o-mini", description="Chat model identifier")
    embedding_model: str = Field(default="text-embedding-3-small")
    max_tokens: int = Field(default=2048, ge=1, le=32_768)
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    batch_size: int = Field(default=32, ge=1)
    log_level: str = Field(default="INFO")

    @field_validator("log_level")
    @classmethod
    def validate_log_level(cls, v: str) -> str:
        allowed = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
        upper = v.upper()
        if upper not in allowed:
            raise ValueError(f"log_level must be one of {allowed}; got {v!r}")
        return upper


# Singleton pattern — import this object, not the class
settings = Settings()  # type: ignore[call-arg]  # reads from env / .env file

import argparse
import sys


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Train an embedding model",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument("--data-dir", required=True, help="Path to training data")
    parser.add_argument("--output-dir", required=True, help="Where to save checkpoints")
    parser.add_argument("--epochs", type=int, default=10)
    parser.add_argument("--batch-size", type=int, default=32)
    parser.add_argument("--lr", type=float, default=1e-4)
    parser.add_argument(
        "--device",
        choices=["cpu", "cuda", "mps"],
        default="cpu",
    )
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> None:
    args = parse_args(argv)
    print(f"Training on {args.device} for {args.epochs} epochs")


if __name__ == "__main__":
    main()

import click


@click.group()
def cli() -> None:
    """my-ai command-line tools."""


@cli.command("train")
@click.option("--data-dir", required=True, type=click.Path(exists=True))
@click.option("--output-dir", required=True, type=click.Path())
@click.option("--epochs", default=10, show_default=True)
@click.option("--batch-size", default=32, show_default=True)
@click.option("--lr", default=1e-4, show_default=True)
@click.option("--device", type=click.Choice(["cpu", "cuda", "mps"]), default="cpu")
def train_cmd(
    data_dir: str,
    output_dir: str,
    epochs: int,
    batch_size: int,
    lr: float,
    device: str,
) -> None:
    """Fine-tune an embedding model."""
    click.echo(f"Training on {device} — {epochs} epochs")


@cli.command("evaluate")
@click.argument("checkpoint", type=click.Path(exists=True))
@click.option("--split", type=click.Choice(["val", "test"]), default="val")
def evaluate_cmd(checkpoint: str, split: str) -> None:
    """Evaluate a saved checkpoint."""
    click.echo(f"Evaluating {checkpoint} on {split} split")


if __name__ == "__main__":
    cli()

# src/my_ai/cli.py
from __future__ import annotations

from pathlib import Path
from typing import Annotated

import typer

app = typer.Typer(help="my-ai command-line tools", pretty_exceptions_enable=False)


@app.command("train")
def train(
    data_dir: Annotated[Path, typer.Option("--data-dir", exists=True, file_okay=False)],
    output_dir: Annotated[Path, typer.Option("--output-dir")],
    epochs: Annotated[int, typer.Option(min=1, max=1000)] = 10,
    batch_size: Annotated[int, typer.Option(min=1)] = 32,
    lr: Annotated[float, typer.Option(min=1e-7, max=1.0)] = 1e-4,
    device: Annotated[str, typer.Option()] = "cpu",
) -> None:
    """Fine-tune an embedding model."""
    typer.echo(f"Training on {device} for {epochs} epochs")
    output_dir.mkdir(parents=True, exist_ok=True)


@app.command("evaluate")
def evaluate(
    checkpoint: Annotated[Path, typer.Argument(exists=True)],
    split: Annotated[str, typer.Option()] = "val",
    batch_size: Annotated[int, typer.Option(min=1)] = 64,
) -> None:
    """Evaluate a saved checkpoint."""
    typer.echo(f"Evaluating {checkpoint} on {split}")


if __name__ == "__main__":
    app()