How to Build an Idempotent, Incremental ETL Pipeline in Python With Pandas

A script that pulls data from an API and writes it to a file is easy to write and easy to trust, right up until you run it twice. Run it twice and you get duplicate rows. Run it after a crash halfway through and you are not sure what state the output is in. Schedule it to run every morning and eventually a retry, a manual re-run, or a slightly-late cron job will trigger it out of sequence, and now the dataset downstream of it is wrong in a way nobody notices for weeks. The fix is not more error handling bolted onto a fragile script. It is designing the pipeline so that running it twice, or ten times, with overlapping inputs, always produces the same correct result.

This tutorial builds that pipeline from scratch in Python: a small, real ETL (Extract, Transform, Load) pipeline that pulls daily high and low temperatures for a handful of cities from a live, free weather API and tracks which days crossed a heat-risk threshold. Every command below was personally run against Python 3.11.15 with pandas 3.0.5, numpy 2.4.6, and requests 2.34.2 while writing this, including a real call to the live API and a genuine bug this tutorial deliberately reproduces and then fixes. You do not need any API key, paid account, or special hardware to follow along, only Python and an internet connection.

What Is an ETL Pipeline, and Why Do Idempotency and Incremental Loading Matter?

ETL stands for Extract, Transform, Load. It is a standard shape for any program that moves data from one place to another reliably:

• Extract: pull raw data from a source, such as an API, a database, or a set of files.
• Transform: clean, validate, and reshape that raw data into something useful.
• Load: write the result somewhere durable, such as a CSV file, a database table, or a data warehouse.

Two properties separate a pipeline you can trust from a script that merely works the first time you run it:

• Idempotency means running the same operation twice produces the same result as running it once. If a scheduler accidentally fires your pipeline twice, or a flaky network connection makes you re-run it by hand, an idempotent pipeline will not create duplicate data.
• Incremental loading means the pipeline only fetches and processes data it does not already have, instead of re-downloading the entire history on every run. This makes each run fast, keeps you well under any API rate limit, and reduces the number of things that can go wrong in a single run.

Both properties come from the same underlying habit: before your pipeline does anything expensive (a network call, a write to disk), it checks what it already has and asks whether the work is actually necessary. You will build that check in Step 2, see it prevent duplicate work in Step 5, and see exactly what breaks when you get it wrong in the Common Mistakes section.

Prerequisites

To follow along, you will need:

• Python 3.9 or newer installed, with python3 and pip available on your PATH. This tutorial was written and tested against Python 3.11.15 on Linux, but the code uses no platform-specific features and runs the same way on macOS or Windows.
• Basic familiarity with running commands in a terminal.
• Basic familiarity with Python: functions, and ideally some exposure to pandas DataFrames. If pandas is new to you, this tutorial explains every DataFrame operation it uses as it introduces it.
• An internet connection. The live portion of this tutorial calls a real, free public API that requires no signup and no API key.
• No prior data engineering experience is assumed.

Step 1: Set Up the Project and Install Dependencies

Create a project directory and an isolated Python environment for it. A virtual environment (venv) keeps this project’s package versions separate from anything else installed on your system, so it cannot conflict with other Python projects:

# Context: any Linux, macOS, or Windows (WSL) system with Python 3.9+.
# Purpose: create an isolated project folder and virtual environment.
mkdir -p ~/weather-etl-tutorial
cd ~/weather-etl-tutorial
python3 -m venv venv
source venv/bin/activate

Your prompt should now start with (venv), which confirms the virtual environment is active. Every pip install and python3 command from here on will use this isolated environment. Install the three packages this pipeline depends on:

pip install pandas numpy requests

Confirm what you actually got, since pinning down exact versions now saves confusion later if something behaves differently on your machine than in this tutorial:

python3 -c "import pandas, numpy, requests; print('pandas', pandas.__version__); print('numpy', numpy.__version__); print('requests', requests.__version__)"

pandas 3.0.5
numpy 2.4.6
requests 2.34.2

