Python Foundation for Data Engineers

Chapter 25 — Week 3 Checkpoint Project

Lesson 25.1 — Checkpoint: Build and Test a Small API Pipeline

Here's where all of Week 3 comes together — HTTP fundamentals, the requests library, secrets, pagination, resilience, DataFrame flattening, Parquet, and mocked testing — into one small, complete, trustworthy pipeline. This mirrors exactly what you built at the end of Week 1 and Week 2: a real, connected piece of work, built entirely by you, proven with tests rather than a hopeful glance at the output.

The scenario: pull every comment from a public API, safely and resiliently, clean and flatten the result, land it as Parquet, and prove the whole thing works — without a single test depending on the real network being available.

Step 1 — The resilient, paginated client.

Build this in api_pipeline.py, combining Chapter 21's pagination with Chapter 22's resilience:

python
import requests
import pandas as pd
import time


def resilient_get(url, params=None, max_attempts=3, timeout=5):
    """
    Make a resilient GET request: timeout-protected, retried with
    backoff on failure, and only treats a 200 with valid JSON as success.
    """
    for attempt in range(1, max_attempts + 1):
        try:
            response = requests.get(url, params=params, timeout=timeout)
            if response.status_code == 200:
                try:
                    return response.json()
                except requests.exceptions.JSONDecodeError:
                    print(f"Attempt {attempt}: got 200 but invalid JSON")
            else:
                print(f"Attempt {attempt}: got status {response.status_code}")
        except requests.exceptions.Timeout:
            print(f"Attempt {attempt}: request timed out")
        except requests.exceptions.RequestException as error:
            print(f"Attempt {attempt}: request failed — {error}")

        if attempt < max_attempts:
            time.sleep(2 ** attempt)

    print(f"All {max_attempts} attempts failed for {url}")
    return None


def fetch_all_comments(base_url, page_size=25, max_pages=50):
    """
    Fetch every page of comments using resilient_get for each page.
    Stops on an empty page, or at max_pages as a safety limit.
    """
    all_records = []
    page = 1

    while page <= max_pages:
        records = resilient_get(base_url, params={"_page": page, "_limit": page_size})

        if records is None:
            print(f"Stopping — page {page} failed after all retries")
            break
        if len(records) == 0:
            break

        all_records.extend(records)
        page += 1
        time.sleep(0.2)

    return all_records

Notice this is genuinely just Chapter 21 and Chapter 22's work, combined — fetch_all_comments now calls resilient_get for every single page, so a temporary failure on any one page gets retried properly, instead of only the very first request being protected.

Step 2 — Clean and flatten.

python
def clean_comments(raw_comments):
    """
    Flatten raw comment records into a clean DataFrame:
    - lowercases emails
    - removes exact duplicate records by id
    - drops rows with a missing body
    """
    df = pd.json_normalize(raw_comments)
    df = df.drop_duplicates(subset=["id"])
    df["email"] = df["email"].str.strip().str.lower()
    df = df.dropna(subset=["body"])
    return df

Step 3 — Land the result.

python
def save_comments(df, output_path):
    df.to_parquet(output_path, index=False)
    return len(df)

Step 4 — Run the full pipeline.

python
from api_pipeline import fetch_all_comments, clean_comments, save_comments

raw_comments = fetch_all_comments("https://jsonplaceholder.typicode.com/comments")
print(f"Fetched {len(raw_comments)} raw comments")

df = clean_comments(raw_comments)
print(f"Cleaned down to {len(df)} rows")

row_count = save_comments(df, "checkpoint_comments.parquet")
print(f"Saved {row_count} rows to checkpoint_comments.parquet")
Output / Note

Fetched 500 raw comments Cleaned down to 500 rows Saved 500 rows to checkpoint_comments.parquet

Step 5 — Prove it with tests, entirely without the real network.

This is the part that ties the whole week together. Every test below uses mocking from Chapter 24 — none of them make a real request, and all of them run in a fraction of a second.

python
from unittest.mock import patch, Mock
import pandas as pd
from api_pipeline import resilient_get, fetch_all_comments, clean_comments, save_comments


def test_resilient_get_succeeds_on_first_try():
    fake_response = Mock()
    fake_response.status_code = 200
    fake_response.json.return_value = {"id": 1, "body": "hello"}

    with patch("api_pipeline.requests.get", return_value=fake_response) as mock_get:
        result = resilient_get("https://example.com/data")

    assert result == {"id": 1, "body": "hello"}
    assert mock_get.call_count == 1


