Build a Production ML Pipeline

Use advanced Python patterns to build a production-grade ML pipeline.

Introduction

30 Second Summary

Every machine learning model in production is surrounded by invisible infrastructure code. Timing, logging, validation, and retries happen behind the scenes and take more engineering effort than the model itself.

In this project, you will build a production-style ML pipeline framework using advanced Python patterns. Your framework will automatically handle timing, logging, and input validation so the core ML logic stays clean and focused.

What You'll Build

A Python ML pipeline framework that trains and evaluates a classification model on the Iris dataset, powered by custom decorators, dataclasses, and Protocol classes.

By the end of this project, you'll have:

  • A pipeline.py file with a custom pipeline framework using decorators, dataclasses, and Protocol classes.
  • A train.py file that uses the framework to train and evaluate a scikit-learn classification model.
  • Terminal output showing the pipeline running with automated timing, logging, and validation built into every step.
  • Secret Mission: A @retry decorator that automatically retries failed pipeline steps with exponential backoff.

Are there any prerequisites?

You need basic Python knowledge (functions, classes, and loops). No machine learning experience is required.

Before We Start

Before you can build a production ML pipeline, you need the right Python version and a clean workspace.

In this step, you will install Python 3.11+, create an isolated environment with your ML dependencies, and set up the project folder in Cursor.

In this step, get ready to:
  • Install and verify Python 3.11+ on your machine.
  • Create a virtual environment and install scikit-learn and numpy.
  • Set up your project folder in Cursor with two starter files.
Verify your Python installation

This project requires Python 3.11 or higher because scikit-learn 1.9.0 needs it, and the code uses modern type hint syntax introduced in 3.11.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Check your Python version by running this command:

macOS

python3 --version

Windows

python --version

You should see a version number printed. Check your result below.

✔️ I see version 3.11 or higher

You are all set. Move on to the next substep below.

ⓧ I see an older version

Your system has an older Python that will not work for this project. You need to install Python 3.11+ alongside it.

On macOS, the system python3 often points to an older version even after you install a newer one. The recommended approach is to use pyenv to manage versions.

  • Visit python.org/downloads and download the latest Python (currently 3.14.5).
  • Run the installer and follow the prompts.
  • After installation, close and reopen your terminal.
  • On macOS, the new version is available as python3.14 (the full versioned command). Your system python3 may still point to the old version.
  • Verify by running the versioned command:

macOS

python3.14 --version

Windows

python --version

On Windows, the python.org installer updates your python command to the new version. On macOS, use the full versioned command (python3.14) for the rest of this project wherever you see python3.

Use pyenv for easier version management

If you want a cleaner experience, install pyenv and run:

pyenv install 3.11
pyenv local 3.11

This sets python3 to 3.11 inside your project folder automatically.

ⓧ Command not found

Python is not installed on your machine yet.

  • Visit python.org/downloads and download the latest Python (currently 3.14.5).
  • Run the installer. On Windows, check the box that says Add Python to PATH before clicking Install Now.
  • Close and reopen your terminal after installation.
  • Verify by re-running the version check command from above.

You should now see a version number of 3.11 or higher.

Still not seeing the right version?

Make sure you fully closed and reopened your terminal after installing. New PATH entries only take effect in new terminal sessions.

Help me get Python 3.11+ working on my machine.

Create your project and install dependencies

A virtual environment gives your project its own isolated set of packages. This prevents conflicts with other Python projects on your machine.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir ml-pipeline
cd ml-pipeline
  • Create a virtual environment by running this command:

macOS

python3 -m venv venv

Windows

python -m venv venv

What is a virtual environment?

The venv module creates a self-contained folder (also called venv) that holds its own copy of pip and any packages you install. When you activate it, your terminal uses this isolated copy instead of the system-wide one.

  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

.\venv\Scripts\Activate.ps1

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

  • Install scikit-learn and NumPy by running this command:
pip install scikit-learn numpy

What are these packages?

scikit-learn is a machine learning library that provides ready-to-use algorithms for classification, regression, and clustering. NumPy is the foundation for numerical computing in Python and is required by scikit-learn.

The install will download several packages (including scipy, joblib, and threadpoolctl as dependencies). This may take a minute.

  • Create the two project files by running this command:

macOS

touch pipeline.py train.py

Windows

New-Item pipeline.py -ItemType File
New-Item train.py -ItemType File
  • Confirm the files exist by running:
ls

You should see pipeline.py, train.py, and the venv folder in the output.

Not seeing (venv) in your prompt?

If (venv) disappeared, you may have closed your terminal or opened a new tab. Re-run the activate command from above before continuing.

On Windows, if you get an execution policy error, run Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser first, then try activating again.

Help me activate my virtual environment.

Open the project in Cursor
  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Cursor and press Enter to open it.
  • In Cursor, click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop and select the ml-pipeline folder, then click Open.

In the Explorer sidebar on the left, you should see pipeline.py and train.py listed. These are currently empty files that you will fill in over the next steps.

Don't see your files in the sidebar?

Make sure you opened the ml-pipeline folder itself, not a parent folder like Desktop. The files should appear at the top level of the Explorer sidebar.

Help me open my project folder in Cursor.

Your environment is ready. Next up, you will build the pipeline foundation using dataclasses and type hints to create clean, self-documenting data models.

Build the Pipeline Foundation