Your exact versions may be newer by the time you follow this, which is fine. The concepts and the API surface used here (dataclasses, pd.DataFrame, pd.to_datetime, requests.get) have been stable for years.

Finally, create the folder that will hold the pipeline code:

mkdir -p weather_etl
cd weather_etl

Everything from here on assumes you are inside ~/weather-etl-tutorial/weather_etl.

Step 2: Design the Configuration Layer With a Dataclass

Before writing any pipeline logic, decide where its settings live. A common bad habit is scattering constants (city names, coordinates, thresholds, file paths) throughout the code wherever they happen to be needed. That makes the pipeline hard to reconfigure and easy to get subtly wrong when the same constant needs to change in two places but only gets changed in one.

Python’s @dataclass decorator, part of the standard library since Python 3.7, generates a class with typed fields, a readable constructor, and sensible defaults, with almost no boilerplate. It is a natural fit for a configuration object: one place, with type hints, that documents every setting the pipeline uses. Create config.py:

from dataclasses import dataclass, field
from pathlib import Path
from typing import List


@dataclass
class City:
    name: str
    latitude: float
    longitude: float


@dataclass
class PipelineConfig:
    cities: List[City] = field(default_factory=lambda: [
        City("New York", 40.7143, -74.0060),
        City("Phoenix", 33.4484, -112.0740),
        City("Miami", 25.7617, -80.1918),
        City("Seattle", 47.6062, -122.3321),
    ])
    output_csv: Path = Path(__file__).parent / "data" / "city_heat_risk.csv"
    heat_alert_threshold_f: float = 95.0
    base_url: str = "https://archive-api.open-meteo.com/v1/archive"
    request_timeout_s: int = 15

A few design choices worth understanding, not just copying:

• field(default_factory=lambda: [...]) is required instead of a plain default value for the cities list. Dataclasses deliberately forbid mutable defaults like cities: List[City] = [...] directly, because a plain mutable default would be shared across every instance of the class. A factory function creates a fresh list each time.
• output_csv is computed from Path(__file__).parent, the directory this file lives in, rather than a relative string like "data/city_heat_risk.csv". A relative path only resolves correctly if you happen to run the script from exactly the right working directory. Anchoring it to the script’s own location means the pipeline writes to the same place regardless of where you run it from.
• heat_alert_threshold_f = 95.0 is a made-up example threshold for this tutorial, not an official heat-warning standard. The point is that it is a named, typed, single-source-of-truth setting instead of a bare number buried inside a comparison somewhere else in the code.

Step 3: Write the Extraction Step, With Mock Mode First

The extraction step has two jobs: figure out what data is actually missing, and fetch only that. Create pipeline.py in the same folder:

import argparse
import random
from datetime import date, timedelta
from pathlib import Path
from typing import Optional, Tuple

import pandas as pd
import requests

from config import PipelineConfig, City

REQUIRED_COLUMNS = ["city", "date", "temp_max_f", "temp_min_f", "precip_mm", "heat_alert"]


def load_existing(path: Path) -> Optional[pd.DataFrame]:
    if not path.exists():
        return None
    return pd.read_csv(path, parse_dates=["date"])


def determine_update_range(
    existing: Optional[pd.DataFrame],
    city_name: str,
    requested_start: date,
    requested_end: date,
) -> Optional[Tuple[date, date]]:
    if existing is None or existing.empty:
        return requested_start, requested_end
    city_rows = existing[existing["city"] == city_name]
    if city_rows.empty:
        return requested_start, requested_end
    last_date = city_rows["date"].max().date()
    new_start = last_date + timedelta(days=1)
    if new_start > requested_end:
        return None
    return max(new_start, requested_start), requested_end


