Build a Data Pipeline Framework

Build a modular ETL pipeline framework from scratch using Python OOP patterns.

Introduction

30 Second Summary

Every company that relies on data faces the same challenge. Raw, messy information needs to travel between systems, getting cleaned and transformed along the way.

In this project, you will build a modular data pipeline framework from scratch using pure Python. You will implement the same design patterns that production tools like Airflow and Prefect use under the hood.

What You'll Build

Run a single command and watch your framework extract sales data from a CSV file, filter out bad records, cast types, rename columns, and output clean JSON with a full execution log showing record counts at every stage.

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

  • A running pipeline that reads messy CSV data, drops invalid records, transforms column names and types, and produces clean JSON output with a step-by-step execution log showing exactly what happened at each stage.
  • Polymorphism in action: swap your data source from CSV to JSON by changing a single line of code, proving both sources work through the same interface.
  • A Factory pattern that builds complete pipelines from configuration dictionaries and a fluent method chaining API for assembling pipelines in a single expression.
  • Secret Mission: Add an Observer pattern so external hooks can react to pipeline events without modifying the core framework.

Are there any prerequisites?

You should be comfortable with basic Python concepts like functions, variables, loops, and dictionaries. No prior experience with object-oriented programming or design patterns is needed.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Committing to your goal up front grounds the project in your personal learning journey. It turns a tutorial into something you own.

Set Up the Project

You need a clean project structure before writing any code. Verifying your Python version first ensures the modern type hint syntax you'll use throughout the project works correctly.

In this step, you'll confirm Python 3.13 is ready, scaffold the project folder, and create your data files and initial data models.

In this step, get ready to:
  • Verify Python 3.13 is available on your machine.
  • Create the project directory with all required files.
  • Add sample data and define dataclass models.
Verify Python 3.13 is available

This project uses modern Python syntax like str | None and dict[str, Any] which require Python 3.10 or higher. You need Python 3.13 to match the project's target version.

  • Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), then type Terminal and press Enter.
  • Check your Python version by running this command:
python3 --version

What do you see in the output?

✔️ I see version 3.13 or higher

Your terminal shows something like Python 3.13.x. You're all set to continue.

ⓧ I see an older version

If your terminal shows a version lower than 3.13 (for example 3.9 or 3.11), you need to install Python 3.13 alongside it.

On macOS, the system Python is often outdated. Install the latest version from the official site:

  • Visit python.org/downloads and download the Python 3.13 installer for your operating system.
  • Run the installer and follow the prompts.
  • After installation, close and reopen your terminal, then run python3 --version again to confirm 3.13 appears.

macOS still showing the old version?

macOS may still point python3 at the old system install. Try running python3.13 --version directly. If that works, use python3.13 instead of python3 throughout this project.

ⓧ Command not found

Python is not installed on your machine yet. Install Python 3.13 from the official source:

  • Visit python.org/downloads and download the Python 3.13 installer for your operating system.
  • Run the installer and follow the prompts.
  • On Windows, make sure to check the Add python.exe to PATH checkbox during installation.
  • After installation, close and reopen your terminal, then run python3 --version (macOS/Linux) or python --version (Windows) to confirm 3.13 appears.
Create the project directory structure

Your pipeline framework needs a dedicated folder with separate files for each component. This keeps each OOP concept isolated and easy to navigate.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create the project folder and its subdirectories by running these commands:
mkdir -p data_pipeline/data data_pipeline/output
cd data_pipeline
  • Create the empty Python files that will hold each component by running:
touch models.py sources.py transforms.py loaders.py pipeline.py factory.py main.py
  • Confirm the structure exists by running:
ls

You should see data, output, and all seven .py files listed in the output.

Don't see all the files?

Make sure you ran the cd data_pipeline command before running touch. If you're inside the wrong folder, the files will be created elsewhere.

Run pwd to confirm you're inside ~/Desktop/data_pipeline.

help me find where my files were created

  • Open VS Code by pressing Cmd+Space (macOS) or the Windows key (Windows), typing VS Code, and pressing Enter.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop and select the data_pipeline folder, then click Open.

You should see all seven Python files and the data and output folders listed in the Explorer sidebar.

Create sample data and data models

Your pipeline needs data to process. You'll create a CSV file with messy sales records and a JSON file with event data. Both intentionally include missing values so the pipeline has something to filter later.

  • In VS Code, open data/sales.csv from the Explorer sidebar by clicking on the data folder, then clicking sales.csv.
  • Paste the following sales data into the file:
id,product,amount,quantity,region
1,Widget A,29.99,10,North
2,Widget B,,5,South
3,Gadget C,49.99,3,East
4,,15.00,7,West
5,Widget A,29.99,,North
6,Gadget D,99.99,2,South
7,Widget B,19.99,12,East
8,Gadget C,,1,North

Why is this data messy on purpose?

Rows 2, 4, 5, and 8 have empty values in the amount, product, or quantity columns. Your pipeline transforms will filter these out later, demonstrating how real ETL pipelines handle dirty data.

  • Save data/sales.csv.
  • Now open data/events.json from the Explorer sidebar and paste the following event data:
[
  {"event_type": "page_view", "timestamp": "2026-07-01T10:00:00", "user_id": "u001", "page": "/home"},
  {"event_type": "click", "timestamp": "2026-07-01T10:05:00", "user_id": "u002", "page": "/products"},
  {"event_type": null, "timestamp": "2026-07-01T10:10:00", "user_id": "u003", "page": "/about"},
  {"event_type": "purchase", "timestamp": null, "user_id": "u001", "page": "/checkout"},
  {"event_type": "page_view", "timestamp": "2026-07-01T10:20:00", "user_id": "u004", "page": "/home"},
  {"event_type": "click", "timestamp": "2026-07-01T10:25:00", "user_id": "u002", "page": "/cart"}
]

What does this data do?

This is a list of website events. Objects 3 and 4 have null values in key fields. Later, you'll build a second pipeline that processes this JSON source, proving the same framework handles multiple data formats.

  • Save data/events.json.
  • Verify both data files exist by running this in your terminal:
ls data/

You should see events.json and sales.csv listed.

Now you'll define the data structures your pipeline will use to pass records between stages. Dataclasses give you clean, typed containers without boilerplate.

  • Open models.py in VS Code and paste the following code to define the Record class:
from dataclasses import dataclass, field
from typing import Any


@dataclass
class Record:
    """Represents a single data record in the pipeline."""
    data: dict[str, Any]

    def get(self, key: str, default: Any = None) -> Any:
        return self.data.get(key, default)

    def __getitem__(self, key: str) -> Any:
        return self.data[key]

    def __setitem__(self, key: str, value: Any) -> None:
        self.data[key] = value

What does this code do?

The @dataclass decorator auto-generates __init__ and __repr__ methods from the type-annotated fields. No manual constructor needed.

The Record class wraps a plain dictionary to demonstrate encapsulation. External code accesses data through record.get() and record["key"] instead of touching the raw dict directly.

  • Save models.py and verify the Record class works by running this in your terminal:
python3 -c "from models import Record; r = Record(data={'product': 'Widget', 'price': 9.99}); print(r)"

You should see Record(data={'product': 'Widget', 'price': 9.99}) printed. The dataclass generated that __repr__ for free.

Seeing a ModuleNotFoundError?

Make sure your terminal is inside the data_pipeline folder. Run pwd to check. If you're in a different directory, run cd ~/Desktop/data_pipeline and try again.

help me fix this ModuleNotFoundError when importing Record

  • Back in models.py, add the following code below the Record class to define StepResult and PipelineResult:

@dataclass
class StepResult:
    """Result of a single pipeline step execution."""
    step_name: str
    records_in: int
    records_out: int
    status: str
    error: str | None = None


@dataclass
class PipelineResult:
    """Final result of a complete pipeline run."""
    records_in: int
    records_out: int
    success: bool
    errors: list[str] = field(default_factory=list)

    @property
    def drop_rate(self) -> float:
        if self.records_in == 0:
            return 0.0
        return (self.records_in - self.records_out) / self.records_in

What does this code do?

StepResult tracks what happened at each stage: how many records went in, how many came out, and whether it succeeded. PipelineResult summarizes the entire run.

The drop_rate property calculates what percentage of records were filtered out. The @property decorator lets you access it like an attribute (result.drop_rate) while computing the value on the fly.

The str | None syntax is modern Python union types. field(default_factory=list) safely provides a new empty list for each instance instead of sharing one mutable default across all instances.

  • Save models.py.

Before you run the next command, what do you think drop_rate will return if 10 records go in and 8 come out?

  • Test the complete models file by running:
python3 -c "from models import PipelineResult; r = PipelineResult(records_in=10, records_out=8, success=True); print(r.drop_rate)"

You should see 0.2 printed. That means 20% of records were dropped. Your data models are working.

Seeing a SyntaxError?

If you see an error about str | None, your Python version is too old. The union type syntax requires Python 3.10 or higher. Run python3 --version to double-check your version.

Also check that both dataclasses are separated by a single blank line and that the @dataclass decorator sits directly above each class definition with no extra lines between them.

help me debug this SyntaxError in models.py

✔️ Awesome, I've got everything!

Great. Double check you've saved models.py, data/sales.csv, and data/events.json.

ⓧ I'd like to double check the full code

Here's the complete models.py file. Compare yours against this:

from dataclasses import dataclass, field
from typing import Any


@dataclass
class Record:
    """Represents a single data record in the pipeline."""
    data: dict[str, Any]

    def get(self, key: str, default: Any = None) -> Any:
        return self.data.get(key, default)

    def __getitem__(self, key: str) -> Any:
        return self.data[key]

    def __setitem__(self, key: str, value: Any) -> None:
        self.data[key] = value
@dataclass
class StepResult:
    """Result of a single pipeline step execution."""
    step_name: str
    records_in: int
    records_out: int
    status: str
    error: str | None = None


@dataclass
class PipelineResult:
    """Final result of a complete pipeline run."""
    records_in: int
    records_out: int
    success: bool
    errors: list[str] = field(default_factory=list)

    @property
    def drop_rate(self) -> float:
        if self.records_in == 0:
            return 0.0
        return (self.records_in - self.records_out) / self.records_in

And here is the complete data/sales.csv:

id,product,amount,quantity,region
1,Widget A,29.99,10,North
2,Widget B,,5,South
3,Gadget C,49.99,3,East
4,,15.00,7,West
5,Widget A,29.99,,North
6,Gadget D,99.99,2,South
7,Widget B,19.99,12,East
8,Gadget C,,1,North

And the complete data/events.json:

[
  {"event_type": "page_view", "timestamp": "2026-07-01T10:00:00", "user_id": "u001", "page": "/home"},
  {"event_type": "click", "timestamp": "2026-07-01T10:05:00", "user_id": "u002", "page": "/products"},
  {"event_type": null, "timestamp": "2026-07-01T10:10:00", "user_id": "u003", "page": "/about"},
  {"event_type": "purchase", "timestamp": null, "user_id": "u001", "page": "/checkout"},
  {"event_type": "page_view", "timestamp": "2026-07-01T10:20:00", "user_id": "u004", "page": "/home"},
  {"event_type": "click", "timestamp": "2026-07-01T10:25:00", "user_id": "u002", "page": "/cart"}
]

Your foundation is in place. You have sample data with intentional gaps and typed data models that will carry records through every stage of the pipeline. Next up, you'll write a quick-and-dirty hardcoded pipeline to see data flow end-to-end, then feel the pain of trying to extend it.

Build a Hardcoded Pipeline

Your project structure is ready, your data models are defined, and sample data is waiting. Now it's time to make data actually flow.

In this step, you'll build the simplest possible working pipeline. It will be rigid and hardcoded to a single CSV file. Seeing real data flow from source to output gives you a tangible win to build on.

In this step, get ready to:
  • Write a hardcoded ETL function that extracts, filters, and loads CSV data.
  • Run it and confirm clean output appears.
  • Attempt to extend it for a second data source to feel the limits of this approach.
Write the hardcoded pipeline function

A data pipeline has three stages: extract (read raw data), transform (clean or reshape it), and load (write the result somewhere). Your first pipeline crams all three into a single function.

  • Open main.py in your editor and add the following code:
import csv
import json