Your environment is locked and loaded. Now it is time to build the core framework that your ML pipeline runs on.

In production ML systems, engineers spend more time on infrastructure code than model code. The patterns you build in this step (dataclasses, type hints, and Protocols) keep that infrastructure clean, self-documenting, and extensible.

In this step, get ready to:
  • Define dataclasses that model pipeline configuration and step results.
  • Create a Protocol class that enforces a contract for all pipeline steps.
  • Build a Pipeline class that registers and executes steps in sequence.
Add imports, logging, and dataclasses to pipeline.py

Every Python module starts with imports. Your pipeline framework relies on several standard library modules for timing, function wrapping, structured data, and type checking.

  • Switch back to Cursor and open pipeline.py.
  • Add the following imports and logging configuration at the top of the file:
"""Production ML Pipeline Framework using Advanced Python patterns."""

import time
import functools
import logging
from dataclasses import dataclass, field
from typing import Any, Protocol, runtime_checkable

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)

What does this code do?

The imports pull in the standard library tools your framework needs. time measures execution duration. functools provides decorator utilities. dataclasses and typing enable type-safe data modeling.

The logging configuration sets up structured log output with timestamps. Every pipeline step will use this logger to report what is happening during execution.

You will not see a change yet. The next chunk adds the data models that use these imports.

  • Below the logging setup, add the PipelineConfig dataclass:

# --- Dataclasses for clean data modeling ---

@dataclass
class PipelineConfig:
    """Configuration for a pipeline run."""

    name: str
    random_seed: int = 42
    test_size: float = 0.2
    verbose: bool = True

What is a dataclass?

A dataclass automatically generates __init__, __repr__, and other boilerplate methods from your field definitions. You define the shape of your data once and Python handles the rest.

PipelineConfig stores settings that apply to an entire pipeline run. Type hints like int = 42 give each field a type AND a default value, making the code self-documenting.

You will not see a change yet. The next dataclass completes the data modeling layer.

  • Below PipelineConfig, add the StepResult dataclass:

@dataclass
class StepResult:
    """Result of a pipeline step execution."""

    step_name: str
    duration_seconds: float
    output: Any
    metadata: dict[str, Any] = field(default_factory=dict)

What does StepResult capture?

Every time a pipeline step runs, it produces a StepResult recording the step name, how long it took, and what it returned. The metadata field uses field(default_factory=dict) to safely create a fresh empty dictionary for each instance.

The dict[str, Any] syntax is a modern Python 3.9+ type hint. It replaces the older Dict[str, Any] import from typing.

  • Save pipeline.py.
  • Verify your code has no syntax errors by running this in your terminal:
python pipeline.py

You should see no output at all. Silence means Python loaded your module without errors.

Seeing a SyntaxError or ImportError?

Check that your @dataclass decorator sits directly above each class line with no blank lines between them. Make sure every field has a colon and type annotation (e.g. name: str).

If you see ModuleNotFoundError, confirm your virtual environment is still active. You should see the environment name in your terminal prompt.

help me fix a SyntaxError in my pipeline.py dataclasses

Define a PipelineStep Protocol class

Your pipeline needs a way to guarantee that any step plugged into it has the right interface. A Protocol class defines this contract without requiring inheritance.

  • Below StepResult, add the PipelineStep Protocol:

# --- Protocol for extensibility ---

@runtime_checkable
class PipelineStep(Protocol):
    """Contract that all pipeline steps must satisfy."""

    @property
    def name(self) -> str: ...

    def execute(self, data: Any) -> Any: ...

What is a Protocol class?

A Protocol defines structural subtyping. Any class that has a name property and an execute method automatically satisfies this contract. No explicit inheritance needed.

The @runtime_checkable decorator lets you use isinstance() checks against this Protocol at runtime. The ... (ellipsis) after each method signature means "no implementation here, just the shape."

Why Protocol instead of an abstract base class?

Abstract base classes require every step to explicitly inherit from them. Protocols check structure, not lineage. If a class walks like a PipelineStep and quacks like a PipelineStep, it IS a PipelineStep.

This is the modern Python approach (called structural subtyping). It makes your pipeline framework more flexible because third-party code can satisfy the contract without knowing about your base class.

  • Save pipeline.py and verify again:
python pipeline.py

No output means everything is still loading cleanly.

Seeing a NameError about Protocol?

Make sure your import line reads from typing import Any, Protocol, runtime_checkable. All three names must be imported on the same line at the top of your file.

help me fix a NameError for Protocol in pipeline.py

Build the Pipeline orchestrator class

The Pipeline class ties everything together. It accepts a configuration, lets you register steps, then executes them in sequence while recording timing results.

  • Below the PipelineStep Protocol, add the full Pipeline class:

# --- Pipeline orchestrator ---