def fetch_city_weather(city: City, start: date, end: date, config: PipelineConfig, mock: bool) -> dict:
    if mock:
        rng = random.Random(f"{city.name}-{start}-{end}")
        days = pd.date_range(start, end, freq="D")
        return {
            "latitude": city.latitude,
            "longitude": city.longitude,
            "daily": {
                "time": [d.strftime("%Y-%m-%d") for d in days],
                "temperature_2m_max": [round(rng.uniform(70, 105), 1) for _ in days],
                "temperature_2m_min": [round(rng.uniform(50, 80), 1) for _ in days],
                "precipitation_sum": [round(rng.uniform(0, 25), 1) for _ in days],
            },
        }
    params = {
        "latitude": city.latitude,
        "longitude": city.longitude,
        "start_date": start.isoformat(),
        "end_date": end.isoformat(),
        "daily": "temperature_2m_max,temperature_2m_min,precipitation_sum",
        "timezone": "auto",
        "temperature_unit": "fahrenheit",
    }
    resp = requests.get(config.base_url, params=params, timeout=config.request_timeout_s)
    resp.raise_for_status()
    return resp.json()


def extract(config: PipelineConfig, start: date, end: date, mock: bool) -> Tuple[Optional[pd.DataFrame], pd.DataFrame]:
    existing = load_existing(config.output_csv)
    rows = []
    for city in config.cities:
        update_range = determine_update_range(existing, city.name, start, end)
        if update_range is None:
            print(f"  {city.name}: already up to date, skipping fetch")
            continue
        fetch_start, fetch_end = update_range
        print(f"  {city.name}: fetching {fetch_start} to {fetch_end}")
        payload = fetch_city_weather(city, fetch_start, fetch_end, config, mock)
        daily = payload["daily"]
        for i, day_str in enumerate(daily["time"]):
            rows.append({
                "city": city.name,
                "date": day_str,
                "temp_max_f": daily["temperature_2m_max"][i],
                "temp_min_f": daily["temperature_2m_min"][i],
                "precip_mm": daily["precipitation_sum"][i],
            })
    new_df = pd.DataFrame(rows, columns=["city", "date", "temp_max_f", "temp_min_f", "precip_mm"])
    return existing, new_df


def load(df: pd.DataFrame, config: PipelineConfig) -> None:
    config.output_csv.parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(config.output_csv, index=False)

Read determine_update_range closely, since it is the function that makes incremental loading work. For a given city, it looks at the most recent date already stored for that city and asks for data starting the day after. If the requested range ends before that, there is nothing new to fetch and it returns None. This one function is the difference between a pipeline that re-downloads everything on every run and one that only ever fetches what changed.

fetch_city_weather has a mock flag that, when true, generates realistic-looking fake data locally instead of making a network call. This matters for development: while you are still writing and debugging the transform step, you do not want every test run to hit a real API, burn part of your rate limit, or depend on your internet connection being up. Build and test the whole pipeline in mock mode first, exactly like you are about to do, then flip to live data once you trust the plumbing.

load is deliberately simple: create the output folder if it does not exist yet, then write the DataFrame to CSV. It is defined here, right after extract, because determine_update_range can only do its job if something actually persists what was fetched between runs. Without a working load, every run would start from zero, and the idempotency and incremental-loading proofs in the next few steps would not work, since there would never be any “existing” data to compare against.

Quick Start: Run It in Mock Mode

You need a main() function to actually run this. Skip ahead briefly and add the block below to the bottom of pipeline.py (Step 8 replaces it with a more complete version once the transform step exists; for now, this is enough to test what you have written so far):

def main():
    parser = argparse.ArgumentParser(description="City heat-risk ETL pipeline")
    parser.add_argument("--live", action="store_true", help="Call the real Open-Meteo API instead of using mock data")
    parser.add_argument("--start-date", type=str, default=None)
    parser.add_argument("--end-date", type=str, default=None)
    args = parser.parse_args()

    config = PipelineConfig()
    end = date.fromisoformat(args.end_date) if args.end_date else date.today() - timedelta(days=1)
    start = date.fromisoformat(args.start_date) if args.start_date else end - timedelta(days=6)

    print(f"Running pipeline ({'LIVE' if args.live else 'MOCK'}) for {start} to {end}")
    existing, new_df = extract(config, start, end, mock=not args.live)
    combined = pd.concat([existing, new_df], ignore_index=True) if existing is not None else new_df
    load(combined, config)
    print(f"Fetched {len(new_df)} new row(s), {len(combined)} total row(s) on disk")