def test_resilient_get_retries_then_succeeds():
    failing_response = Mock()
    failing_response.status_code = 500

    success_response = Mock()
    success_response.status_code = 200
    success_response.json.return_value = {"id": 1, "body": "hello"}

    with patch("api_pipeline.requests.get") as mock_get:
        mock_get.side_effect = [failing_response, success_response]
        result = resilient_get("https://example.com/data", max_attempts=3)

    assert result == {"id": 1, "body": "hello"}
    assert mock_get.call_count == 2


def test_fetch_all_comments_stops_on_empty_page():
    page_one = [{"id": 1, "body": "first"}, {"id": 2, "body": "second"}]
    page_two = []

    with patch("api_pipeline.requests.get") as mock_get:
        response_one = Mock(status_code=200)
        response_one.json.return_value = page_one
        response_two = Mock(status_code=200)
        response_two.json.return_value = page_two

        mock_get.side_effect = [response_one, response_two]

        result = fetch_all_comments("https://example.com/comments", page_size=25)

    assert len(result) == 2


def test_clean_comments_removes_duplicates_and_lowercases_email():
    raw = [
        {"id": 1, "email": "Priya.Shah@Company.com", "body": "hi"},
        {"id": 1, "email": "Priya.Shah@Company.com", "body": "hi"},
        {"id": 2, "email": "Raj@Company.com", "body": "hello"},
    ]

    result = clean_comments(raw)

    assert len(result) == 2
    assert result["email"].iloc[0] == "priya.shah@company.com"


def test_clean_comments_drops_rows_with_missing_body():
    raw = [
        {"id": 1, "email": "priya@company.com", "body": "hi"},
        {"id": 2, "email": "raj@company.com", "body": None},
    ]

    result = clean_comments(raw)

    assert len(result) == 1


def test_save_comments_writes_expected_row_count(tmp_path):
    df = pd.DataFrame({"id": [1, 2, 3], "body": ["a", "b", "c"]})
    output_path = tmp_path / "test_output.parquet"

    row_count = save_comments(df, str(output_path))

    assert row_count == 3
    reloaded = pd.read_parquet(output_path)
    assert len(reloaded) == 3
Output / Note

test_api_pipeline.py ...... [100%] 6 passed in 0.04s

Take a moment on what you've actually proven here. test_resilient_get_retries_then_succeeds confirms your retry logic genuinely works, without needing a real server to cooperate by failing on command. test_fetch_all_comments_stops_on_empty_page confirms your pagination stops correctly, using two fake pages instead of waiting on 500 real API calls. test_clean_comments_removes_duplicates_and_lowercases_email and its neighbor confirm your cleaning logic handles exactly the messy situations real data tends to produce. And test_save_comments_writes_expected_row_count uses the tmp_path tool from Week 1, Chapter 9.4 to confirm the file actually lands correctly, without cluttering your real project folder.

Six tests, running in milliseconds, that will still pass next month regardless of whether jsonplaceholder.typicode.com happens to be reachable at that exact moment — because they're not testing the website. They're testing your code, which is genuinely the entire point.

Looking Back at the Whole Course

Think back to where this course started: print("Hello, Data Engineer"), back in Week 1, Chapter 1. From there:

  • Week 1 gave you the raw material — variables, strings, collections, control flow, functions, honest error handling, real files, and the core habit that's run through everything since: don't just look at your output, prove it's correct.
  • Week 2 gave you the tools built specifically for tables of real, messy data — NumPy's speed, pandas's cleaning and grouping power, real file formats, and testing extended to whole DataFrames.
  • Week 3 gave you the ability to reach beyond your own machine entirely — pulling live data safely, resiliently, and respectfully from the outside world, and proving that resilience works without depending on the outside world cooperating during a test run.

Every single exercise across these three weeks was built to connect to the ones before it — not because that made the course tidier, but because that's genuinely how real data engineering work actually happens: cleaning feeds into transformation, transformation feeds into loading, and every piece of it needs to be trustworthy on its own, and provably so.

You didn't just learn Python. You built the specific, connected set of skills a working data engineer reaches for daily — and you built the habit of proving your own work, rather than hoping it's right. That habit is worth more than any single tool covered in this course, because it's the one that keeps working no matter which new tool you meet next.

Try It Yourself: Go back through all three weeks' checkpoint projects — Week 1's customer cleaner, Week 2's revenue-by-region pipeline, and this week's comment pipeline — and count the total number of tests you've written across the whole course. That number is your own, concrete proof of exactly how far your confidence in your own code has come since Chapter 1.