class Pipeline:
    """Orchestrates pipeline step execution."""

    def __init__(self, config: PipelineConfig):
        self.config = config
        self.steps: list[tuple[str, Any]] = []
        self.results: list[StepResult] = []

    def add_step(self, name: str, step_func) -> "Pipeline":
        """Register a step. Returns self for chaining."""
        self.steps.append((name, step_func))
        return self

    def run(self, initial_data: Any = None) -> list[StepResult]:
        """Execute all steps in sequence, passing data through."""
        logger.info(f"{'='*50}")
        logger.info(f"Pipeline: {self.config.name}")
        logger.info(f"{'='*50}")

        data = initial_data
        self.results = []

        for name, step_func in self.steps:
            start = time.perf_counter()
            data = step_func(data)
            elapsed = time.perf_counter() - start

            result = StepResult(
                step_name=name,
                duration_seconds=elapsed,
                output=data,
            )
            self.results.append(result)

        logger.info(f"{'='*50}")
        total_time = sum(r.duration_seconds for r in self.results)
        logger.info(f"Pipeline complete! Total time: {total_time:.4f}s")
        logger.info(f"{'='*50}")

        return self.results

    def summary(self) -> str:
        """Return a formatted summary of the pipeline run."""
        lines = [f"\nPipeline Summary: {self.config.name}", "-" * 40]
        for result in self.results:
            lines.append(
                f"  {result.step_name}: {result.duration_seconds:.4f}s"
            )
        total = sum(r.duration_seconds for r in self.results)
        lines.append("-" * 40)
        lines.append(f"  Total: {total:.4f}s")
        return "\n".join(lines)

What does the Pipeline class do?

  • __init__ stores the config and creates empty lists for steps and results.
  • add_step registers a named function. It returns self so you can chain calls like pipeline.add_step(...).add_step(...).
  • run loops through each step, passes data forward, and times each execution with time.perf_counter().
  • summary formats a human-readable timing report after the run completes.
  • Save pipeline.py.
  • Run a quick import check to confirm everything works together:
python -c "from pipeline import Pipeline, PipelineConfig, StepResult, PipelineStep; print('Pipeline module loaded successfully')"

You should see Pipeline module loaded successfully printed in your terminal.

Seeing an error on import?

If you see IndentationError, check that every method inside the Pipeline class is indented one level (4 spaces) from the class body. The code inside each method is indented two levels (8 spaces) from the class line.

If you see NameError for PipelineConfig or StepResult, make sure the Pipeline class appears BELOW both dataclasses in the file. Python reads top to bottom.

help me debug my Pipeline class import error

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and move on to the next step.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py at this point:

"""Production ML Pipeline Framework using Advanced Python patterns."""

import time
import functools
import logging
from dataclasses import dataclass, field
from typing import Any, Protocol, runtime_checkable

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)


# --- Dataclasses for clean data modeling ---

@dataclass
class PipelineConfig:
    """Configuration for a pipeline run."""

    name: str
    random_seed: int = 42
    test_size: float = 0.2
    verbose: bool = True


@dataclass
class StepResult:
    """Result of a pipeline step execution."""

    step_name: str
    duration_seconds: float
    output: Any
    metadata: dict[str, Any] = field(default_factory=dict)


# --- Protocol for extensibility ---

@runtime_checkable
class PipelineStep(Protocol):
    """Contract that all pipeline steps must satisfy."""

    @property
    def name(self) -> str: ...

    def execute(self, data: Any) -> Any: ...


# --- Pipeline orchestrator ---

class Pipeline:
    """Orchestrates pipeline step execution."""

    def __init__(self, config: PipelineConfig):
        self.config = config
        self.steps: list[tuple[str, Any]] = []
        self.results: list[StepResult] = []

    def add_step(self, name: str, step_func) -> "Pipeline":
        """Register a step. Returns self for chaining."""
        self.steps.append((name, step_func))
        return self

    def run(self, initial_data: Any = None) -> list[StepResult]:
        """Execute all steps in sequence, passing data through."""
        logger.info(f"{'='*50}")
        logger.info(f"Pipeline: {self.config.name}")
        logger.info(f"{'='*50}")

        data = initial_data
        self.results = []

        for name, step_func in self.steps:
            start = time.perf_counter()
            data = step_func(data)
            elapsed = time.perf_counter() - start

            result = StepResult(
                step_name=name,
                duration_seconds=elapsed,
                output=data,
            )
            self.results.append(result)

        logger.info(f"{'='*50}")
        total_time = sum(r.duration_seconds for r in self.results)
        logger.info(f"Pipeline complete! Total time: {total_time:.4f}s")
        logger.info(f"{'='*50}")

        return self.results

    def summary(self) -> str:
        """Return a formatted summary of the pipeline run."""
        lines = [f"\nPipeline Summary: {self.config.name}", "-" * 40]
        for result in self.results:
            lines.append(
                f"  {result.step_name}: {result.duration_seconds:.4f}s"
            )
        total = sum(r.duration_seconds for r in self.results)
        lines.append("-" * 40)
        lines.append(f"  Total: {total:.4f}s")
        return "\n".join(lines)

Your pipeline foundation is solid. You have type-safe data models, a structural contract for steps, and an orchestrator that times and records everything. Next up, you will add decorators that inject logging, timing, and validation into any function without touching its core logic.

Add Timing and Logging Decorators

Your pipeline framework has a solid foundation with dataclasses for configuration, a Protocol for extensibility, and a Pipeline class for orchestration.

Now you need a way to add behavior to your pipeline steps without cluttering the core logic. Decorators let you wrap any function with extra functionality (timing, logging, validation) that runs automatically every time the function is called.

In this step, get ready to:
  • Build a timing decorator that measures how long any function takes to run.
  • Build a logging decorator that records when a function starts and finishes.
  • Stack both decorators together to see how they compose.
Build the @timed decorator