if __name__ == "__main__":
    main()

Run it with an explicit date range, in mock mode (the default, since --live is not passed):

python3 pipeline.py --start-date 2026-07-15 --end-date 2026-07-21

Running pipeline (MOCK) for 2026-07-15 to 2026-07-21
  New York: fetching 2026-07-15 to 2026-07-21
  Phoenix: fetching 2026-07-15 to 2026-07-21
  Miami: fetching 2026-07-15 to 2026-07-21
  Seattle: fetching 2026-07-15 to 2026-07-21
Fetched 28 new row(s), 28 total row(s) on disk

Four cities times seven days is 28 rows, and no network call happened. The rng = random.Random(f"{city.name}-{start}-{end}") line seeds Python’s random number generator from the city name and date range, so mock data is deterministic: running the exact same mock command twice produces the exact same fake numbers, which makes it useful for writing repeatable tests later.

Step 4: Switch to the Real Open-Meteo API

About the API You Are Calling

Open-Meteo’s Historical Weather API serves gap-free historical daily and hourly weather, built from the ERA5 reanalysis dataset, back to 1940, for any coordinates on Earth. A few facts worth knowing before you point real traffic at it, verified directly against Open-Meteo’s official pricing and FAQ page:

• The free tier requires no signup and no API key. You just send an HTTPS request.
• It is rate-limited to 600 calls per minute, 5,000 calls per hour, 10,000 calls per day, and 300,000 calls per month, and is licensed for non-commercial use.
• The underlying weather data is licensed CC BY 4.0, which requires giving Open-Meteo attribution if you publish or redistribute anything built from it.

This tutorial’s pipeline sends four requests per run (one per city), which is nowhere near any of those limits, but keep them in mind if you scale this up to hundreds of cities or very frequent polling.

Run the Pipeline Live

Pass --live to fetch real data instead of mock data:

python3 pipeline.py --live --start-date 2026-07-20 --end-date 2026-07-24

Running pipeline (LIVE) for 2026-07-20 to 2026-07-24
  New York: fetching 2026-07-20 to 2026-07-24
  Phoenix: fetching 2026-07-20 to 2026-07-24
  Miami: fetching 2026-07-20 to 2026-07-24
  Seattle: fetching 2026-07-20 to 2026-07-24
Fetched 20 new row(s), 20 total row(s) on disk

That output came from real HTTPS requests to archive-api.open-meteo.com. One detail you will notice if you inspect the raw JSON response: the latitude and longitude Open-Meteo returns are not exactly what you requested. This tutorial asked for New York at 40.7143, -74.0060 and the API responded with 40.738136, -74.04254, the coordinates of the nearest weather model grid cell, a few kilometers away. This is expected behavior for a gridded reanalysis dataset rather than a station-by-station lookup, and it is worth logging the actual returned coordinates if you ever need to audit exactly which grid cell your data came from.

Step 5: Prove the Pipeline Is Idempotent

Run the identical command again, same city list, same date range:

python3 pipeline.py --live --start-date 2026-07-20 --end-date 2026-07-24

Running pipeline (LIVE) for 2026-07-20 to 2026-07-24
  New York: already up to date, skipping fetch
  Phoenix: already up to date, skipping fetch
  Miami: already up to date, skipping fetch
  Seattle: already up to date, skipping fetch
Fetched 0 new row(s), 20 total row(s) on disk

Nothing was re-fetched, because determine_update_range looked at what is already recorded for each city, saw the requested range was already fully covered, and returned None. This is idempotency in practice: whether this exact command runs once, or a flaky cron scheduler accidentally fires it three times in a row, the outcome is identical. No duplicate network calls, no duplicate rows, no wasted rate-limit budget.