def run_pipeline() -> None:
    # Extract from CSV
    records = []
    with open("data/sales.csv", "r", newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            records.append(dict(row))

    # Filter out rows where amount is empty
    filtered = [r for r in records if r.get("amount") not in ("", None)]

    # Load to JSON
    with open("output/sales_clean.json", "w") as f:
        json.dump(filtered, f, indent=2)

    print(f"Records processed: {len(filtered)}")


if __name__ == "__main__":
    run_pipeline()

What does this code do?

  • csv.DictReader reads each CSV row as a dictionary, using the header row as keys.
  • The filter uses a list comprehension to keep only rows where the amount field has a real value.
  • json.dump writes the cleaned records to a JSON file with readable indentation.
  • The if __name__ == "__main__" guard ensures the function only runs when you execute this file directly.
  • Save main.py.

Before you run this, how many records do you think will make it through the filter? The CSV has 8 rows, and the filter removes any row where amount is empty.

  • Run the pipeline by executing this command:
python3 main.py

You should see:

Records processed: 6

Two rows had empty amount fields (rows 2 and 8 in the CSV), so they were filtered out. Six clean records landed in output/sales_clean.json.

  • Confirm the output file was created by running:
ls output/

You should see sales_clean.json listed.

Seeing a FileNotFoundError?

  • Make sure you're running the command from inside the data_pipeline/ directory. Run ls and confirm you see main.py and the data/ folder.
  • Confirm the output/ directory exists. If not, create it with mkdir output.
  • Check that data/sales.csv exists and contains the sample data from Step 1.

Why is my pipeline throwing a FileNotFoundError?

✔️ Awesome, I've got everything!

Great. Make sure you've saved main.py and confirmed the output file exists.

ⓧ I'd like to double check the full code

Here is the complete main.py at this point:

import csv
import json


def run_pipeline() -> None:
    # Extract from CSV
    records = []
    with open("data/sales.csv", "r", newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            records.append(dict(row))

    # Filter out rows where amount is empty
    filtered = [r for r in records if r.get("amount") not in ("", None)]

    # Load to JSON
    with open("output/sales_clean.json", "w") as f:
        json.dump(filtered, f, indent=2)

    print(f"Records processed: {len(filtered)}")


if __name__ == "__main__":
    run_pipeline()
Try extending for a second data source

Your pipeline works for CSV sales data. But your project also has data/events.json sitting there with event data that also needs cleaning. How would you add JSON extraction to this function?

  • In main.py, look at your run_pipeline() function and imagine adding the JSON source below the existing CSV logic:
    # Now extract from JSON (duplicated logic!)
    json_records = []
    with open("data/events.json", "r") as f:
        data = json.load(f)
        for item in data:
            json_records.append(item)

    # Filter out rows where event_type is null (duplicated filter!)
    json_filtered = [
        r for r in json_records
        if r.get("event_type") not in (None, "null")
    ]

    # Load to another JSON file (duplicated load!)
    with open("output/events_clean.json", "w") as f:
        json.dump(json_filtered, f, indent=2)

    print(f"JSON records processed: {len(json_filtered)}")

What's the problem here?

Every new data source requires copy-pasting the same extract-filter-load pattern. You have duplicated extraction logic, duplicated filtering logic, and duplicated loading logic.

If you needed ten sources, this function would be hundreds of lines long with repetitive if/else branches. Changing how filtering works means editing every copy. This is exactly the problem that Object-Oriented Programming solves.

  • Do NOT save this duplicated code. Leave main.py as it was after the first substep (just the CSV pipeline). The next step replaces this entire approach with a clean OOP architecture.

You have a working pipeline and you've felt the limits of hardcoding everything into one function. Next up, you'll break this apart with Abstract Base Classes so adding a new source means writing one small class, not duplicating an entire block of logic.

Build Abstract Base Classes and Inheritance

In the last step, you felt the pain of the hardcoded approach firsthand. Adding a second data source meant duplicating extraction logic, mixing formats in the same function, and watching complexity spiral out of control.

Abstract Base Classes solve this by defining a contract that every data source must follow. Any class that inherits from the contract is guaranteed to have an extract() method. In this step, you'll build that contract layer for both sources and loaders.

In this step, get ready to:
  • Define the DataSource ABC and implement CSVSource and JSONSource.
  • Define the DataLoader ABC and implement JSONLoader and CSVLoader.
  • Rewrite main.py to use the new classes through a shared interface.
Define the DataSource abstract base class

An Abstract Base Class is a blueprint that says "any data source must have these methods." You cannot instantiate it directly. You can only create subclasses that fill in the blanks.

  • Create a new file called sources.py in your data_pipeline folder and add the following code:
import csv
import json
from abc import ABC, abstractmethod

from models import Record


class DataSource(ABC):
    """Abstract base class for all data sources."""

    def __init__(self, name: str) -> None:
        self._name = name
        self._records: list[Record] = []

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

    @abstractmethod
    def extract(self) -> list[Record]:
        """Extract records from the data source."""
        ...

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(name='{self._name}')"

What does this code do?

ABC from Python's abc module marks this class as abstract. You cannot create a DataSource() directly.

  • @abstractmethod on extract() forces every subclass to implement its own extraction logic. Without it, Python raises a TypeError.
  • self._name and self._records use Python's underscore convention to signal "private" state. External code accesses the name through the @property getter instead.

The ABC defines the contract. Now you need a concrete class that fulfills it.

  • Add the CSVSource class below DataSource in sources.py:
class CSVSource(DataSource):
    """Extracts data from a CSV file."""

    def __init__(self, name: str, filepath: str) -> None:
        super().__init__(name)
        self._filepath = filepath

    def extract(self) -> list[Record]:
        records: list[Record] = []
        with open(self._filepath, "r", newline="") as f:
            reader = csv.DictReader(f)
            for row in reader:
                records.append(Record(data=dict(row)))
        self._records = records
        return records

What does this code do?

  • CSVSource(DataSource) inherits from the ABC. Python enforces that it must implement extract().
  • super().__init__(name) calls the parent's constructor to set up _name and _records. The subclass only adds what's unique to it: _filepath.
  • Each row from the CSV is wrapped in a Record dataclass. This keeps the pipeline working with a consistent data type regardless of the source.
  • Save sources.py.
  • Test that CSVSource works by running this in your terminal:
python3 -c "from sources import CSVSource; s = CSVSource('test', 'data/sales.csv'); print(len(s.extract()), 'records')"

You should see 8 records printed. That confirms the ABC contract and the CSV implementation both work.

Seeing a ModuleNotFoundError?

  • Make sure you're running the command from inside the data_pipeline folder, where both sources.py and models.py live.
  • Check that from models import Record at the top of sources.py matches the filename exactly.

Help me debug my DataSource import error.

Now add the JSON equivalent. The power of the ABC is that JSONSource follows the same contract as CSVSource. Any code that works with a DataSource can use either one interchangeably.

  • Add the JSONSource class below CSVSource in sources.py:
class JSONSource(DataSource):
    """Extracts data from a JSON file."""

    def __init__(self, name: str, filepath: str) -> None:
        super().__init__(name)
        self._filepath = filepath

    def extract(self) -> list[Record]:
        records: list[Record] = []
        with open(self._filepath, "r") as f:
            data = json.load(f)
            if isinstance(data, list):
                for item in data:
                    records.append(Record(data=item))
            else:
                records.append(Record(data=data))
        self._records = records
        return records

What does this code do?

  • JSONSource handles both JSON arrays (a list of objects) and single JSON objects. This makes it flexible without exposing that complexity to the caller.
  • The caller doesn't care whether data came from CSV or JSON. Both return list[Record]. This is polymorphism in action.
  • Save sources.py.
  • Verify JSONSource works by running:
python3 -c "from sources import JSONSource; s = JSONSource('test', 'data/events.json'); print(len(s.extract()), 'records')"

You should see 6 records printed. Two different source types, one shared interface.

Seeing 0 records or a JSONDecodeError?

  • Open data/events.json and confirm it starts with [ and ends with ]. A missing bracket means invalid JSON.
  • Make sure the file path data/events.json matches the actual file location relative to where you run the command.

Help me debug my JSONSource extraction issue.

✔️ Awesome, I've got everything!

Great. Double check you've saved sources.py before moving on.

ⓧ I'd like to double check the full code

Your sources.py should contain these three blocks in order. Scroll up and confirm each is present:

  • Imports and the DataSource ABC (with __init__, name property, abstract extract(), and __repr__).
  • CSVSource class that reads CSV rows into Record objects.
  • JSONSource class that reads JSON data into Record objects.

Why use an ABC instead of just writing the classes?

Without the ABC, a developer could create a new source class and forget to implement extract(). The bug would only surface at runtime, deep in the pipeline. With the ABC, Python raises a TypeError the moment you try to instantiate the incomplete class.

This is the difference between a guideline and a contract. Production frameworks like Airflow use ABCs for exactly this reason.

Define the DataLoader abstract base class

Sources handle the "E" in ETL. Now you need the same pattern for the "L" (Load). A loader takes records and writes them to a destination.

  • Create a new file called loaders.py in your data_pipeline folder and add the DataLoader ABC:
import csv
import json
from abc import ABC, abstractmethod

from models import Record


class DataLoader(ABC):
    """Abstract base class for all data loaders."""

    def __init__(self, name: str) -> None:
        self._name = name

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

    @abstractmethod
    def load(self, records: list[Record]) -> int:
        """Load records to destination. Returns count of records loaded."""
        ...

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(name='{self._name}')"

What does this code do?

  • DataLoader mirrors the same pattern as DataSource: an ABC with a private _name, a @property getter, and one abstract method.
  • load() takes a list of Record objects and returns an int count of how many were loaded. This gives the pipeline a way to confirm the write succeeded.
  • Add the JSONLoader class below DataLoader in loaders.py:
class JSONLoader(DataLoader):
    """Loads records to a JSON file."""

    def __init__(self, name: str, filepath: str) -> None:
        super().__init__(name)
        self._filepath = filepath

    def load(self, records: list[Record]) -> int:
        data = [record.data for record in records]
        with open(self._filepath, "w") as f:
            json.dump(data, f, indent=2)
        return len(data)

What does this code do?

  • JSONLoader unwraps each Record back into its raw .data dictionary before writing. The file format is a detail the loader owns.
  • Returning len(data) gives the pipeline a concrete count it can log and verify against.
  • Save loaders.py and verify the JSONLoader works by running:
python3 -c "from sources import CSVSource; from loaders import JSONLoader; r = CSVSource('t', 'data/sales.csv').extract(); print('Loaded:', JSONLoader('t', 'output/test.json').load(r))"

You should see Loaded: 8 printed. The source extracts, the loader writes. Two ABCs, one clean interface.

Seeing a TypeError about abstract methods?

  • This means your JSONLoader is missing the load() method or has a typo in the method name. The method signature must exactly match the abstract definition: def load(self, records: list[Record]) -> int:.

Help me fix my DataLoader TypeError.

Now add the CSV loader for completeness. This gives you a second output format you can swap to later.

  • Add the CSVLoader class below JSONLoader in loaders.py:
class CSVLoader(DataLoader):
    """Loads records to a CSV file."""

    def __init__(self, name: str, filepath: str) -> None:
        super().__init__(name)
        self._filepath = filepath

    def load(self, records: list[Record]) -> int:
        if not records:
            return 0
        fieldnames = list(records[0].data.keys())
        with open(self._filepath, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            for record in records:
                writer.writerow(record.data)
        return len(records)

What does this code do?

  • CSVLoader infers column names from the first record's keys. This keeps it generic enough to handle any data shape.
  • The empty-records guard (if not records: return 0) prevents an IndexError when there's nothing to write.
  • Save loaders.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved loaders.py before moving on.

ⓧ I'd like to double check the full code

Your loaders.py should contain these three blocks in order. Scroll up and confirm each is present:

  • Imports and the DataLoader ABC (with __init__, name property, abstract load(), and __repr__).
  • JSONLoader class that writes records to a JSON file.
  • CSVLoader class that writes records to a CSV file.
Wire everything together in main.py

You now have two source classes and two loader classes, all built on clean ABC contracts. Time to replace the old hardcoded function with the new OOP approach.

  • Open main.py and replace its entire contents with this code:
from loaders import JSONLoader
from sources import CSVSource, JSONSource


def main() -> None:
    csv_source = CSVSource(name="sales_csv", filepath="data/sales.csv")
    csv_records = csv_source.extract()
    print(f"CSV extracted: {len(csv_records)} records")

    json_source = JSONSource(name="events_json", filepath="data/events.json")
    json_records = json_source.extract()
    print(f"JSON extracted: {len(json_records)} records")

    loader = JSONLoader(name="json_output", filepath="output/sales_clean.json")
    loaded = loader.load(csv_records)
    print(f"Loaded: {loaded} records")


if __name__ == "__main__":
    main()

What does this code do?

  • Both CSVSource and JSONSource are used through the same .extract() call. No if/else branches. No duplicated file-reading code.
  • The old run_pipeline() function is gone. Every responsibility now lives in its own class with a clear interface.
  • Adding a third source (database, API, Parquet) would mean writing one new class that inherits from DataSource. Zero changes to existing code.
  • Save main.py.

Before you run this, what do you expect to see? Think about how many records are in each data file.

  • Run your updated pipeline by executing:
python3 main.py

You should see three lines printed:

CSV extracted: 8 records
JSON extracted: 6 records
Loaded: 8 records

Both sources extract data through the same interface. The CSV file has 8 rows and the JSON file has 6 objects. The loader writes the CSV records to JSON output.

Seeing an ImportError?

  • Confirm all three files (main.py, sources.py, loaders.py) are in the same data_pipeline folder.
  • Check that your import lines match exactly: from sources import CSVSource, JSONSource and from loaders import JSONLoader.
  • Make sure your terminal's working directory is data_pipeline. Run ls to confirm you can see all your .py files.

Help me debug my import error in main.py.

Your pipeline can now read from any source and write to any destination through clean, interchangeable interfaces. Next up, you'll build the Transform layer to filter, rename, and type-cast records as they flow through the pipeline.

Implement the Strategy Pattern with Transforms

Your pipeline can extract data from CSV and JSON sources, then load records to a file. But there is no way to clean or reshape data between those two stages.

The Strategy pattern solves this by encapsulating each transformation algorithm in its own class. You can add, remove, or reorder transforms without touching extraction or loading code.

In this step, get ready to:
  • Define a Transform abstract base class and implement three concrete strategies.
  • Chain transforms in main.py to clean and reshape sales data.
  • Demonstrate swappability by changing transform parameters at runtime.
Define the Transform ABC and FilterNulls strategy

Just like DataSource defines a contract for extraction, you need an abstract base class that defines the contract for transformations. Every transform must accept a list of records and return a new list of records.

  • Open transforms.py in your code editor and add the Transform ABC by pasting this code:
from abc import ABC, abstractmethod

from models import Record


class Transform(ABC):
    """Abstract base class for all transforms (Strategy pattern)."""

    def __init__(self, name: str) -> None:
        self._name = name

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

    @abstractmethod
    def apply(self, records: list[Record]) -> list[Record]:
        """Apply the transformation to a list of records."""
        ...

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(name='{self._name}')"

What does this code do?

This is the same pattern you used for DataSource and DataLoader. The abstract apply() method forces every subclass to implement a records-in, records-out interface.

The underscore prefix on _name signals that external code should use the name property instead of accessing the attribute directly. This is Python's convention for encapsulation.

Now you need the first concrete strategy. FilterNulls removes records where specified fields are empty or null.

  • Add the FilterNulls class below the Transform class in transforms.py:
class FilterNulls(Transform):
    """Removes records where specified fields are null or empty."""

    def __init__(self, name: str, fields: list[str]) -> None:
        super().__init__(name)
        self._fields = fields

    def apply(self, records: list[Record]) -> list[Record]:
        result: list[Record] = []
        for record in records:
            if all(
                record.get(field) not in (None, "", "null", "None")
                for field in self._fields
            ):
                result.append(record)
        return result

What does this code do?

FilterNulls takes a list of field names to check. For each record, it uses all() to verify that every specified field has a real value. Records with empty, null, or missing fields get dropped.

This is the Strategy pattern in action. The filtering algorithm is encapsulated in its own class. You can swap which fields are checked by creating a different instance with different parameters.

  • Save transforms.py.
  • Confirm the code parses correctly by running this command:
python3 -c "from transforms import FilterNulls; print('FilterNulls loaded successfully')"

You should see FilterNulls loaded successfully printed with no errors.

Seeing an ImportError or SyntaxError?

  • Check that the from models import Record line is at the top of transforms.py.
  • Make sure you are running the command from inside the data_pipeline folder where both transforms.py and models.py live.

Help me debug this import error in transforms.py.

Add the RenameColumns and TypeCast strategies

The next strategy renames fields in each record using a mapping dictionary. This lets you standardize column names from messy source data.

  • Add the RenameColumns class below FilterNulls in transforms.py:
class RenameColumns(Transform):
    """Renames fields in each record."""

    def __init__(self, name: str, mapping: dict[str, str]) -> None:
        super().__init__(name)
        self._mapping = mapping

    def apply(self, records: list[Record]) -> list[Record]:
        result: list[Record] = []
        for record in records:
            new_data = {}
            for key, value in record.data.items():
                new_key = self._mapping.get(key, key)
                new_data[new_key] = value
            result.append(Record(data=new_data))
        return result

What does this code do?

RenameColumns loops through each record's fields. If a field name appears in the mapping dictionary, it gets renamed. Fields not in the mapping pass through unchanged.

The final strategy is TypeCast. It converts field values from strings to specific Python types like float or int.

  • Add the TypeCast class below RenameColumns in transforms.py:
class TypeCast(Transform):
    """Casts specified fields to given types."""

    def __init__(self, name: str, schema: dict[str, type]) -> None:
        super().__init__(name)
        self._schema = schema

    def apply(self, records: list[Record]) -> list[Record]:
        result: list[Record] = []
        for record in records:
            new_data = dict(record.data)
            for field_name, target_type in self._schema.items():
                if field_name in new_data:
                    try:
                        new_data[field_name] = target_type(new_data[field_name])
                    except (ValueError, TypeError):
                        new_data[field_name] = None
            result.append(Record(data=new_data))
        return result

What does this code do?

TypeCast takes a schema dictionary mapping field names to Python types. For each record, it tries to cast each field to its target type. If a cast fails (like converting an empty string to int), it sets the value to None instead of crashing.

The try/except block makes this strategy resilient to bad data. CSV files store everything as strings, so this step is critical for downstream calculations.

  • Save transforms.py.
  • Verify all three strategies load correctly by running this command:
python3 -c "from transforms import FilterNulls, RenameColumns, TypeCast; print('All transforms loaded successfully')"

You should see All transforms loaded successfully with no errors.

Seeing a NameError or IndentationError?

  • Make sure each class inherits from Transform with the correct spelling, for example class RenameColumns(Transform):.
  • Check that the method bodies are indented consistently. Mixing tabs and spaces causes silent indentation errors in Python.

Help me fix my transforms.py file.

✔️ Awesome, I've got everything!

Great. Make sure you have saved transforms.py before moving on.

ⓧ I'd like to double check the full code

Scroll up and confirm your transforms.py contains these four sections in order:

  • Imports and the Transform ABC with __init__, name property, abstract apply(), and __repr__.
  • FilterNulls class with _fields attribute and apply() that filters using all().
  • RenameColumns class with _mapping attribute and apply() that renames keys.
  • TypeCast class with _schema attribute and apply() that casts with try/except.
Chain transforms in main.py

With three strategies defined, you can now wire them into a sequence. Each transform takes the output of the previous one, forming a processing chain.

  • Replace the entire contents of main.py with this code that chains all three transforms:
from loaders import JSONLoader
from sources import CSVSource
from transforms import FilterNulls, RenameColumns, TypeCast


def main() -> None:
    # Extract records from CSV
    source = CSVSource(name="sales_csv", filepath="data/sales.csv")
    records = source.extract()
    print(f"Extracted: {len(records)} records")

    # Apply transforms in sequence
    filter_transform = FilterNulls(name="remove_empty", fields=["amount", "product"])
    records = filter_transform.apply(records)
    print(f"After FilterNulls: {len(records)} records")

    cast_transform = TypeCast(name="cast_types", schema={"amount": float, "quantity": int})
    records = cast_transform.apply(records)
    print(f"After TypeCast: {len(records)} records")

    rename_transform = RenameColumns(name="standardize", mapping={"product": "product_name"})
    records = rename_transform.apply(records)
    print(f"After RenameColumns: {len(records)} records")

    # Load clean records to JSON
    loader = JSONLoader(name="json_output", filepath="output/sales_clean.json")
    loaded = loader.load(records)
    print(f"Loaded: {loaded} records")


if __name__ == "__main__":
    main()

What does this code do?

Each transform reassigns the records variable with its output. FilterNulls drops rows where amount or product is empty. TypeCast converts amount to float and quantity to int. RenameColumns renames product to product_name.

The print statements after each stage let you see exactly how many records survive. This is the chain running: extract, filter, cast, rename, load.

  • Save main.py.

Before you run this, how many records do you think will survive FilterNulls? The CSV has 8 rows, and you are checking both the amount and product fields.

  • Run the pipeline by executing this command:
python3 main.py

You should see this output:

Extracted: 8 records
After FilterNulls: 5 records
After TypeCast: 5 records
After RenameColumns: 5 records
Loaded: 5 records

Three records were dropped by FilterNulls (rows 2, 4, and 8 from the CSV had empty amount or product values). TypeCast and RenameColumns reshape data without removing records.

Seeing a different record count or an error?

  • If you see 8 records after FilterNulls, check that the fields list includes both "amount" and "product".
  • If you get an ImportError, verify the import line reads from transforms import FilterNulls, RenameColumns, TypeCast exactly.

Help me debug my pipeline output.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py.

ⓧ I'd like to double check the full code

from loaders import JSONLoader
from sources import CSVSource
from transforms import FilterNulls, RenameColumns, TypeCast


def main() -> None:
    source = CSVSource(name="sales_csv", filepath="data/sales.csv")
    records = source.extract()
    print(f"Extracted: {len(records)} records")

    filter_transform = FilterNulls(name="remove_empty", fields=["amount", "product"])
    records = filter_transform.apply(records)
    print(f"After FilterNulls: {len(records)} records")

    cast_transform = TypeCast(name="cast_types", schema={"amount": float, "quantity": int})
    records = cast_transform.apply(records)
    print(f"After TypeCast: {len(records)} records")

    rename_transform = RenameColumns(name="standardize", mapping={"product": "product_name"})
    records = rename_transform.apply(records)
    print(f"After RenameColumns: {len(records)} records")

    loader = JSONLoader(name="json_output", filepath="output/sales_clean.json")
    loaded = loader.load(records)
    print(f"Loaded: {loaded} records")


if __name__ == "__main__":
    main()
See the Strategy pattern in action

The power of the Strategy pattern is that you can change behavior by swapping configuration, not by rewriting transform logic. Each class is a self-contained algorithm that knows nothing about the others.

  • In main.py, find this line:
filter_transform = FilterNulls(name="remove_empty", fields=["amount", "product"])
  • Change the fields list to only check "amount", so the line looks like this:
filter_transform = FilterNulls(name="remove_empty", fields=["amount"])
  • Save main.py and run it again:
python3 main.py

Now you see 6 records after FilterNulls instead of 5. By removing "product" from the fields list, the row with an empty product name now passes through. You changed the pipeline's behavior without modifying a single line inside the FilterNulls class.

Why does this matter?

In production, transform requirements change constantly. A business rule might add a new required field or relax a filter. With the Strategy pattern, you reconfigure at the call site. The algorithms themselves stay stable and tested.

  • Change the fields list back to ["amount", "product"] so your pipeline filters on both fields:
filter_transform = FilterNulls(name="remove_empty", fields=["amount", "product"])
  • Save main.py and run it one final time to confirm you are back to 5 records:
python3 main.py

You should see the original output: 8 extracted, 5 after FilterNulls, 5 after TypeCast, 5 after RenameColumns, and 5 loaded.

You now have three interchangeable transform algorithms that slot into any position in the chain. Next up, you will build a Pipeline orchestrator that composes sources, transforms, and loaders into a single object with method chaining and a Factory that constructs pipelines from configuration dictionaries.

Build the Pipeline Orchestrator and Factory

Your main.py currently wires every source, transform, and loader together by hand. Adding a second pipeline means duplicating all that orchestration logic. Every new data flow requires copy-pasting the same extract-transform-load sequence and managing errors manually.

A Pipeline class solves this using composition. It holds references to a source, transforms, and a loader, then orchestrates them in sequence. A Factory takes it further by building entire pipelines from configuration dictionaries. This is the architecture pattern used by tools like Airflow and Prefect.

In this step, get ready to:
  • Build a Pipeline class that composes components and orchestrates execution with method chaining.
  • Build a StepFactory that creates pipeline components from config dictionaries.
  • Run two pipelines and see full execution logs with record counts and drop rates.
Create the PipelineLog and Pipeline class

The Pipeline class needs a way to track what happened during execution. A PipelineLog dataclass collects results from each step and formats a readable summary.

  • Open pipeline.py and add the imports and PipelineLog class:
from dataclasses import dataclass, field

from loaders import DataLoader
from models import PipelineResult, Record, StepResult
from sources import DataSource
from transforms import Transform


@dataclass
class PipelineLog:
    """Tracks execution results for each pipeline step."""
    steps: list[StepResult] = field(default_factory=list)

    def add(self, result: StepResult) -> None:
        self.steps.append(result)

    def summary(self) -> str:
        lines = ["Pipeline Execution Summary", "=" * 40]
        for step in self.steps:
            status_icon = "OK" if step.status == "success" else "FAIL"
            lines.append(
                f"  [{status_icon}] {step.step_name}: "
                f"{step.records_in} in -> {step.records_out} out"
            )
            if step.error:
                lines.append(f"        Error: {step.error}")
        lines.append("=" * 40)
        return "\n".join(lines)

What does this code do?

PipelineLog is a dataclass that collects StepResult objects as the pipeline runs.

The summary() method formats a table showing each step's name, record counts, and whether it succeeded or failed. This gives you a full audit trail of every pipeline run.

  • Below the PipelineLog class, add the Pipeline class with its constructor and method-chaining setters:
class Pipeline:
    """Orchestrates data pipeline execution using composition."""

    def __init__(self, name: str) -> None:
        self._name = name
        self._source: DataSource | None = None
        self._transforms: list[Transform] = []
        self._loader: DataLoader | None = None
        self._log = PipelineLog()

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

    def set_source(self, source: DataSource) -> "Pipeline":
        """Set the data source. Returns self for method chaining."""
        self._source = source
        return self

    def add_transform(self, transform: Transform) -> "Pipeline":
        """Add a transform step. Returns self for method chaining."""
        self._transforms.append(transform)
        return self

    def set_loader(self, loader: DataLoader) -> "Pipeline":
        """Set the data loader. Returns self for method chaining."""
        self._loader = loader
        return self

What does this code do?

This is composition in action. The Pipeline doesn't inherit from DataSource or Transform. It holds references to them as private attributes.

Each setter returns self, which enables method chaining. That means you can write Pipeline('name').set_source(...).add_transform(...).set_loader(...) as a single fluent expression.

  • Save pipeline.py.
  • Verify the class imports correctly by running this command:
python3 -c "from pipeline import Pipeline; p = Pipeline('test'); print(p.name)"

You should see test printed. The class shell works.

Seeing an ImportError?

Make sure you are running the command from inside your data_pipeline folder. Check that models.py, sources.py, transforms.py, and loaders.py all exist in the same directory.

Help me debug my Pipeline import error

Now you need the run() method that orchestrates the full extract-transform-load sequence. It validates that a source and loader are set, then executes each stage with error handling.

  • Inside the Pipeline class, below set_loader(), add the first part of the run() method handling validation and the Extract phase:
    def run(self) -> PipelineResult:
        """Execute the pipeline end-to-end."""
        if not self._source:
            raise ValueError("Pipeline has no data source configured")
        if not self._loader:
            raise ValueError("Pipeline has no data loader configured")

        errors: list[str] = []

        # Extract
        try:
            records = self._source.extract()
            self._log.add(StepResult(
                step_name=f"Extract: {self._source.name}",
                records_in=0,
                records_out=len(records),
                status="success"
            ))
        except Exception as e:
            self._log.add(StepResult(
                step_name=f"Extract: {self._source.name}",
                records_in=0,
                records_out=0,
                status="failed",
                error=str(e)
            ))
            return PipelineResult(
                records_in=0, records_out=0,
                success=False, errors=[str(e)]
            )

        records_in = len(records)

What does this code do?

The method first validates that both a source and loader are configured. Then it tries to extract records. If extraction fails, it logs the failure and immediately returns a failed PipelineResult.

The records_in variable captures the starting count so the final result can calculate a drop rate.

  • Continuing inside the same run() method, add the Transform loop directly below records_in = len(records):
        # Transform
        for transform in self._transforms:
            count_before = len(records)
            try:
                records = transform.apply(records)
                self._log.add(StepResult(
                    step_name=f"Transform: {transform.name}",
                    records_in=count_before,
                    records_out=len(records),
                    status="success"
                ))
            except Exception as e:
                errors.append(f"{transform.name}: {str(e)}")
                self._log.add(StepResult(
                    step_name=f"Transform: {transform.name}",
                    records_in=count_before,
                    records_out=len(records),
                    status="failed",
                    error=str(e)
                ))

What does this code do?

This loops through every transform in the list, applying each one in sequence. Unlike the Extract phase, a failed transform does not abort the pipeline. It logs the error and continues with whatever records survived.

Each iteration tracks count_before so the log shows how many records each transform dropped.

  • Save pipeline.py and confirm it still imports cleanly:
python3 -c "from pipeline import Pipeline; print('OK')"

You should see OK. This confirms your method syntax is valid so far.

  • Back in pipeline.py, add the Load phase, the final return, and the get_log() method directly below the Transform loop:
        # Load
        try:
            loaded_count = self._loader.load(records)
            self._log.add(StepResult(
                step_name=f"Load: {self._loader.name}",
                records_in=len(records),
                records_out=loaded_count,
                status="success"
            ))
        except Exception as e:
            errors.append(f"Load failed: {str(e)}")
            self._log.add(StepResult(
                step_name=f"Load: {self._loader.name}",
                records_in=len(records),
                records_out=0,
                status="failed",
                error=str(e)
            ))
            return PipelineResult(
                records_in=records_in, records_out=0,
                success=False, errors=errors
            )

        return PipelineResult(
            records_in=records_in,
            records_out=loaded_count,
            success=len(errors) == 0,
            errors=errors
        )

    def get_log(self) -> PipelineLog:
        """Return the execution log."""
        return self._log

What does this code do?

The Load phase calls the loader's load() method. If it fails, the pipeline returns immediately with a failure result. If everything succeeds, it returns a PipelineResult with the full counts.

The get_log() method exposes the internal log so callers can print the execution summary after a run.

  • Save pipeline.py and verify one more time:
python3 -c "from pipeline import Pipeline; print('Pipeline ready')"

You should see Pipeline ready. The entire Pipeline class is complete.

✔️ Awesome, I've got everything!

Great. Make sure you have saved pipeline.py before moving on.

ⓧ I'd like to double check the full code

Your pipeline.py should contain these sections in order. Scroll up and confirm each is present:

  • Imports from dataclasses, loaders, models, sources, and transforms.
  • The PipelineLog dataclass with steps list, add() method, and summary() method.
  • The Pipeline class with __init__, name property, set_source(), add_transform(), and set_loader() chaining methods.
  • The run() method with Extract, Transform, and Load phases, each wrapped in try/except.
  • The get_log() method returning self._log.
Build the StepFactory

The Factory pattern lets you create objects from configuration data instead of manually instantiating them. Instead of writing CSVSource(name='x', filepath='y') everywhere, you pass a dictionary and the factory figures out which class to use.

  • Open factory.py and add the imports and class shell with registries:
from typing import Any

from loaders import CSVLoader, DataLoader, JSONLoader
from sources import CSVSource, DataSource, JSONSource
from transforms import FilterNulls, RenameColumns, Transform, TypeCast


class StepFactory:
    """Factory for creating pipeline components from configuration."""

    _source_registry: dict[str, type[DataSource]] = {
        "csv": CSVSource,
        "json": JSONSource,
    }

    _transform_registry: dict[str, type[Transform]] = {
        "filter_nulls": FilterNulls,
        "rename_columns": RenameColumns,
        "type_cast": TypeCast,
    }

    _loader_registry: dict[str, type[DataLoader]] = {
        "json": JSONLoader,
        "csv": CSVLoader,
    }

What are class-level registries?

The _source_registry, _transform_registry, and _loader_registry are class variables, not instance variables. They belong to the class itself, not to any particular object created from it.

Each registry maps a simple string key (like "csv") to the actual class it corresponds to (like CSVSource). This is what lets the factory turn config dicts into real objects.

  • Below the registries, add the three create_* class methods:
    @classmethod
    def create_source(cls, config: dict[str, Any]) -> DataSource:
        """Create a data source from a config dict."""
        config = dict(config)
        source_type = config.pop("type")
        if source_type not in cls._source_registry:
            raise ValueError(f"Unknown source type: {source_type}")
        source_class = cls._source_registry[source_type]
        return source_class(**config)

    @classmethod
    def create_transform(cls, config: dict[str, Any]) -> Transform:
        """Create a transform from a config dict."""
        config = dict(config)
        transform_type = config.pop("type")
        if transform_type not in cls._transform_registry:
            raise ValueError(f"Unknown transform type: {transform_type}")
        transform_class = cls._transform_registry[transform_type]
        return transform_class(**config)

    @classmethod
    def create_loader(cls, config: dict[str, Any]) -> DataLoader:
        """Create a data loader from a config dict."""
        config = dict(config)
        loader_type = config.pop("type")
        if loader_type not in cls._loader_registry:
            raise ValueError(f"Unknown loader type: {loader_type}")
        loader_class = cls._loader_registry[loader_type]
        return loader_class(**config)

Why copy the config with dict(config)?

Each method starts with config = dict(config) to make a shallow copy before calling .pop("type"). Without the copy, popping would mutate the caller's original dictionary. This is a defensive programming pattern that prevents subtle bugs.

The remaining keys after popping "type" are passed directly as keyword arguments with **config. The dict keys must match the constructor parameter names of the target class.

  • Save factory.py and verify it imports:
python3 -c "from factory import StepFactory; print('Factory ready')"

You should see Factory ready.

Seeing a NameError or ImportError?

Check that your import lines at the top of factory.py match the class names exactly. CSVLoader and JSONLoader come from loaders. CSVSource and JSONSource come from sources.

Help me debug my StepFactory import error

The factory works, but it only creates from the built-in registries. The Open/Closed Principle says classes should be open for extension but closed for modification. Register methods let new types be added without editing the factory code.

  • Below the create_loader() method, add the three register methods:
    @classmethod
    def register_source(cls, name: str, source_class: type[DataSource]) -> None:
        """Register a new source type."""
        cls._source_registry[name] = source_class

    @classmethod
    def register_transform(cls, name: str, transform_class: type[Transform]) -> None:
        """Register a new transform type."""
        cls._transform_registry[name] = transform_class

    @classmethod
    def register_loader(cls, name: str, loader_class: type[DataLoader]) -> None:
        """Register a new loader type."""
        cls._loader_registry[name] = loader_class

What does this code do?

These methods let anyone add a new source, transform, or loader type to the factory at runtime. You could write a PostgreSQLSource class and register it with StepFactory.register_source("postgres", PostgreSQLSource) without changing any existing factory code.

  • Save factory.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved factory.py before moving on.

ⓧ I'd like to double check the full code

Your factory.py should contain these sections in order:

  • Imports from typing, loaders, sources, and transforms.
  • The StepFactory class with three class-level registries (_source_registry, _transform_registry, _loader_registry).
  • Three create_* class methods: create_source(), create_transform(), create_loader().
  • Three register_* class methods: register_source(), register_transform(), register_loader().
Wire up main.py with both approaches

Now you can see both patterns in action. The first pipeline uses direct composition with method chaining. The second uses the Factory to build components from config dictionaries.

  • Replace the entire contents of main.py with the direct composition pipeline:
from factory import StepFactory
from loaders import JSONLoader
from pipeline import Pipeline
from sources import CSVSource
from transforms import FilterNulls, RenameColumns, TypeCast


def main() -> None:
    # Build pipeline using direct composition and method chaining
    pipeline = (
        Pipeline("sales_etl")
        .set_source(CSVSource(name="sales_csv", filepath="data/sales.csv"))
        .add_transform(FilterNulls(name="remove_empty", fields=["amount", "product"]))
        .add_transform(TypeCast(name="cast_types", schema={"amount": float, "quantity": int}))
        .add_transform(RenameColumns(name="standardize", mapping={"product": "product_name"}))
        .set_loader(JSONLoader(name="json_output", filepath="output/sales_clean.json"))
    )

    # Run the pipeline
    result = pipeline.run()

    # Print results
    print(pipeline.get_log().summary())
    print(f"\nRecords processed: {result.records_in}")
    print(f"Records loaded: {result.records_out}")
    print(f"Drop rate: {result.drop_rate:.1%}")
    print(f"Success: {result.success}")

What does this code do?

This is method chaining in action. Each call to .set_source(), .add_transform(), and .set_loader() returns the same Pipeline object, so you can chain them into one fluent expression.

Compare this to the old main.py where you manually called source.extract(), looped through transforms, and called loader.load(). The Pipeline handles all that orchestration internally.

  • Save main.py and run it:
python3 main.py

You should see the Pipeline Execution Summary showing Extract, Transform, and Load stages with record counts and status indicators.

Seeing a TypeError or ValueError?

If you see Pipeline has no data source configured, check that the method chaining is formatted correctly with each . at the start of each new line.

If you see a FileNotFoundError, make sure you are running the command from inside the data_pipeline folder and that data/sales.csv exists.

Help me debug my pipeline run error

Now add the Factory-built pipeline below the first one. This demonstrates building an entire pipeline from configuration dictionaries.

  • In main.py, add the following code below the print(f"Success: {result.success}") line, and replace the existing if __name__ block at the bottom:
    # Demonstrate Factory pattern
    print("\n--- Factory-built Pipeline ---\n")

    factory_pipeline = Pipeline("factory_demo")

    source = StepFactory.create_source({
        "type": "json",
        "name": "events_json",
        "filepath": "data/events.json"
    })

    transform = StepFactory.create_transform({
        "type": "filter_nulls",
        "name": "clean_events",
        "fields": ["event_type", "timestamp"]
    })

    loader = StepFactory.create_loader({
        "type": "csv",
        "name": "csv_output",
        "filepath": "output/events_clean.csv"
    })

    factory_pipeline.set_source(source)
    factory_pipeline.add_transform(transform)
    factory_pipeline.set_loader(loader)

    result = factory_pipeline.run()
    print(factory_pipeline.get_log().summary())


if __name__ == "__main__":
    main()

What does this code do?

Instead of importing and instantiating classes directly, you pass plain dictionaries to the factory. The "type" key tells the factory which class to use. The remaining keys become constructor arguments.

In production, these config dicts could come from a YAML file, a database, or an API. That means you can define pipelines without writing any Python. This is exactly how tools like Airflow let users define DAGs in config.

  • Save main.py.

Before you run this, what do you think the output will show? Two separate execution summaries? How many records will the second pipeline process from events.json?

  • Run the full script:
python3 main.py

You should see two complete execution summaries. The first pipeline processes 8 CSV records, drops some with missing amount or product values, and outputs clean JSON. The second pipeline reads 6 events from JSON, filters out records with null event_type or timestamp, and writes clean CSV.

Factory pipeline failing?

If you see Unknown source type, make sure the "type" value in your config dict matches a key in the registry. For sources, valid keys are "csv" and "json".

If you see a TypeError about unexpected keyword arguments, check that the dict keys (after removing "type") match the constructor parameters of the target class exactly.

Help me debug my factory pipeline error

✔️ Awesome, I've got everything!

Great. Confirm main.py is saved and both pipelines ran successfully.

ⓧ I'd like to double check the full code

Your main.py should contain these sections in order:

  • Imports from factory, loaders, pipeline, sources, and transforms.
  • A main() function that builds a Pipeline("sales_etl") with method chaining, runs it, and prints the log summary plus stats.
  • A second Pipeline("factory_demo") built using StepFactory.create_source(), StepFactory.create_transform(), and StepFactory.create_loader() with config dicts.
  • An if __name__ == "__main__": block calling main().

Your framework now has two ways to build pipelines. Direct composition gives you full control with type safety. The Factory gives you config-driven construction for automation. Next up, you will put this framework to the test in the Secret Mission by adding an Observer pattern for event monitoring.

Secret mission

Add the Observer Pattern

Your pipeline runs great, but external code has no way to react to what happens inside it. Want to add logging, metrics, or alerts? Right now you'd have to crack open the Pipeline class and hardcode them in. The Observer pattern decouples event producers from consumers, letting you bolt on monitoring without touching core framework code.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally with no cloud services or paid APIs. There are no ongoing costs to worry about.

Resources you used:

  • The data_pipeline project folder on your Desktop (including source files, sample data, and output files).

Keep everything running

No action needed. Choose this if you plan to keep extending your pipeline framework or want to reference the code later.

  • Your project files stay in the data_pipeline folder on your Desktop.
  • You can run python3 main.py again any time to re-execute the pipelines.

Pause - I'll come back to this later

Since this is a local project with no running processes, there is nothing to pause. Your files sit on disk until you return.

  • Close any open editor tabs for the project.
  • Your files remain in data_pipeline on your Desktop, ready whenever you want to pick up again.

Delete - I don't want to use this again

Remove the entire project folder to free up disk space and start fresh.

  • Close any open editor tabs for the project files.
  • On macOS or Linux, delete the project folder by running this command in your terminal:
rm -rf ~/Desktop/data_pipeline
  • On Windows, open PowerShell and run:
Remove-Item -Recurse -Force ~\Desktop\data_pipeline
  • Confirm the folder is gone by running ls ~/Desktop (macOS/Linux) or ls ~\Desktop (Windows). You should no longer see data_pipeline listed.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-style data pipeline framework from scratch using only Python's standard library.

You've learned how to:

  • Define contracts with Abstract Base Classes and build specialized implementations through inheritance (CSVSource, JSONSource, FilterNulls, TypeCast).
  • Orchestrate components using composition in the Pipeline class with method chaining, and swap algorithms at runtime with the Strategy pattern.
  • Build entire pipelines from config dictionaries using the Factory pattern with class-level registries and dataclasses for type-safe data models.
  • Added the Observer pattern (Secret Mission) to decouple event handling from the core framework, letting multiple observers react to pipeline events independently.

Ready to quiz yourself?