In production ML systems, knowing how long each step takes is critical for spotting bottlenecks. A @timed decorator wraps any function and automatically prints how long it took to execute.

  • In pipeline.py, find the comment that says # --- Pipeline orchestrator ---.
  • Add the following code directly above that comment:
# --- Decorators for cross-cutting concerns ---

def timed(func):
    """Decorator that measures and prints execution time."""

    @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:.4f}s")
        return result

    return wrapper

What does this code do?

This decorator uses the classic wrapper pattern. It defines an inner wrapper function that runs instead of the original function.

  • time.perf_counter() captures a high-resolution timestamp before and after the function runs.
  • functools.wraps(func) preserves the original function's name and docstring so debugging tools still work correctly.
  • The decorator returns the original result unchanged. It only adds a side effect (logging the elapsed time).
  • You won't see a change yet. The next decorator builds on this same pattern, and you'll test both together in a moment.
Build the @log_step decorator

Timing tells you how long a step took, but logging tells you which step is running and what it produced. The @log_step decorator prints entry and exit messages for any function it wraps.

  • In pipeline.py, add the following code directly below the timed function (still above the # --- Pipeline orchestrator --- comment):
def log_step(func):
    """Decorator that logs step entry and exit."""

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f">>> Starting: {func.__name__}")
        result = func(*args, **kwargs)
        result_type = type(result).__name__
        logger.info(f"<<< Finished: {func.__name__} -> {result_type}")
        return result

    return wrapper

What does this code do?

This follows the same wrapper pattern as @timed, but with different side effects.

  • Before calling the original function, it logs a >>> Starting: message with the function name.
  • After the function completes, it logs a <<< Finished: message showing both the function name and the type of value it returned.
  • type(result).__name__ gives a human-readable label like dict or list instead of a raw Python type object.
  • Save pipeline.py.
Stack both decorators to see them compose

The real power of decorators is that they compose. You can stack multiple decorators on a single function, and each one adds its own behavior independently.

  • Test both decorators together by running this command in your terminal:
python3 -c "
from pipeline import timed, log_step

@timed
@log_step
def sample_step(data):
    return [x * 2 for x in data]

sample_step([1, 2, 3])
"

What does stacking decorators mean?

When you write @timed above @log_step, Python applies them from bottom to top. First @log_step wraps sample_step, then @timed wraps that combined result.

The call order goes outside-in: timed starts the timer, then log_step logs entry, then sample_step runs, then log_step logs exit, then timed prints the elapsed time.

You should see output similar to this in your terminal:

>>> Starting: sample_step
<<< Finished: sample_step -> list
  [sample_step] completed in 0.0000s

Not seeing the output?

  • Make sure you saved pipeline.py before running the test command. Python imports the saved version, not what's in your editor.
  • Check that your terminal is in the project folder (the one containing pipeline.py). If you're in a different directory, the import will fail.
  • If you see ModuleNotFoundError, make sure your virtual environment is still activated (you should see (venv) at the start of your terminal prompt).

Help me debug why my decorator test isn't producing output.

Why this approach over print statements?

You could manually add timing and logging to every function, but that clutters the core logic. Decorators let you add these concerns once and apply them anywhere with a single line.

In a production ML system with dozens of pipeline steps, this pattern keeps each function focused on its actual job. The infrastructure behavior lives in the decorators, not scattered across your codebase.

✔️ Awesome, I've got everything!

Great. Double check you've saved pipeline.py with both decorators in place.

ⓧ I'd like to double check the full code

Here's the full pipeline.py file as it should look now:

"""Production ML Pipeline Framework using Advanced Python patterns."""

import time
import functools
import logging
from dataclasses import dataclass, field
from typing import Any, Protocol, runtime_checkable

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)


# --- Dataclasses for clean data modeling ---

@dataclass
class PipelineConfig:
    """Configuration for a pipeline run."""

    name: str
    random_seed: int = 42
    test_size: float = 0.2
    verbose: bool = True


@dataclass
class StepResult:
    """Result of a pipeline step execution."""

    step_name: str
    duration_seconds: float
    output: Any
    metadata: dict[str, Any] = field(default_factory=dict)


# --- Protocol for extensibility ---

@runtime_checkable
class PipelineStep(Protocol):
    """Contract that all pipeline steps must satisfy."""

    @property
    def name(self) -> str: ...

    def execute(self, data: Any) -> Any: ...


# --- Decorators for cross-cutting concerns ---

def timed(func):
    """Decorator that measures and prints execution time."""

    @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:.4f}s")
        return result

    return wrapper


def log_step(func):
    """Decorator that logs step entry and exit."""

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f">>> Starting: {func.__name__}")
        result = func(*args, **kwargs)
        result_type = type(result).__name__
        logger.info(f"<<< Finished: {func.__name__} -> {result_type}")
        return result

    return wrapper


# --- Pipeline orchestrator ---