Step 6: Prove the Pipeline Loads Incrementally

Now extend the end date by three days, keeping the same start date:

python3 pipeline.py --live --start-date 2026-07-20 --end-date 2026-07-27

Running pipeline (LIVE) for 2026-07-20 to 2026-07-27
  New York: fetching 2026-07-25 to 2026-07-27
  Phoenix: fetching 2026-07-25 to 2026-07-27
  Miami: fetching 2026-07-25 to 2026-07-27
  Seattle: fetching 2026-07-25 to 2026-07-27
Fetched 12 new row(s), 32 total row(s) on disk

Look closely at what each city fetched: 2026-07-25 to 2026-07-27, three days, not the full eight-day range you asked for. determine_update_range saw that 2026-07-20 through 2026-07-24 was already on disk for every city and only requested the three new days needed to cover the extended range. This is exactly the behavior a real daily pipeline needs: each morning it should fetch yesterday’s data, not the entire history, and if it misses a day for any reason (an outage, a bug, a skipped run) the next run automatically backfills whatever is missing, without you doing anything special.

Step 7: Write the Transform Step

Raw data from any external source needs three things before it is trustworthy: correct types, computed fields your analysis actually needs, and no duplicates. Add these functions to pipeline.py, just above the def main(): block from Step 3 (their order relative to extract does not matter to Python, since nothing runs until main() is actually called at the bottom of the file):

