3. Transfer to Python

Introduction

Your notebook from the previous activity already contains a correct pipeline. But look closely: you filled missing values the same way for every column, you will repeat the same cleaning and feature steps the next time you touch this data, and nothing in the notebook can be reused without copy-pasting cells. This activity turns the parts of the notebook you keep repeating into small, tested Python functions — and from here on, the notebook stops being the place where the pipeline lives. You will not come back to edit it after this activity: pytest and the functions themselves are how you verify your work from now on.

Before this activity, read Towards Production-Ready ML if you have not already: it explains why notebook code and production code solve different problems, and what a first, minimal separation of concerns looks like.

Read workflows.py First

Before touching any file, open src/air_quality/workflows.py. It already exists, complete, and you will not need to change it — it is the map. It shows exactly which functions the pipeline needs and in which order; your job in this activity is to make those functions actually work.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

@dataclass
class PipelineConfig:
    train_city: str = "Kampala"
    test_city: str = "Nairobi"
    columns: list[str] = field(default_factory=lambda: list(DEFAULT_COLUMNS))
    max_rows_per_city: int = 1200
    random_state: int = 42


def run_baseline(config: PipelineConfig | None = None) -> dict[str, float]:
    config = config or PipelineConfig()
    train_df, _ = data.load_datasets()
    cities = [config.train_city, config.test_city]

    scoped = data.restrict_to_scope(
        train_df, cities, config.columns, config.max_rows_per_city, config.random_state
    )
    fillable_columns = [c for c in config.columns if c not in ("city", "date")]
    cleaned = data.fill_missing_by_city(scoped, fillable_columns)
    enriched = features.add_temporal_features(cleaned)

    train_split = enriched[enriched["city"] == config.train_city]
    test_split = enriched[enriched["city"] == config.test_city]

    feature_cols = features.feature_columns(enriched)
    model = LinearRegression()
    return evaluation.evaluate_manual_split(model, train_split, test_split, feature_cols)

A few things worth understanding before you go further, since none of this appeared in the notebook:

• @dataclass writes the boring part of a class for you: given typed attributes like the ones below, it generates the code that assigns them from constructor arguments. That is why PipelineConfig(train_city="Nairobi", test_city="Kampala") already works — the same reversal you tried in the previous activity’s reflection question — without a single line of __init__ written by hand. See the dataclasses documentation.
• field(default_factory=lambda: list(DEFAULT_COLUMNS)) — a dataclass field can’t default directly to a mutable object like a list: columns: list[str] = DEFAULT_COLUMNS would make every PipelineConfig share, and risk mutating, that exact same list — the same trap you’ll see again just below with config = config or PipelineConfig(), just inside a dataclass field this time instead of a function argument. default_factory takes a function to call fresh for each new instance; here, that function builds a new list copied from DEFAULT_COLUMNS.
• config = config or PipelineConfig() — why not write the default directly in the signature, as config: PipelineConfig = PipelineConfig()? A default argument value is built exactly once, when Python reads the function definition — not once per call. Writing it directly there would mean every call that omits config shares that same single instance. Writing config: PipelineConfig | None = None, then substituting a fresh instance inside the function body, avoids this classic Python trap. See the Common Gotchas: mutable default arguments.
• train_df, _ = data.load_datasets() — load_datasets() returns two DataFrames, train and test; this pipeline only needs the train one. _ is the Python convention for “a value this line must unpack, but that nothing afterward will use.”
• Every call is written as data.load_datasets(...), features.add_temporal_features(...), evaluation.evaluate_manual_split(...) — module name first, function name second. That is deliberate: reading workflows.py, you can always tell which file to open to find, or fix, a given function.

None of the functions run_baseline calls are implemented yet — each exists only as a signature and a docstring. That is what you write next.

Implementation

Files: src/air_quality/data.py, src/air_quality/features.py, src/air_quality/evaluation.py, tests/test_data.py, tests/test_features.py, tests/test_evaluation.py, tests/test_workflows.py, scripts/run_pipeline.py

The project already contains an (empty-of-your-code) package at src/air_quality/ and a matching test suite at tests/. You are not translating the notebook cell by cell: group what it does into three small modules, by responsibility rather than by the order the cells happen to appear in.

src/air_quality/data.py

1
2
3
4
5
6
7
8

def load_datasets(data_dir=DEFAULT_DATA_DIR) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Load the raw train and test CSV files."""

def restrict_to_scope(df, cities, columns, max_rows_per_city=None, random_state=42) -> pd.DataFrame:
    """Keep only the given cities and columns, optionally capping rows per city."""

def fill_missing_by_city(df, columns, city_col="city", date_col="date") -> pd.DataFrame:
    """Forward/backward-fill the given columns within each city, ordered by date."""

src/air_quality/features.py

1
2
3
4
5

def add_temporal_features(df, date_col="date") -> pd.DataFrame:
    """Add month and day-of-week features extracted from the date column."""

def feature_columns(df, target_col="pm2_5") -> list[str]:
    """List the numeric columns usable as model features (excludes the target, identifiers, and site_latitude/site_longitude)."""

src/air_quality/evaluation.py

1
2
3
4
5

def regression_metrics(y_true, y_pred) -> dict[str, float]:
    """Compute RMSE, MAE and R2 for a set of predictions."""

def evaluate_manual_split(model, train_df, test_df, feature_cols, target_col="pm2_5") -> dict[str, float]:
    """Fit a model on train_df and evaluate it on test_df (a different city)."""

Principles to follow

• One function, one responsibility — each function above does exactly one thing your notebook currently does inline.
• No notebook-global state — a function receives everything it needs as a parameter and returns its result; it must not silently depend on a variable defined three cells earlier.
• Not everything has to become a function. The manual train/test split (df[df["city"] == TRAIN_CITY]) is simple enough to stay a plain expression, as you can see directly inside run_baseline. Extracting it would add a layer of indirection without making anything clearer. Recognizing when not to extract is as much the point of this activity as extracting.
• Test as you go, one module at a time, not everything at the end.

Testing Strategy

Run tests module by module as you complete each function — this is also how you will notice quickly if a function you have not implemented yet is blocking one you have:

1
2
3

uv run pytest tests/test_data.py -q
uv run pytest tests/test_features.py -q
uv run pytest tests/test_evaluation.py -q

Once all three pass, check that the whole pipeline works end to end, exactly as workflows.run_baseline assembles it:

1
2

uv run pytest tests/test_workflows.py -q
uv run python scripts/run_pipeline.py

Commit your work

You already introduced Conventional Commits to publish the Sales project in Session 1 — this is the moment to actually apply that discipline, not just remember it exists. Commit per module as you implement it, not as one pile at the end, so your history shows the transformation happening one reviewable step at a time:

1
2
3
4
5
6
7
8

git add src/air_quality/data.py tests/test_data.py
git commit -m "refactor(air-quality): implement data cleaning functions in data.py"

git add src/air_quality/features.py tests/test_features.py
git commit -m "refactor(air-quality): implement feature engineering functions in features.py"

git add src/air_quality/evaluation.py tests/test_evaluation.py
git commit -m "refactor(air-quality): implement evaluation functions in evaluation.py"

Choose the type the same way you did in Session 1: refactor: for moving already-working logic out of the notebook, test: if you add a test for its own sake, fix: if implementing a function surfaces a bug the notebook was hiding.

Reflection

Files: reflection/session_2/module_3.md