class Pipeline:
    """Orchestrates pipeline step execution."""

    def __init__(self, config: PipelineConfig):
        self.config = config
        self.steps: list[tuple[str, Any]] = []
        self.results: list[StepResult] = []

    def add_step(self, name: str, step_func) -> "Pipeline":
        """Register a step. Returns self for chaining."""
        self.steps.append((name, step_func))
        return self

    def run(self, initial_data: Any = None) -> list[StepResult]:
        """Execute all steps in sequence, passing data through."""
        logger.info(f"{'='*50}")
        logger.info(f"Pipeline: {self.config.name}")
        logger.info(f"{'='*50}")

        data = initial_data
        self.results = []

        for name, step_func in self.steps:
            start = time.perf_counter()
            data = step_func(data)
            elapsed = time.perf_counter() - start

            result = StepResult(
                step_name=name,
                duration_seconds=elapsed,
                output=data,
            )
            self.results.append(result)

        logger.info(f"{'='*50}")
        total_time = sum(r.duration_seconds for r in self.results)
        logger.info(f"Pipeline complete! Total time: {total_time:.4f}s")
        logger.info(f"{'='*50}")

        return self.results

    def summary(self) -> str:
        """Return a formatted summary of the pipeline run."""
        lines = [f"\nPipeline Summary: {self.config.name}", "-" * 40]
        for result in self.results:
            lines.append(
                f"  {result.step_name}: {result.duration_seconds:.4f}s"
            )
        total = sum(r.duration_seconds for r in self.results)
        lines.append("-" * 40)
        lines.append(f"  Total: {total:.4f}s")
        return "\n".join(lines)

Your decorators are working and composing cleanly. Next up, you'll build validation and caching decorators, then combine everything into a single @pipeline_step decorator that does it all.

Add Validation and Caching Decorators

Your pipeline already knows how to time steps and log their entry and exit. That covers observability. But what about catching bad data before it causes cryptic errors deep inside your ML logic?

In this step, you'll build decorators that guard your pipeline inputs and combine all your cross-cutting concerns into a single convenient decorator. You'll also learn about Python's built-in caching decorator for expensive computations.

In this step, get ready to:
  • Build a @validate_input decorator that checks data types before processing.
  • Learn about functools.lru_cache for caching expensive computations.
  • Create a @pipeline_step decorator that combines timing and logging in one.
Build the @validate_input decorator factory

The @timed and @log_step decorators you built earlier are simple decorators. They take a function and return a wrapped function. But what if you need to configure the decorator itself?

A decorator factory is a function that returns a decorator. It adds a layer of configuration. You call it with parameters, and it gives you back a customized decorator.

  • In pipeline.py, find the log_step decorator (the last decorator before the # --- Pipeline orchestrator --- comment).
  • Add the following validate_input decorator factory below log_step:
def validate_input(expected_type: type):
    """Decorator factory that validates the first argument's type."""

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Skip 'self' if it's a method
            data_arg = args[1] if len(args) > 1 else args[0] if args else None
            if data_arg is not None and not isinstance(data_arg, expected_type):
                raise TypeError(
                    f"{func.__name__} expected {expected_type.__name__}, "
                    f"got {type(data_arg).__name__}"
                )
            return func(*args, **kwargs)

        return wrapper

    return decorator

What does this code do?

This decorator has three nested layers, which is what makes it a "factory":

  • validate_input(expected_type) is the factory. You call it with the type you expect (like dict or list).
  • decorator(func) is the actual decorator that wraps your function.
  • wrapper(*args, **kwargs) runs the type check before calling the original function. If the type is wrong, it raises a TypeError with a clear message.

The args[1] if len(args) > 1 logic handles methods where self is the first argument. For plain functions, it grabs args[0] directly.

  • Save pipeline.py.

You won't see a change yet. The next decorator builds on this, and you'll test everything together shortly.

Learn about functools.lru_cache for caching

Python's standard library already includes a powerful caching decorator. Before building your own, it's worth knowing what's available out of the box.

What is functools.lru_cache?

functools.lru_cache is a built-in memoization decorator. It stores the results of function calls so that repeated calls with the same arguments return instantly from cache instead of recomputing.

In ML, this is useful for expensive computations like hyperparameter searches. If you search the same parameter combination twice, the cached version returns immediately. The maxsize parameter controls how many recent results to store (defaults to 128).

You already have functools imported in pipeline.py. In the next step, you'll see where caching fits alongside the decorators you're building here.

Create the @pipeline_step combined decorator

Stacking multiple decorators on every function gets repetitive. A combined decorator bundles logging and timing into a single clean annotation.

  • In pipeline.py, add the following pipeline_step decorator directly below the validate_input function you just added:
def pipeline_step(func):
    """Combined decorator: logging + timing + wraps."""

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f">>> Starting: {func.__name__}")
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        result_type = type(result).__name__
        logger.info(
            f"<<< Finished: {func.__name__} -> {result_type} ({elapsed:.4f}s)"
        )
        return result

    return wrapper

What does this code do?

Instead of stacking @timed and @log_step on every function, @pipeline_step does both in one pass:

  • Logs the step name on entry (>>> Starting:).
  • Measures elapsed time with time.perf_counter().
  • Logs the result type and duration on exit (<<< Finished:).

This is the decorator you'll use on every step in your training pipeline. One annotation gives you full observability with zero clutter in the business logic.

  • Save pipeline.py.
  • Verify everything loads without errors by running this in your terminal:
python3 -c "from pipeline import validate_input, pipeline_step; print('All decorators loaded!')"

You should see All decorators loaded! printed with no errors.

Seeing a SyntaxError or ImportError?

  • Check that validate_input has three levels of def (factory, decorator, wrapper). A missing level causes an import error.
  • Make sure both new functions are placed between the log_step decorator and the # --- Pipeline orchestrator --- comment. Indentation matters.
  • Confirm your virtual environment is still active. You should see (venv) at the start of your terminal prompt.

