From Notebook to Something That Runs — Python Primer

Your analysis works. It produced a number somebody liked. Now it has to run every Monday morning without you, and that turns out to be a different problem entirely.

Why Notebooks Do Not Deploy

A notebook has hidden state. Cells can run in any order, and the variables in memory may reflect a sequence that exists nowhere in the file.

# Cell 3
df = df[df.revenue > 0]      # run twice, filters twice

# Cell 7
df = load_data()             # run after cell 3, undoes it

Reading the file top to bottom does not tell you what happened. The person who inherits it cannot reproduce your result, and neither can you next month.

The habit that catches most of this: restart the kernel and run everything, top to bottom, before believing any result. If it fails, the notebook was lying to you. It takes five seconds and it is the single most valuable notebook discipline there is.

Notebooks are excellent for exploration. Keep them for that. Anything worth running twice moves into a module.

Extracting a Module

The move is mechanical. Cells become functions, and the notebook becomes a thin caller.

# src/churn/pipeline.py
import pandas as pd
from pathlib import Path

def load(path: Path) -> pd.DataFrame:
    """Read the raw export with explicit types."""
    return pd.read_csv(path, dtype={'customer_id': 'string'},
                       parse_dates=['created_at'])

def clean(raw: pd.DataFrame) -> pd.DataFrame:
    """Return a cleaned copy. The input is not modified."""
    df = raw.copy()
    df['city'] = df.city.str.strip().str.lower()
    df = df.sort_values('updated_at').drop_duplicates('customer_id', keep='last')
    return df

def check(df: pd.DataFrame) -> None:
    """Assumptions that must hold."""
    assert df.customer_id.is_unique, "customer_id is not unique"
    assert df.revenue.ge(0).all(), "negative revenue"

def build_features(df: pd.DataFrame, snapshot: pd.Timestamp) -> pd.DataFrame:
    """Features as of a fixed point in time."""
    return df.assign(
        days_since_order=(snapshot - df.last_order).dt.days,
        revenue_vs_tier=df.revenue / df.groupby('tier').revenue.transform('mean'),
    )

The notebook then reads:

from churn.pipeline import load, clean, check, build_features

raw = load(Path('data/customers.csv'))
df = clean(raw)
check(df)
features = build_features(df, snapshot=pd.Timestamp('2026-03-01'))

Four properties you now have that you did not before. The functions can be tested. They can be reused. The notebook reads as a story. And running it top to bottom gives the same answer every time.

Testing Data Code

Testing a transformation is not like testing a web handler. You are checking properties of a result, and small hand-made inputs work well.

# tests/test_pipeline.py
import pandas as pd
import pytest
from churn.pipeline import clean, check

def test_clean_normalises_city():
    raw = pd.DataFrame({'city': ['  Delhi ', 'DELHI'],
                        'customer_id': ['a', 'b'],
                        'updated_at': pd.to_datetime(['2026-01-01', '2026-01-02'])})
    out = clean(raw)
    assert out.city.tolist() == ['delhi', 'delhi']

def test_clean_keeps_most_recent_duplicate():
    raw = pd.DataFrame({'customer_id': ['a', 'a'],
                        'city': ['delhi', 'sydney'],
                        'updated_at': pd.to_datetime(['2026-01-01', '2026-02-01'])})
    out = clean(raw)
    assert len(out) == 1
    assert out.city.iloc[0] == 'sydney'

def test_check_rejects_negative_revenue():
    df = pd.DataFrame({'customer_id': ['a'], 'revenue': [-1.0]})
    with pytest.raises(AssertionError, match='negative revenue'):
        check(df)

Three rows of made-up data each. Run with pytest.

The tests worth writing first, in order of value:

The bug you just fixed. Write it as a test so it cannot return.
The edge cases. Empty input, one row, all values missing, duplicate keys.
The invariants. Row counts, uniqueness, ranges, “no nulls in this column”.

You are not aiming for full coverage. You are aiming to be told when a change breaks something you rely on.

Configuration

Hard-coded paths and magic numbers are what make code impossible to run anywhere else.

# Before
df = pd.read_csv('/Users/sam/Desktop/data/customers_v3_FINAL.csv')
df = df[df.revenue > 100]

# After
from dataclasses import dataclass
from pathlib import Path

@dataclass
class Config:
    data_path: Path
    min_revenue: float = 100.0
    snapshot: str = '2026-03-01'
    random_state: int = 42

def run(cfg: Config):
    df = pd.read_csv(cfg.data_path)
    df = df[df.revenue > cfg.min_revenue]