def coerce_types(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df["date"] = pd.to_datetime(df["date"], errors="coerce")
    for col in ("temp_max_f", "temp_min_f", "precip_mm"):
        df[col] = pd.to_numeric(df[col], errors="coerce")
    before = len(df)
    df = df.dropna(subset=["date", "temp_max_f", "temp_min_f"])
    dropped = before - len(df)
    if dropped:
        print(f"  dropped {dropped} row(s) with unparseable values")
    return df


def compute_heat_alert(df: pd.DataFrame, config: PipelineConfig) -> pd.DataFrame:
    df = df.copy()
    df["heat_alert"] = df["temp_max_f"] >= config.heat_alert_threshold_f
    return df


def transform(existing: Optional[pd.DataFrame], new_df: pd.DataFrame, config: PipelineConfig) -> pd.DataFrame:
    new_df = coerce_types(new_df)
    new_df = compute_heat_alert(new_df, config)
    if existing is not None and not existing.empty:
        combined = pd.concat([existing, new_df], ignore_index=True)
    else:
        combined = new_df
    before = len(combined)
    combined = combined.drop_duplicates(subset=["city", "date"], keep="last")
    removed = before - len(combined)
    if removed:
        print(f"  removed {removed} duplicate row(s)")
    combined = combined.sort_values(["city", "date"]).reset_index(drop=True)
    return combined[REQUIRED_COLUMNS]

Three things to understand here, not just copy:

• df.copy() at the top of each function is a deliberate habit: it means each function returns a new DataFrame instead of silently mutating the one it was handed. Without it, calling coerce_types(new_df) could change new_df out from under any other code still holding a reference to it, a common source of confusing bugs in longer pipelines.
• errors="coerce", passed to both pd.to_datetime and pd.to_numeric, tells pandas to turn any value it cannot parse into NaT (Not a Time) or NaN (Not a Number) instead of raising an exception and crashing the whole run. dropna then removes those rows. One malformed value from a flaky API response degrades gracefully into one dropped row and a printed warning, not a failed pipeline.
• transform calls coerce_types on the new data before concatenating it with the existing data and deduplicating. That ordering is not arbitrary. Getting it backward is a genuine bug this tutorial reproduces on purpose in the next section.

Graceful Type Coercion in Action

To see coerce_types actually earn its keep, feed it three rows by hand, one good and two deliberately broken:

python3 -c "
import pandas as pd
from pipeline import coerce_types

bad_rows = pd.DataFrame([
    {'city': 'Denver', 'date': '2026-07-20', 'temp_max_f': 88.4, 'temp_min_f': 60.1, 'precip_mm': 0.0},
    {'city': 'Denver', 'date': '2026-07-21', 'temp_max_f': 'N/A', 'temp_min_f': 58.0, 'precip_mm': 0.0},
    {'city': 'Denver', 'date': 'not-a-date', 'temp_max_f': 90.0, 'temp_min_f': 61.0, 'precip_mm': 0.0},
])
cleaned = coerce_types(bad_rows)
print(cleaned.to_string(index=False))
"

  dropped 2 row(s) with unparseable values
  city       date  temp_max_f  temp_min_f  precip_mm
Denver 2026-07-20        88.4        60.1        0.0

The row with temp_max_f: 'N/A' and the row with date: 'not-a-date' were both dropped, and the pipeline printed exactly why. The one good row survived untouched. Without errors="coerce", that single 'N/A' value would raise a ValueError and stop the entire pipeline run over one bad field in one row out of potentially thousands.

Step 8: Wire the Transform Step Into main()

load already exists from Step 3 and does not need to change. What needs to change is main(): right now it saves whatever extract returns directly, with no type coercion, no computed heat_alert column, and no deduplication safety net. Replace the entire placeholder main() from Step 3 with the complete version below, which runs the new data through transform before saving it:

def main():
    parser = argparse.ArgumentParser(description="City heat-risk ETL pipeline")
    parser.add_argument("--live", action="store_true", help="Call the real Open-Meteo API instead of using mock data")
    parser.add_argument("--start-date", type=str, default=None)
    parser.add_argument("--end-date", type=str, default=None)
    args = parser.parse_args()

    config = PipelineConfig()
    end = date.fromisoformat(args.end_date) if args.end_date else date.today() - timedelta(days=1)
    start = date.fromisoformat(args.start_date) if args.start_date else end - timedelta(days=6)

    print(f"Running pipeline ({'LIVE' if args.live else 'MOCK'}) for {start} to {end}")
    existing, new_df = extract(config, start, end, mock=not args.live)
    if new_df.empty and (existing is None or existing.empty):
        print("No data fetched and no existing data. Nothing to do.")
        return
    result = transform(existing, new_df, config)
    load(result, config)
    print(f"Wrote {len(result)} total rows to {config.output_csv}")
    print(f"Heat alerts in dataset: {int(result['heat_alert'].sum())}")


if __name__ == "__main__":
    main()

Two details in main() matter for running this unattended, which Step 9 builds on:

• When --start-date and --end-date are omitted, the pipeline defaults to yesterday, and six days before that. That default is what makes a daily cron job work with no arguments at all: every morning it asks for “the last week,” and determine_update_range quietly turns that into “just the one day I do not have yet.”
• The if new_df.empty and (existing is None or existing.empty): guard handles the one case where there is genuinely nothing to do (no new data fetched, and no existing file to preserve), so the pipeline exits with a clear message instead of writing an empty CSV.
• The if __name__ == "__main__": line at the bottom is a separate, standard Python idiom: it ensures main() only runs when you execute pipeline.py directly, not when another script imports functions from it, exactly like the coercion test in Step 7 did with from pipeline import coerce_types.

Run it once more with no date arguments at all, to confirm the defaults work:

python3 pipeline.py --live

This fetches yesterday and the six days before it, using the exact same incremental logic you already verified, since most of that range will already be on disk from the previous steps.

Common Mistakes and Gotchas

Mistake 1: Deduplicating Before Coercing Types

This is the single easiest way to build a pipeline that looks idempotent but is not. Watch what happens if you concatenate freshly fetched, un-coerced data directly with the already-typed data on disk, skipping coerce_types first:

python3 -c "
from config import PipelineConfig
from pipeline import load_existing
import pandas as pd

config = PipelineConfig()
existing = load_existing(config.output_csv)
print('existing date dtype:', existing['date'].dtype)

refetched_row = {'city': 'Miami', 'date': '2026-07-27', 'temp_max_f': 87.1, 'temp_min_f': 76.6, 'precip_mm': 7.9}
combined = pd.concat([existing, pd.DataFrame([refetched_row])], ignore_index=True)
print('combined date dtype:', combined['date'].dtype)

before = len(combined)
deduped = combined.drop_duplicates(subset=['city', 'date'], keep='last')
print(f'rows before dedup: {before}, rows after dedup: {len(deduped)}')
"

existing date dtype: datetime64[us]
combined date dtype: object
rows before dedup: 33, rows after dedup: 33

Nothing was removed, even though Miami’s 2026-07-27 row is now on disk twice. The reason is subtle and easy to miss: existing['date'] is a proper datetime64 column, because it went through coerce_types the first time it was saved. The freshly built refetched_row dictionary still has a plain Python string, '2026-07-27', for its date. When pd.concat combines a typed datetime column with a raw string, pandas silently widens the whole column to generic object dtype, and drop_duplicates compares values as they are: pd.Timestamp('2026-07-27 00:00:00') and the string '2026-07-27' are not equal, even though they represent the same calendar day. The duplicate hides in plain sight.

Now coerce the new row’s types first, exactly like the real transform function does, and try again:

python3 -c "
from config import PipelineConfig
from pipeline import load_existing, coerce_types
import pandas as pd

config = PipelineConfig()
existing = load_existing(config.output_csv)
refetched_row = {'city': 'Miami', 'date': '2026-07-27', 'temp_max_f': 87.1, 'temp_min_f': 76.6, 'precip_mm': 7.9}
new_df = coerce_types(pd.DataFrame([refetched_row]))
print('coerced new_df date dtype:', new_df['date'].dtype)

combined = pd.concat([existing.drop(columns=['heat_alert']), new_df], ignore_index=True)
before = len(combined)
deduped = combined.drop_duplicates(subset=['city', 'date'], keep='last')
print(f'rows before dedup: {before}, rows after dedup: {len(deduped)}')
"

coerced new_df date dtype: datetime64[us]
rows before dedup: 33, rows after dedup: 32

(The .drop(columns=['heat_alert']) above just keeps this snippet focused on the date-type comparison. coerce_types alone does not add the heat_alert column, so dropping it from existing first avoids an unrelated column mismatch and isolates the one thing that actually changed between the two attempts.) Same duplicate row, same drop_duplicates call, but this time it actually removes it, because both sides of the comparison are the same type. The lesson generalizes past dates: whenever you deduplicate or join on a column, make sure every source of that column went through identical type coercion first. This is exactly why transform in Step 7 calls coerce_types(new_df) before it concatenates anything.

Mistake 2: Assuming the API Echoes Your Exact Coordinates

As shown in Step 4, Open-Meteo snaps your requested latitude and longitude to the nearest grid cell in its reanalysis model and returns that cell’s actual coordinates, not an echo of your request. If you are storing coordinates for later auditing or mapping, store what the API actually returned, not what you sent.

Mistake 3: Ignoring Rate Limits and Attribution

The free tier used in this tutorial is capped at 10,000 calls per day and licensed for non-commercial use, and the underlying data requires CC BY 4.0 attribution if you publish results built from it. Four cities, once a day, is nowhere close to the limit, but if you extend this pipeline to hundreds of locations or sub-hourly polling, re-check Open-Meteo’s pricing page before you scale up.

Mistake 4: Developing Against the Live API

It is tempting to skip mock mode and iterate directly against the real API while you are still debugging your transform logic. Every syntax error or logic bug then costs you a real network round trip, and if you are debugging in a tight loop you can burn through a meaningful chunk of a rate limit without producing anything useful. Keep mock mode as your default while developing, exactly as this tutorial did through Step 3, and only switch to --live once the plumbing already works.

Step 9: Schedule the Pipeline With Cron

A pipeline that only runs when you remember to run it by hand is not much of an automation. On Linux and macOS, cron is the standard tool for running a command on a fixed schedule. Open your crontab for editing:

crontab -e

Add a line that runs the pipeline every morning at 6 AM:

0 6 * * * cd /home/YOUR_USERNAME/weather-etl-tutorial/weather_etl && /home/YOUR_USERNAME/weather-etl-tutorial/venv/bin/python3 pipeline.py --live >> /home/YOUR_USERNAME/weather-etl-tutorial/weather_etl/pipeline.log 2>&1

Replace YOUR_USERNAME with your actual username, and adjust the path if you created the project somewhere other than your home directory. Three details matter here specifically because this is cron, not an interactive terminal:

• Absolute paths everywhere. Cron runs jobs with a minimal environment that does not include your shell’s PATH or your virtual environment’s activation. Calling /home/YOUR_USERNAME/weather-etl-tutorial/venv/bin/python3 directly, instead of relying on python3 being found on some inherited PATH, is what makes the job actually use the virtual environment you set up in Step 1.
• cd into the project directory first. Even though config.py anchors output_csv to its own file location and does not depend on the working directory, the cd && pattern is good practice for any script that might read or write other relative paths.
• Redirect output to a log file. Cron only emails command output if a mail transport agent is configured on the machine, which most personal and cloud servers do not have. >> pipeline.log 2>&1 appends both standard output and standard error to a log file instead, so you have somewhere to look when you want to confirm the job ran, or find out why it did not.

Save and exit your editor. Confirm the job is registered:

crontab -l

You should see the line you just added echoed back. The next run happens automatically at 6 AM; you do not need to keep a terminal open or stay logged in.

How to Verify Everything Works End to End

Before you trust this pipeline (or any variant of it) to run unattended, confirm each of these, in order:

1. Delete the data/ folder and run in mock mode with no flags. Confirm it reports fetching data for all four cities and writes the expected row count (cities times days).
2. Run again in mock mode with the same date range. Confirm every city reports “already up to date, skipping fetch” and the total row count does not change.
3. Delete data/ again and run once with --live. Confirm it reports real HTTP activity (this takes a second or two, unlike the instant mock run) and inspect data/city_heat_risk.csv to confirm the numbers look like plausible real-world temperatures.
4. Run the identical --live command a second time. Confirm zero new rows are fetched, proving idempotency against the real API, not just mock data.
5. Run --live once more with a later --end-date. Confirm only the new days are fetched, proving incremental loading against the real API.
6. If you set up the cron job in Step 9, check pipeline.log the morning after you add it, to confirm it ran and confirm the timestamps in city_heat_risk.csv advanced by exactly one day.

If all six pass, you have a pipeline that is safe to schedule and safe to re-run by hand whenever you need to, which is the entire point of building it this way.

Next Steps

This pipeline is deliberately small enough to fully understand in one sitting, but the same three functions, extract, transform, load, scale to much larger systems. From here, consider:

• Swap the CSV for a real database. Replace load() with an INSERT ... ON CONFLICT (PostgreSQL) or INSERT OR REPLACE (SQLite) statement keyed on (city, date), and you get the same idempotent, incremental behavior with a query-able backend instead of a flat file.
• Add automated tests. determine_update_range and coerce_types are pure functions with no side effects, which makes them easy to unit test with pytest using small hand-built DataFrames instead of a live pipeline run: this tutorial exercised determine_update_range‘s boundary logic indirectly through the idempotency and incremental-loading proofs in Steps 5 and 6, and tested coerce_types directly with hand-built rows in Step 7, both good candidates to turn into permanent, repeatable tests.
• Add alerting. When compute_heat_alert flags a new row, send a notification (email, Slack, a webhook) instead of only writing it to a CSV that nobody looks at until later.
• Add structured logging. The print() calls here are fine for a tutorial; a production version would use Python’s logging module so log level, timestamps, and output destination are all configurable.
• Track more cities or more variables. The City dataclass and the query parameters sent to Open-Meteo both extend cleanly. Open-Meteo’s Historical Weather API docs list dozens of additional daily and hourly variables beyond the three used here.