Help me debug a SyntaxError when importing my decorators from pipeline.py

Why a combined decorator instead of stacking?

Stacking @timed and @log_step works, but each decorator wraps the function separately. That means two layers of function calls and two separate log entries for timing vs. completion.

The combined @pipeline_step produces a single clean log line with both the result type and the duration. One wrap, one log entry, less noise in your output.

✔️ Awesome, I've got everything!

Great. Double check you've saved pipeline.py and your import test printed successfully.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py with all decorators added:

"""Production ML Pipeline Framework using Advanced Python patterns."""

import time
import functools
import logging
from dataclasses import dataclass, field
from typing import Any, Protocol, runtime_checkable

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)


# --- Dataclasses for clean data modeling ---

@dataclass
class PipelineConfig:
    """Configuration for a pipeline run."""

    name: str
    random_seed: int = 42
    test_size: float = 0.2
    verbose: bool = True


@dataclass
class StepResult:
    """Result of a pipeline step execution."""

    step_name: str
    duration_seconds: float
    output: Any
    metadata: dict[str, Any] = field(default_factory=dict)


# --- Protocol for extensibility ---

@runtime_checkable
class PipelineStep(Protocol):
    """Contract that all pipeline steps must satisfy."""

    @property
    def name(self) -> str: ...

    def execute(self, data: Any) -> Any: ...


# --- Decorators for cross-cutting concerns ---

def timed(func):
    """Decorator that measures and prints execution time."""

    @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:.4f}s")
        return result

    return wrapper


def log_step(func):
    """Decorator that logs step entry and exit."""

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f">>> Starting: {func.__name__}")
        result = func(*args, **kwargs)
        result_type = type(result).__name__
        logger.info(f"<<< Finished: {func.__name__} -> {result_type}")
        return result

    return wrapper


def validate_input(expected_type: type):
    """Decorator factory that validates the first argument's type."""

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Skip 'self' if it's a method
            data_arg = args[1] if len(args) > 1 else args[0] if args else None
            if data_arg is not None and not isinstance(data_arg, expected_type):
                raise TypeError(
                    f"{func.__name__} expected {expected_type.__name__}, "
                    f"got {type(data_arg).__name__}"
                )
            return func(*args, **kwargs)

        return wrapper

    return decorator


def pipeline_step(func):
    """Combined decorator: logging + timing + wraps."""

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f">>> Starting: {func.__name__}")
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        result_type = type(result).__name__
        logger.info(
            f"<<< Finished: {func.__name__} -> {result_type} ({elapsed:.4f}s)"
        )
        return result

    return wrapper


# --- Pipeline orchestrator ---

class Pipeline:
    """Orchestrates pipeline step execution."""

    def __init__(self, config: PipelineConfig):
        self.config = config
        self.steps: list[tuple[str, Any]] = []
        self.results: list[StepResult] = []

    def add_step(self, name: str, step_func) -> "Pipeline":
        """Register a step. Returns self for chaining."""
        self.steps.append((name, step_func))
        return self

    def run(self, initial_data: Any = None) -> list[StepResult]:
        """Execute all steps in sequence, passing data through."""
        logger.info(f"{'='*50}")
        logger.info(f"Pipeline: {self.config.name}")
        logger.info(f"{'='*50}")

        data = initial_data
        self.results = []

        for name, step_func in self.steps:
            start = time.perf_counter()
            data = step_func(data)
            elapsed = time.perf_counter() - start

            result = StepResult(
                step_name=name,
                duration_seconds=elapsed,
                output=data,
            )
            self.results.append(result)

        logger.info(f"{'='*50}")
        total_time = sum(r.duration_seconds for r in self.results)
        logger.info(f"Pipeline complete! Total time: {total_time:.4f}s")
        logger.info(f"{'='*50}")

        return self.results

    def summary(self) -> str:
        """Return a formatted summary of the pipeline run."""
        lines = [f"\nPipeline Summary: {self.config.name}", "-" * 40]
        for result in self.results:
            lines.append(
                f"  {result.step_name}: {result.duration_seconds:.4f}s"
            )
        total = sum(r.duration_seconds for r in self.results)
        lines.append("-" * 40)
        lines.append(f"  Total: {total:.4f}s")
        return "\n".join(lines)

Your pipeline framework now has input validation and a clean combined decorator. Next up, you'll put all of this to work by writing train.py and training a real ML model through your pipeline.

Train a Model Using Your Pipeline

Your pipeline framework is complete. It has decorators for timing, logging, and validation. It has a Pipeline class that orchestrates steps in sequence.

Now it is time to put the framework to work. You will write a training script that loads a real dataset, preprocesses features, trains a classifier, and evaluates accuracy. Every step function gets decorated with your @pipeline_step and @validate_input decorators so the heavy cross-cutting work happens automatically.

In this step, get ready to:
  • Write the imports and file setup for train.py.
  • Implement five decorated pipeline step functions for loading, splitting, scaling, training, and evaluating.
  • Run the full pipeline and observe decorator-powered logging, timing, and validation in the terminal.
Set up train.py with imports

Your training script needs two categories of imports: external ML libraries and your own pipeline framework.

  • Switch back to train.py in Cursor (which is currently empty) and paste the following imports:
"""Train a classification model using the pipeline framework."""

import numpy as np
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, classification_report

from pipeline import (
    Pipeline,
    PipelineConfig,
    pipeline_step,
    validate_input,
    logger,
)

What do these imports do?

The first group pulls in scikit-learn utilities: a built-in dataset, a train/test splitter, a feature scaler, a classifier, and evaluation metrics.

The second group imports from your own pipeline.py module. You get the Pipeline orchestrator, PipelineConfig dataclass, both decorators, and the shared logger so every step logs through the same handler.

Implement the pipeline step functions

Each pipeline step is a plain function decorated with @pipeline_step (for automatic logging and timing) and @validate_input (for type checking). Data flows from one step to the next as a dictionary.

  • Below the imports in train.py, add the data loading and splitting functions:

# --- Pipeline steps using decorators ---

@pipeline_step
def load_data(_input):
    """Load the Iris dataset."""
    iris = load_iris()
    logger.info(f"  Loaded {len(iris.data)} samples, {iris.data.shape[1]} features")
    return {"X": iris.data, "y": iris.target, "feature_names": iris.feature_names}


@pipeline_step
@validate_input(dict)
def split_data(data: dict):
    """Split into train/test sets."""
    X_train, X_test, y_train, y_test = train_test_split(
        data["X"], data["y"], test_size=0.2, random_state=42
    )
    logger.info(f"  Train: {len(X_train)} samples, Test: {len(X_test)} samples")
    return {
        "X_train": X_train,
        "X_test": X_test,
        "y_train": y_train,
        "y_test": y_test,
        "feature_names": data["feature_names"],
    }

What does this code do?

The load_data function accepts _input (which will be None for the first step) and returns the Iris dataset as a dictionary. The underscore prefix signals that the parameter is intentionally unused.

The split_data function stacks two decorators: @pipeline_step handles logging and timing, while @validate_input(dict) confirms the incoming data is actually a dictionary before the function body runs.

  • Next, add the feature scaling, model training, and evaluation functions below split_data:

@pipeline_step
@validate_input(dict)
def scale_features(data: dict):
    """Standardize features using StandardScaler."""
    scaler = StandardScaler()
    data["X_train"] = scaler.fit_transform(data["X_train"])
    data["X_test"] = scaler.transform(data["X_test"])
    logger.info("  Features scaled to zero mean and unit variance")
    return data


@pipeline_step
@validate_input(dict)
def train_model(data: dict):
    """Train a Random Forest classifier."""
    model = RandomForestClassifier(n_estimators=100, random_state=42)
    model.fit(data["X_train"], data["y_train"])
    data["model"] = model
    logger.info(f"  Trained RandomForest with {model.n_estimators} trees")
    return data


@pipeline_step
@validate_input(dict)
def evaluate_model(data: dict):
    """Evaluate the trained model."""
    model = data["model"]
    y_pred = model.predict(data["X_test"])
    accuracy = accuracy_score(data["y_test"], y_pred)

    logger.info(f"  Accuracy: {accuracy:.4f}")
    logger.info(f"\n{classification_report(data['y_test'], y_pred)}")

    data["accuracy"] = accuracy
    data["predictions"] = y_pred
    return data

What does this code do?

Each function follows the same pattern: accept a dictionary, do one job, return the updated dictionary. The decorators handle all the cross-cutting concerns.

  • The scale_features function fits a StandardScaler on the training data and transforms both train and test sets.
  • The train_model function creates a RandomForestClassifier with 100 trees and fits it on the scaled training data.
  • The evaluate_model function generates predictions, computes accuracy, and prints a full classification report.

Why this approach works

Notice that none of these functions contain any timing, logging, or type-checking code. Those concerns live entirely in the decorators. If you need to add retry logic or caching later, you add another decorator instead of editing every function.

Write the main function and run the pipeline

The main() function wires everything together: it creates a config, builds the pipeline, registers all five steps, and kicks off execution.

  • Add the main() function and entry point at the bottom of train.py:

# --- Run the pipeline ---

def main():
    config = PipelineConfig(name="Iris Classification", random_seed=42)

    pipeline = Pipeline(config)
    pipeline.add_step("Load Data", load_data)
    pipeline.add_step("Split Data", split_data)
    pipeline.add_step("Scale Features", scale_features)
    pipeline.add_step("Train Model", train_model)
    pipeline.add_step("Evaluate Model", evaluate_model)

    results = pipeline.run()

    print(pipeline.summary())


if __name__ == "__main__":
    main()

What does this code do?

The PipelineConfig dataclass holds the pipeline name and settings. The add_step method registers each function with a human-readable label.

Calling pipeline.run() executes all steps in sequence, passing each step's output as the next step's input. Finally, pipeline.summary() prints a timing breakdown for every step.

The if __name__ == "__main__" guard ensures main() only runs when you execute the file directly, not when it is imported elsewhere.

  • Save train.py.
  • Run the pipeline by executing this command in your terminal:
python train.py
  • You should see logged output showing each step starting and finishing with its execution time, followed by the model's accuracy score and a classification report, then a pipeline summary with per-step timing.

Seeing an error when you run the script?

  • If you see a ModuleNotFoundError for pipeline, make sure you are running the command from the same folder that contains both pipeline.py and train.py. Python looks for local modules in the current working directory.
  • If you see a ModuleNotFoundError for sklearn, your virtual environment may not be active. Activate it first, then run the command again.
  • If you see a SyntaxError, double-check that every decorator line starts with @ and that your function definitions are properly indented.