A dataclass beats a dictionary because a typo in an attribute name raises instead of silently returning None. Secrets stay in environment variables, never in the config file and never in the code.

Pipelines That Cannot Leak

Everything the evaluation chapter argued for, made structural:

from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer

pipe = Pipeline([
    ('prep', ColumnTransformer([...])),
    ('model', GradientBoostingClassifier(random_state=cfg.random_state)),
])
pipe.fit(X_train, y_train)

The whole object is one artifact. Whoever deploys it cannot accidentally apply the preprocessing differently, because there is only one implementation and it travels with the model.

Persisting a Model Properly

import joblib, sklearn, datetime, json

artifact = {
    'pipeline': pipe,
    'sklearn_version': sklearn.__version__,
    'trained_at': datetime.datetime.now(datetime.timezone.utc).isoformat(),
    'features': list(X_train.columns),
    'train_rows': len(X_train),
    'cv_score': float(cv_score),
    'config': cfg.__dict__,
}
joblib.dump(artifact, 'models/churn-v4.joblib')

Three rules. Version the filename, never overwrite. Store the metadata alongside, because in six months someone will ask what this was trained on. And never load a pickle you do not trust, since loading one executes code.

Scheduling

Start with the simplest thing that works and only add machinery when you feel the pain.

# scripts/run_scoring.py
import logging, sys
from churn.pipeline import load, clean, check, build_features

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s %(levelname)s %(message)s')
log = logging.getLogger(__name__)

def main() -> int:
    try:
        log.info("starting")
        df = clean(load(cfg.data_path))
        check(df)
        scores = model.predict_proba(build_features(df, cfg.snapshot))[:, 1]
        write_results(scores)
        log.info("wrote %d scores", len(scores))
        return 0
    except Exception:
        log.exception("scoring failed")     # logs the full traceback
        return 1

if __name__ == '__main__':
    sys.exit(main())

cron runs that once a day and that is genuinely enough for a great many jobs. Move to Airflow or Prefect when you have dependencies between tasks, retries that matter, and backfills. Not before.

Two things the script above gets right. It exits non-zero on failure, so the scheduler knows something went wrong. And it logs the traceback, so you can diagnose it without reproducing it.

Monitoring

A model that has silently degraded is worse than one that is down, because nobody knows.

flowchart TB
    A["<b>Input distribution</b><br/>are today's features shaped<br/>like the training data?"] --> D["Alert"]
    B["<b>Output distribution</b><br/>has the score distribution<br/>shifted?"] --> D
    C["<b>Actual accuracy</b><br/>once outcomes arrive,<br/>weeks or months later"] --> D
    style A fill:#eef1fc,stroke:#2141c8
    style B fill:#eef1fc,stroke:#2141c8

The first two need no labels and no waiting, which is why they are the highest-return engineering in the whole field:

def check_drift(train_stats, today, threshold=0.25):
    """Flag features whose mean has moved more than a quarter of a training SD."""
    alerts = []
    for col, (mean, std) in train_stats.items():
        if std == 0: continue
        shift = abs(today[col].mean() - mean) / std
        if shift > threshold:
            alerts.append(f"{col}: shifted {shift:.2f} SD")
    return alerts

Also watch the null rate per column. A silently emptying feed is common and produces confident nonsense.

Rollback

You need to return to the previous version in minutes, not hours. That requires versioned model files, a config that names which version is live, and a deployment that reads it:

MODEL_VERSION = os.environ.get('MODEL_VERSION', 'churn-v4')
artifact = joblib.load(f'models/{MODEL_VERSION}.joblib')

Rolling back becomes changing one environment variable. Keep the previous two versions available.

A Checklist Before Anything Runs Unattended

Restart-and-run-all produces the same answer
Logic lives in importable modules, not notebook cells
Tests exist for the transformations and the edge cases
Assertions guard the assumptions, and they fail loudly
Paths and parameters come from config; secrets from the environment
Preprocessing is inside a Pipeline
The model artifact is versioned and carries its metadata
The script exits non-zero on failure and logs tracebacks
Input and output distributions are monitored
Rollback is one config change
A named person owns this in six months

Most stalled projects fail on the same three: no module, no tests, no owner.

What Carries Forward

Restart and run all before believing anything.

Notebooks call functions; modules define them.

Assert your assumptions. A crash beats a quiet wrong number.

Version the model and store its metadata. Someone will ask what it was trained on.

Monitor input and output distributions. They need no labels and catch most failures first.

Next, the last chapter: what to do when it works but is too slow.