You've spent the last few modules learning about APIs, data cleaning, and scheduling. Now we're putting it all together. But before we build the "perfect" pipeline, I want to show you a piece of code that looks correct at a glance but will absolutely crash your production server at 3:00 AM.

import requests

def run_price_monitor():
    # Fetching product data from a mock API
    response = requests.get("https://api.store.com/products")
    products = response.json() 
    
    for product in products:
        # Logic: Alert me if the price is under 50
        if product['price'] < 50:
            print(f"Deal found: {product['name']}")
            # send_slack_notification(product['name'])

run_price_monitor()

The TypeError that kills the pipeline

If you ran this against a real API, there is a high probability it would throw a TypeError: '<' not supported between instances of 'str' and 'int'. Why? Because APIs are messy. Often, a price isn't returned as 49.99 (a float), but as "$49.99" (a string).

In a small script, this is a minor annoyance. In a full automation pipeline, this is a catastrophe. If this script is scheduled to run every hour, and the API suddenly changes its format or returns a null value for one single item in a list of a thousand, the entire pipeline crashes. You don't get your alerts, and you don't even know it's broken until you manually check the logs three days later.

Sanitizing inputs and graceful degradation

To build a professional pipeline, you have to assume that every external data source is lying to you or broken. We fix this by decoupling the extraction from the logic and adding a layer of sanitization.

import requests
import logging

def clean_price(price_str):
    """Convert '$49.99' or '49.99' to float 49.99"""
    try:
        if isinstance(price_str, str):
            price_str = price_str.replace('$', '').replace(',', '')
        return float(price_str)
    except (ValueError, TypeError):
        return None # Signal that this data point is junk

def run_price_monitor():
    try:
        response = requests.get("https://api.store.com/products", timeout=10)
        response.raise_for_status()
        products = response.json()
    except Exception as e:
        logging.error(f"Network failure: {e}")
        return # Stop here, but don't crash the whole system

    for product in products:
        price = clean_price(product.get('price'))
        
        if price is None:
            logging.warning(f"Skipping {product.get('name')} due to invalid price")
            continue
            
        if price < 50:
            print(f"Deal found: {product['name']}")

run_price_monitor()

Notice the change. I added a timeout to the request (never let a request hang forever), used .get() to avoid KeyError if the 'price' key is missing, and created a dedicated clean_price helper. Now, one bad piece of data doesn't kill the whole run; it just logs a warning and moves to the next item.

Connecting the dots with an Orchestrator function

When building an end-to-end pipeline, avoid writing one giant function. I prefer the "Orchestrator" pattern. You write small, testable functions for each stage—Extract, Transform, Load (ETL)—and then one main function that manages the flow of data between them.

• Extract: Fetching the raw data (API calls, database queries).
• Transform: Cleaning the data, filtering, and calculating.
• Load: Sending the final result to its destination (Email, Slack, CSV, Database).

By separating these, you can test your Transform logic with a fake list of data without actually having to hit the API every time you run your tests. It makes your code maintainable and, more importantly, it makes it easier to debug when something inevitably goes wrong in the middle of the night.

📋 Practical Task

Build a Daily Currency Arbitrage Monitor

Your task is to build a full automation pipeline that monitors the exchange rate between two currencies and logs a "Trade Alert" if the difference between two different sources exceeds a certain threshold.

Requirements:

• Extraction: Create two functions that simulate fetching the exchange rate for USD to EUR from two different providers. (Since you don't have API keys, you can use random.uniform(0.90, 0.95) to simulate the rate).
• Transformation: Create a function that takes these two rates, calculates the absolute difference, and determines if the "spread" is greater than 0.02.
• Loading: If the spread is high enough, append a timestamped alert to a file named arbitrage_log.txt. If not, log that "No opportunity found".
• Orchestration: Wrap the entire process in a main()` function. Ensure that if one of the "API calls" fails (simulate this by randomly raising an exception), the program catches the error and logs it without crashing.