Why isn't my pipeline script running?

✔️ Awesome, I've got everything!

Great. Double check you have saved train.py and that your terminal output shows timing for all five steps.

ⓧ I'd like to double check the full code

"""Train a classification model using the pipeline framework."""

import numpy as np
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, classification_report

from pipeline import (
    Pipeline,
    PipelineConfig,
    pipeline_step,
    validate_input,
    logger,
)


# --- Pipeline steps using decorators ---

@pipeline_step
def load_data(_input):
    """Load the Iris dataset."""
    iris = load_iris()
    logger.info(f"  Loaded {len(iris.data)} samples, {iris.data.shape[1]} features")
    return {"X": iris.data, "y": iris.target, "feature_names": iris.feature_names}


@pipeline_step
@validate_input(dict)
def split_data(data: dict):
    """Split into train/test sets."""
    X_train, X_test, y_train, y_test = train_test_split(
        data["X"], data["y"], test_size=0.2, random_state=42
    )
    logger.info(f"  Train: {len(X_train)} samples, Test: {len(X_test)} samples")
    return {
        "X_train": X_train,
        "X_test": X_test,
        "y_train": y_train,
        "y_test": y_test,
        "feature_names": data["feature_names"],
    }


@pipeline_step
@validate_input(dict)
def scale_features(data: dict):
    """Standardize features using StandardScaler."""
    scaler = StandardScaler()
    data["X_train"] = scaler.fit_transform(data["X_train"])
    data["X_test"] = scaler.transform(data["X_test"])
    logger.info("  Features scaled to zero mean and unit variance")
    return data


@pipeline_step
@validate_input(dict)
def train_model(data: dict):
    """Train a Random Forest classifier."""
    model = RandomForestClassifier(n_estimators=100, random_state=42)
    model.fit(data["X_train"], data["y_train"])
    data["model"] = model
    logger.info(f"  Trained RandomForest with {model.n_estimators} trees")
    return data


@pipeline_step
@validate_input(dict)
def evaluate_model(data: dict):
    """Evaluate the trained model."""
    model = data["model"]
    y_pred = model.predict(data["X_test"])
    accuracy = accuracy_score(data["y_test"], y_pred)

    logger.info(f"  Accuracy: {accuracy:.4f}")
    logger.info(f"\n{classification_report(data['y_test'], y_pred)}")

    data["accuracy"] = accuracy
    data["predictions"] = y_pred
    return data


# --- Run the pipeline ---

def main():
    config = PipelineConfig(name="Iris Classification", random_seed=42)

    pipeline = Pipeline(config)
    pipeline.add_step("Load Data", load_data)
    pipeline.add_step("Split Data", split_data)
    pipeline.add_step("Scale Features", scale_features)
    pipeline.add_step("Train Model", train_model)
    pipeline.add_step("Evaluate Model", evaluate_model)

    results = pipeline.run()

    print(pipeline.summary())


if __name__ == "__main__":
    main()

Your pipeline is alive. Five functions, zero manual logging calls inside them, and you get full observability for free. Next up, see if you can extend the framework even further in the Secret Mission.

Secret mission

Add a Retry Decorator with Exponential Backoff

Production ML pipelines deal with transient failures all the time. API timeouts, resource contention, temporary network blips. Can you build a decorator that automatically retries failed steps with exponential backoff, so your pipeline handles these gracefully without any manual intervention?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your files for future reference, pause the virtual environment, or delete everything entirely.

Resources you used:

  • Project folder containing pipeline.py and train.py.
  • Python virtual environment with scikit-learn and numpy installed.

Keep everything running

No action needed. Choose this if you're still actively building or plan to extend the pipeline with new steps and decorators.

  • Your virtual environment stays activated in the terminal session. If you close the terminal, reactivate it later with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).
  • Your project files remain on your Desktop, ready to edit in Cursor.

Pause - I'll come back to this later

Deactivate the virtual environment to free up your terminal session. Your files and installed packages stay intact for when you return.

  • Deactivate the virtual environment by running:
deactivate
  • Close Cursor and your terminal. When you're ready to resume, reopen the project folder in Cursor and reactivate the virtual environment.

Delete - I don't want to use this again

Remove all project files, the virtual environment, and installed packages. This frees up disk space completely.

  • Deactivate the virtual environment first by running:
deactivate
  • Navigate to your Desktop by running:
cd ~/Desktop
  • Delete the project folder (macOS/Linux) by running:
rm -rf ml-pipeline
  • On Windows, run this instead:
rmdir /s /q ml-pipeline
  • This removes pipeline.py, train.py, the virtual environment, and all installed packages in one step.

Nice Work!

Nice Work!

Nice work! You just built a production-grade ML pipeline framework from scratch using advanced Python patterns.

You've learned how to:

  • Build custom decorators that automatically time, log, and validate pipeline steps without cluttering core ML logic.
  • Use dataclasses and Protocol classes to create a type-safe, extensible pipeline framework.
  • Train and evaluate a Random Forest classifier using your custom framework with automated logging, timing, and validation.
  • Added a retry decorator with exponential backoff for fault-tolerant pipelines.

Ready to quiz yourself?