Production ETL pipelines need to keep running reliably, even when data is messy or APIs fail.

In this handbook, Brooklyn shows you how to build a production-grade ETL pipeline in Python using real flood data.

You’ll learn about incremental loading, type coercion, deduplication, idempotency, and other patterns for building reliable data pipelines.

https://freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/…

The ETL Pipeline Handbook: How to Build a Production-Grade Pipeline in Python

Source: https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/ Tracking flood risk takes one unglamorous but essential thing: clean and structured data.

In this tutorial, you’ll build a data pipeline yourself. You’ll create a Python ETL (Extract, Transform, Load) pipeline that pulls daily water-level readings fromHub’Eau, France’s official open water-data API. Then, you’ll clean that data and publish it as a public dataset, just like thelive versiondoes.

This tutorial is based on a real pipeline that runs once a week, on a schedule, and it keeps theParis Flood Datasetupdated automatically.

You won’t just copy and paste code, though. The real goal is to understandwhythe pipeline works the way it does. You’ll walk through the design decisions that separate a script meant to run once from a script that keeps working, unattended, for years.

You can code along with thisnotebook. For most of the tutorial, the pipeline runs onsimulated (mock) API data. This lets you run every cell safely, without hammering a real server. A later section shows how to switch to the live API.

By the end, you’ll be able to:

• Explain and implement the Extract, Transform, Load pattern
• Manage configuration with Python@dataclassinstead of scattering constants everywhere
• Write API-fetching code that survives network failures and paginated responses
• Apply robust type-coercion so one bad row can’t crash a whole pipeline run
• Deduplicate and merge incremental data safely
• Wire everything into a single, idempotent, schedulablemain\(\)entry point

Table of Contents:

• Prerequisites
• Part 1: The Design Logic - What is an ETL pipeline? - The architecture, at a glance - Two patterns that make it production-grade
• Part 2: Setup and Dependencies
• Part 3: Manage Configuration with Dataclasses - Why bother with a config layer at all? - Code-level walkthrough
• Part 4: The Extraction Step - Graceful file loading - Incremental update logic - Simulate the API - Fetch data for real
• Part 5: The Transform Step - Type parsing and graceful coercion - Schema translation with bidirectional mappings - Compute flood alerts - Column ordering - Deduplication - Put it all together inpostprocess()
• Part 6: The Load Step - Design logic - Code-level walkthrough
• Part 7: Assemble the Full Pipeline - The global rehearsal (mock mode) - main(): pipeline orchestration - Theif name == “main”:guard
• Part 8: Go Live and Switch to the Real API - Post-run validation
• Summary
• References

Prerequisites

• Python 3.10+. The code uses type hints and dataclasses. These work from Python 3.7 onward, but 3.10+ is best.
• Knowledge of pandas DataFrames: read data, filter rows, and basic column operations.
• Comfort withfunctions and basic OOP(Object-Oriented programming) in Python. Don’t worry, this guide explains every non-obvious piece as you go.
• Optional: a freeKaggleaccount and theKaggle CLI, only if you want to run the final publishing step for real.

Install the dependencies:

pip install requests pandas numpy ipykernel

You’ll do the work inside aJupyter notebook. Download the\.ipynbfile from Kaggle, or click “Copy and Edit” to work directly on Kaggle. You’ll need a Kaggle account for that.

**Optional:**If you plan to run the notebook locally, install thenotebookpackage too:

pip install notebook

Part 1: The Design Logic

Before you touch a single line of code, we’ll spend three minutes onwhythe pipeline is shaped this way.

This is thebig-picture view. Every code-level decision later traces back to one of these ideas. Read this section even if you skim everything else.

What is an ETL Pipeline?

ETL stands forExtract, Transform, Load. It’s the standard pattern for moving data from a source to a destination in a reliable, repeatable way.

• Extract: pull data from a source, like an API, a database, or files.
• Transform: clean, standardize, enrich, and validate the data.
• Load: write the result to a destination, like a warehouse, a CSV, or a public platform.

Here’s how those three stages map onto this project:

StageWhat happens hereExtractLoad existing data if it exists, then call the Hub’Eau API (viarequests) for each gauging stationTransformTranslate French columns and entries to English, fix data types, remove duplicatesLoadWrite a CSV and a metadata file, then publish to Kaggle via the CLI### The Architecture, at a Glance

Notice that “Extract” already touches two different sources: theexistingdataset (what you already have) and thenewdata from the API. That distinction is the seed of the next idea.

Two Patterns that Make it Production-Grade

There two patterns that make this pipeline safe to run unattended, scheduled, for years.

1. Idempotency

Idempotencemeans when the same operation runs twice it gives the same result as if it runs once. Deduplication is what makes this pipeline idempotent. If the scheduler accidentally triggers twice, or a network retry fetches the same day again, the second run won’t create duplicate rows. This matters a lot for anything that runs on a schedule with no one watching it.

2. Incremental loading

A naïve pipeline would re-download theentirehistory on every run. That’s slow, it wastes your API quota, and it’s fragile: the more data you transfer, the more chances something fails.

Anincrementalpipeline avoids this. Instead, it:

• Checks the most recent date already in the dataset
• Requests only the datafrom that date onward
• Merges the new records into the existing dataset

You’ll find this logic built explicitly indetermine\_update\_range.

Always ask yourself:**“Do I actually need to do this?”**That one habit separates a fragile script from a production pipeline. For example, every “expensive” or “external” step in this pipeline (network calls, disk writes, publishing) is guarded by a cheap, local check first.

Part 2: Setup and Dependencies

Here’s the import block. It looks unremarkable, but how it’s organized is itself a best practice worth calling out.

# Standard library imports
import json
import os
import random
import subprocess
from dataclasses import dataclass, field
from datetime import date, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple

# Third-party libraries
import numpy as np
import pandas as pd
import requests

# Display config for notebooks
pd.set_option('display.max_columns', None)   # All columns will show
pd.set_option('display.max_colwidth', None)  # Prevents cutting long column text with ...

Code-level notes:

• Imports fall into three blocks:standard library, then third-party, then local, with blank lines between them. This followsPEP 8’s import-ordering convention. Mostauto-formatters(isort,ruff) enforce this same grouping.
• Nothing is imported withfrom module import \*. That syntax pollutes the current namespace. It makes it hard to trace where a name came from when someone debugs the code six months later. Python style guides echo this advice, includingGoogle’s Python Style Guide.
• Thepd\.set\_option\(\.\.\.\)calls exist purely fornotebook readability, so wide DataFrames don’t get truncated with\.\.\.. They have zero effect on the pipeline’s logic. You’d typically remove or scope them differently in a\.pyscript.

Part 3: Manage Configuration with Dataclasses

Why Bother with a Config Layer at All?

Every pipeline hasknobs: which stations to monitor, what the flood threshold is, and where to publish. The bad approach sprinkles these values as literals throughout the code as you write it. You end up with anif level \> 6000:buried three functions deep. Then changinganysetting means hunting through the whole file, and it’s easy to update one spot and miss another.

The fix:**centralize all settings in one place.**Python’s@dataclassdecorator is a natural fit for that.

FeatureWhy it mattersAuto-generated\_\_init\_\_,\_\_repr\_\_,\_\_eq\_\_You don’t need to write them yourselfType hintsGives you IDE autocomplete and self-documenting codeOptionalfrozen=TrueGives youtrue immutabilityif you want config knobs that can’t change after creation\_\_post\_init\_\_hookValidates or computes derived fields once, right after constructionCompare that to a configuration written as a plaindict:

config = {
    "stations": ["STN001", "STN002"],
    "flood_threshold": 6000,
    "publish_url": "https://example.com/alerts",
    "retry_count": 2,
    "timeout_seconds": 5,
}

A plain dict gives you none of that: no immutability, no type checking, and no autocomplete.

Code-Level Walkthrough

The pipeline defines three configuration classes. Each one hasa single responsibility: API details, station rules, and publishing destination. Each class becomes onemodule-level singleton instance. Every other function in the pipeline reads from these singletons.

@dataclass
class APIConfig:
    """API configuration for HubEau data fetching.

    Think of this as the "address book" for the API.
    """
    use_mock: bool = True
    base_url: str = "https://hubeau.eaufrance.fr/api/v2/hydrometrie/obs_elab"
    metric: str = "HIXnJ"  # Daily max water level (elaborated observations)
    # Pagination: fetch 20k records per request (API Limit)
    max_per_page: int = 20000
    timeout_seconds: int = 60  # Network timeout

    def __post_init__(self):
        """Validate configuration after initialization."""
        if self.max_per_page <= 0:
            raise ValueError("max_per_page must be positive")
        if self.timeout_seconds <= 0:
            raise ValueError("timeout_seconds must be positive")

Notice the validation inside\_\_post\_init\_\_. It runs right after the auto-generated\_\_init\_\_. A misconfiguredAPIConfig\(max\_per\_page=\-1\)fails loudly and immediately at startup, instead of surfacing as a bug later during an actual pipeline run.

@dataclass
class StationConfig:
    """Station monitoring configuration.

    The 'what' of data collection: which stations, what's a flood?
    """
    station_codes: List[str] = field(default_factory=lambda: [
        "F700000109", "F700000110", "F700000111",
        "F700000102", "F700000103",
    ])
    flood_threshold_mm: int = 6000  # Flood alert threshold
    earliest_date: str = "1900-01-01"  # How far back to go

**The mutable-default trap:**Look closely atstation\_codes. It isn’t written asstation\_codes: List\[str\] = \[\.\.\.\]. That’s deliberate, and it dodges one of the most common gotchas in Python. If you use a plain mutable object (a list, dict, or set) as a default argument or dataclass field,every instance shares the same underlying object. Mutate it on one instance, and you silently mutate it everywhere else too.

Stack Overflow covers this at length in its“least astonishment” mutable-default-argument discussion, as doesReal Python’s guide to optional arguments. The fix isfield\(default\_factory=\.\.\.\). It calls afreshfactory function (here, alambda) for every new instance, so each one gets its own independent list.

Explanation:

# Bad: shared mutable default
@dataclass
class BadConfig:
    station_codes: list[str] = []

a = BadConfig()
b = BadConfig()

a.station_codes.append("ALERT")
print(a.station_codes)  # ['ALERT']
print(b.station_codes)  # ['ALERT']  <-- same list!!

# Good: fresh list per instance
@dataclass
class GoodConfig:
    station_codes: list[str] = field(default_factory=list)

x = GoodConfig()
y = GoodConfig()

x.station_codes.append("ALERT")
print(x.station_codes)  # ['ALERT']
print(y.station_codes)  # []  <-- independent list

And the last Config singleton, the Kaggle settings:

@dataclass
class KaggleConfig:
    """Kaggle dataset publishing configuration."""
    dataset_slug: str = "grimespoint/paris-flood-dataset"
    input_csv: str = "kaggle/input/datasets/{slug}/paris_flood_dataset.csv"
    output_dir: Path = field(default_factory=lambda: Path("kaggle/working/kaggle_dataset"))
    mock_output_dir: Path = field(default_factory=lambda: Path("mock_output"))
    output_filename: str = "paris_flood_dataset.csv"
    mock_output_filename: str = "mock_flood_dataset.csv"
    metadata_filename: str = "dataset-metadata.json"

    # Metadata
    title: str = "Paris flood dataset"
    keywords: list = field(default_factory=lambda: [
        "tabular", "weather and climate", "environment", "europe", "time series analysis"
    ])
    geospatial_coverage: str = "Paris, France"
    update_frequency: str = "Weekly"
    license_name: str = "CC0-1.0"

    # Computed fields (set in __post_init__)
    output_csv_path: Path = field(init=False)
    metadata_path: Path = field(init=False)

    def __post_init__(self):
        """Compute derived paths after initialization."""
        self.input_csv = self.input_csv.format(slug=self.dataset_slug)
        self.output_csv_path = self.output_dir / self.output_filename
        self.metadata_path = self.output_dir / self.metadata_filename
        self.mock_output_filename = self.mock_output_dir / self.mock_output_filename

# Initialize configs - module-level singletons
API_CONFIG = APIConfig()
STATION_CONFIG = StationConfig()
KAGGLE_CONFIG = KaggleConfig()

This is the most common use of\_\_post\_init\_\_.output\_csv\_pathandmetadata\_pathare markedfield\(init=False\), so you can’t set them directly through the constructor. Instead,\_\_post\_init\_\_computes them from other fields (output\_dirandoutput\_filename).

Use this pattern forderived, computed values: compute them once, in one place, instead of recomputingoutput\_dir / output\_filenameevery time you need the path elsewhere in the codebase.

SeeReal Python’s data classes guideor the data classes chapter of O’Reilly’sFluent Pythonfor more on this pattern.

**Tip:**the auto-generated\_\_repr\_\_gives you a readable printout for free. Callprint\(API\_CONFIG\): it shows every field and value without a single line of formatting code. It’s handy for quick sanity checks when you’re debugging a pipeline run.

print(APIConfig)
# prints APIConfig(use_mock=True, base_url='https://hubeau.eaufrance.fr/api/v2/hydrometrie/obs_elab', ...)

The Extract phase pulls data from source systems and reads it into memory. Here, you genuinely havetwosources to extract from: theexistingdataset (what you already published last time) and thenew(recent) data from the Hub’Eau API.

Graceful File Loading

Design logic

What should happen the very first time this pipeline runs (and there’s no existing dataset yet)? A naïve implementation would crash with aFileNotFoundError.

Here’s the trick:load\_csv\(\)follows theNull Object pattern. Instead of raising an error, it returns anemptyDataFrame. Every downstream function can then treat “no existing data” and “some existing data” the same way, with no special-casing needed.

Code-level walkthrough

This function loads a CSV file into a DataFrame. If the file doesn’t exist, it returns an empty DataFrame instead of crashing.

low\_memory=Falsetells pandas to read the file carefully, so it avoids mixed-type guesses.parse\_dates=Truetries to automatically convert date-like columns into dates.delimiter=","tells pandas the file is comma-separated.

def load_csv(path: str) -> pd.DataFrame:
    """Load CSV file or return empty DataFrame if file does not exist.

    Args:
        path (str): Full path to the CSV file.

    Returns:
        pd.DataFrame: Loaded data, or empty DataFrame if file not found.

    Raises:
        pd.errors.ParserError: If the CSV is malformed.
    """
    if os.path.exists(path):
        return pd.read_csv(path, low_memory=False, parse_dates=True, delimiter=",")
    return pd.DataFrame()   # Null Object: consistent return type

To test, try it against a path that doesn’t exist:

df_missing = load_csv("/tmp/does_not_exist.csv")
print(df_missing.empty)  # True: no crash

# Callers can always do this, instead of an `is None` check:
if df_missing.empty:
    print("No existing data. Will run a full fetch from earliest date.")

Best practice:return aconsistent typefrom every code path in a function. A function that sometimes returns aDataFrameand sometimesNoneforces every caller to add aNonecheck before using the result.

Return aDataFrame, empty or not, which keeps things simpler. You never have to ask*“did I get a real result, or*None*?”*before using it.

Incremental Update Logic

Design logic

This is the “incremental loading” pattern from Part 1 in action. Before you fetch anything, ask:“What’s the most recent record I already have, and do I actually need more?”

**Strategy:**check whether the existing data already coversyesterday.

• Yes: skip the update entirely. Nothing to do, the dataset is current.
• No: fetch starting from the day after the last known date.

Existing data: Jan. 1 – Jan. 15
Yesterday: Jan. 19

Decision: fetch from Jan. 16 onwards (not from Jan. 1)

Whyyesterdayand nottoday? Today’s measurement might not be finalized on the source system yet. Hub’Eau’s “elaborated observations” are a processed daily aggregate, so the safest check is against the lastfully completedday.

Code-level walkthrough

determine\_update\_range\(\)checks the newest saved date. It tells you either “you’re already up to date” or “start downloading from this next date.”

Step by step:

1. The function starts with existing data - existingis a pandasDataFramethat already has some rows of data.
2. If there is no data at all - if existing\.empty: - If the DataFrame has zero rows, it says:“Nothing is saved yet, so fetch everything.” - It returns: - True= update needed - STATION\_CONFIG\.earliest\_date= start from the earliest allowed date
3. Find the date column - The code checks which column contains dates: - first tries"date\_obs\_elab" - then"record\_date" - If neither exists, it raises an error because it doesn’t know which column to use.
4. Convert the date column into real dates - pd\.to\_datetime\(\.\.\.\)turns the column into date objects pandas can work with. - errors="coerce"means bad date values become missing values instead of crashing.
5. Find the latest date in the data - last\_day = s\.max\(\)\.date\(\) - This gets the newest date already in the dataset.
6. Compare it with yesterday - yesterday = date\.today\(\) \- timedelta\(days=1\) - The function checks whether data already includes yesterday.
7. If data is already up to date - Iflast\_day \>= yesterday, it returns: - False= no update needed - None= no start date needed
8. If data is behind - It setsnext\_dayto the day after the last saved date. - Then it returns: - True= update needed - that next day as a string like"2026\-07\-12"

def determine_update_range(existing: pd.DataFrame) -> Tuple[bool, Optional[str]]:
    """Determine whether an update is needed and from what date.

    Logic:
    1. Check if existing data covers yesterday's date
    2. If yes → no update needed
    3. If no → start fetching from day after last data

    Returns:
        Tuple[should_update, start_date]
    """
    # Case 1: nothing on disk yet
    if existing.empty:
        print("No existing data found. Will fetch all data from earliest date.")
        return True, STATION_CONFIG.earliest_date

    if "date_obs_elab" in existing.columns:
        record_colname = "date_obs_elab"
    elif "record_date" in existing.columns:
        record_colname = "record_date"
    else:
        raise KeyError("Missing date column: expected 'date_obs_elab' or 'record_date'")

    s = pd.to_datetime(existing[record_colname], errors="coerce")
    last_day = s.max().date()
    yesterday = date.today() - timedelta(days=1)

    # Case 2: already current
    if last_day >= yesterday:
        print("\nDataset already covers yesterday or later. No update needed.")
        return False, None

    # Case 3: fetch the gap
    next_day = (last_day + pd.Timedelta(days=1))
    print(f"\nWill retrieve data starting from: {next_day}")
    return True, next_day.isoformat()

A few things worth a note here.

First, the function checks fortwo possible column names:date\_obs\_elab, the raw API name, or the already-renamed English namerecord\_date. It doesn’t assume just one. This makes the function work whether you call it on freshly-fetched raw data or an already-processed CSV loaded from disk.

errors="coerce"shows up here, and it’ll show up again (we’ll dig into this in Part 5). Any date pandas can’t parse becomesNaT(Not a Time) instead of raising an exception.

The return type isTuple\[bool, Optional\[str\]\]. Thistuplebundles two related results:*should I update, and from when?*That beats returning two separate values, or worse, one ambiguous value that means different things depending on context.

**Best practice:**useTuplereturn types (or, for more fields, a small dataclass orNamedTuple) to bundle related results together. Document clearly what each position means. If you return differenttypesfrom different code paths without documentation, you’ll hit a common source of confusion and bugs:“why is thisNonesometimes and a string other times?”.

Run a quick check against three scenarios:

# Test 1: No existing data
should_update, start_date = determine_update_range(pd.DataFrame())
# → True, "1900-01-01"

# Test 2: Old existing data (covers only Jan 10-15)
# → True, "2026-01-16"  (the day after the last known date)

# Test 3: Recent data that already covers yesterday
# → False, None

Simulate the API

Design logic

In production, fetching means a real HTTP call:

requests.get(
    "https://hubeau.eaufrance.fr/api/v2/hydrometrie/obs_elab",
    params={"code_entite": "F700000109", "size": 20000, ...}
)

Real API calls bring real challenges. Rather than fight those on your first read-through, this tutorial first builds and tests everything against amockgenerator. It returns data shaped exactly like the real API. Only once the logic works does it swap in the real endpoint.

This technique is useful well beyond this project. Build and test your transform logic against fixtures or mocks first. That way, you’re not into “is my parsing wrong?” and “is the network flaky right now?” at the same time.

Code-level walkthrough

generate\_mock\_api\_data\(\)generates fake sample data for a station, one record per day, starting fromstart\_date.

How it works:

• It convertsstart\_dateinto a real date.
• It loops fornum\_days.
• For each day, it creates a mock observation dictionary.
• It adds a small random variation to the water level so the data looks realistic.
• It randomly picks status, quality, and method labels.
• It returns a list of these dictionaries.

def generate_mock_api_data(station_code: str, start_date: str, num_days: int = 10) -> List[Dict]:
    """Generate realistic mock API data for demonstration.

    Simulates what HubEau API would return: list of observation dicts.
    """
    start = pd.to_datetime(start_date).date()
    records = []

    validation_statuses = ["Donnée validée", "Donnée brute", "Donnée pré-validée"]
    qualities = ["Bonne", "Non qualifiée", "Douteuse"]
    methods = ["Mesurée", "Calculée", "Expertisée"]

    for i in range(num_days):
        obs_date = start + timedelta(days=i)
        base_level = 5500 + int(station_code[-2:])  # Varies by station
        noise = random.randint(-200, 200)
        water_level = base_level + noise

        record = {
            "code_site": "mock_" + station_code[1:],
            "code_station": "mock_" + station_code,
            "date_obs_elab": obs_date.isoformat(),
            "resultat_obs_elab": water_level,
            "date_prod": (obs_date + timedelta(days=1)).isoformat(),
            "code_statut": "1",
            "libelle_statut": random.choice(validation_statuses),
            "code_methode": "1",
            "libelle_methode": random.choice(methods),
            "code_qualification": "1",
            "libelle_qualification": random.choice(qualities),
            "longitude": 2.3522 + random.uniform(-0.01, 0.01),
            "latitude": 48.8566 + random.uniform(-0.01, 0.01),
            "grandeur_hydro_elab": "mock_HIXnJ",
        }
        records.append(record)

    return records

Fetch Data for Real

Design logic

Two functions handle extraction. They’re deliberately split according to theSingle Responsibility Principle(SRP): each function should have one reason to change.

fetch_all_data() (orchestrator)
    ├── fetch_single_station_data(station_1) ← handles all complexity
    ├── fetch_single_station_data(station_2) ← handles all complexity
    └── fetch_single_station_data(station_n) ← handles all complexity

• fetch\_single\_station\_data\(\)ownsallthe messy per-station complexity: pagination, cursor advancement, stop conditions, and network error handling.
• fetch\_all\_data\(\)owns none of that. It just loops over stations and delegates.

This split has two payoffs. First, you can debug or swap out the pagination strategy for one station without touching the orchestration code at all. Second, if you ever want to parallelize fetching (withconcurrent\.futuresorasyncio, for example), the orchestrator is theonlyplace you’d need to touch.

Code-level walkthrough

fetch\_single\_station\_data\(\)fetches station data either from mock test data or from the real API, page by page, until it has everything it needs.

• Ifuse\_mock=True, it usesfake datainstead of calling the real API. 1. It callsgenerate\_mock\_api\_data\(\) 2. Turns the result into a DataFrame 3. Converts the date column into real pandas dates 4. Returns that DataFrame
• Ifuse\_mock=False, it does thereal API request: 1. Creates a reusable HTTP session 2. Starts fromstart\_date 3. Repeatedly asks the API for a page of data 4. Stops when: - the API returns no data - the latest date reaches yesterday - the page is smaller than expected - a network error happens 5. Combines all pages into one DataFrame 6. Returns an empty DataFrame if nothing was fetched

def fetch_single_station_data(station_code: str, start_date: str, use_mock: bool = True) -> pd.DataFrame:
    """Fetch all hydrometric data for a single station from mock API (default) or from the real endpoint.

    In production (cursor based pagination strategy):
    - Fetches max_per_page records per request
    - Continues until no new data or yesterday's date reached
    - Stop when: no data returned | last date >= yesterday | page was not full
    - Handles network errors gracefully
    """
    if use_mock:
        data = generate_mock_api_data(station_code, start_date, num_days=7)
        page_df = pd.DataFrame(data)
        page_df["date_obs_elab"] = pd.to_datetime(
            page_df["date_obs_elab"], errors="coerce").dt.normalize()
        return page_df

    # Real implementation
    else:
        session = requests.Session()  # Reuse TCP connection across pages
        frames = []
        cursor = start_date

        while True:
            params = {
                "code_entite": station_code,
                "grandeur_hydro_elab": API_CONFIG.metric,
                "date_debut_obs_elab": cursor,
                "size": API_CONFIG.max_per_page,
            }

            try:
                response = session.get(
                    API_CONFIG.base_url,
                    params=params,
                    timeout=API_CONFIG.timeout_seconds  # Best practice, always set
                )
                response.raise_for_status()
            except requests.RequestException as e:  # Don't let one station kill the whole pipeline
                print(f"Error fetching data for station {station_code}: {e}")
                break

            data = response.json().get("data", [])
            if not data:  # Empty response: we've exhausted this station
                break

            page_df = pd.DataFrame(data)
            page_df["date_obs_elab"] = pd.to_datetime(
                page_df["date_obs_elab"], errors="coerce").dt.normalize()
            frames.append(page_df)

            last_page_date = page_df["date_obs_elab"].max()
            yesterday = date.today() - timedelta(days=1)

            # Prevent infinite loops
            if pd.isna(last_page_date) or last_page_date.date() >= yesterday:
                break

            cursor = (last_page_date + pd.Timedelta(days=1)).strftime("%Y-%m-%d")

            if len(data) < API_CONFIG.max_per_page:
                break

    if frames:
        return pd.concat(frames, ignore_index=True)
    return pd.DataFrame()

How does the pagination loop actually work?

Let’s walk through it step by step. This is the densest bit of logic in the whole notebook.

1. Send a request withdate\_debut\_obs\_elab=cursor: “give me records from this date on.”
2. If the request fails outright (requests\.RequestException), log it andbreak.breakstops the loop right away and moves on. One station’s network hiccup shouldn’t kill the pipeline for every other station.
3. If the response has no data at all, you’ve caught up:break.
4. Otherwise, note thelatestdate seen on this page (last\_page\_date).
5. If that latest date is already\>= yesterday, you’ve caught up:break.
6. Otherwise, advance the cursor tolast\_page\_date \+ 1 dayand loop again for the next page.
7. As a safety net: if the page returnedfewerrecords thanmax\_per\_page, that also means you’ve reached the end. The API wouldn’t return a partial page unless it ran out of data, so:break.

That last check (step 7) is a classicpagination termination heuristic. You don’t always need anext\_pagetoken from the API. If a full page issize=20000and you get back only4213records, there’s nothing left to fetch.

Three specific, deliberate choices, called out:

session = requests.Session()

ASessionobject reuses the underlying TCP connection across multiple requests to the same host. That avoids a fresh TCP/TLS handshake on every single page request. In short, it’s faster for you and more polite to Hub’Eau’s servers.

**Best practice:**any time you callrequests\.get\(\)more than once against the same host in a loop, reach for aSession.

except requests.RequestException as e:

RequestExceptionis the base class forevery exceptionrequestscan raise: timeouts, connection errors, HTTP errors fromraise\_for\_status\(\)and more. Catching the base class here meansanynetwork hiccup gets handled the same forgiving way: log it, stop fetching this station, move on.

timeout=API_CONFIG.timeout_seconds

By default,requestscallsnever time out. Without an explicit timeout, a hung server can freeze your entire pipeline indefinitely. Therequests advanced usage docsstates that requests to external servers should have a timeout attached.

**Best practice:**wrap every external I/O call intry/except. Always fail gracefully. Log the error and let the pipeline recover or move on. This avoids one flaky request taking down an unattended weekly job.

Now the orchestrator,fetch\_all\_data\(\), is deliberately much simpler:

1. Loop through all station codes. - Fetch each station’s data. - Keep the non-empty results.
2. Combine them into one big DataFrame.

def fetch_all_data(start_date: str, use_mock: bool = True) -> pd.DataFrame:
    """Orchestrator: Fetch data for all configured stations."""
    frames = []

    for station_code in STATION_CONFIG.station_codes:
        print(f"Fetching data for station {station_code}...")
        df_station = fetch_single_station_data(station_code, start_date, use_mock=use_mock)

        if not df_station.empty:
            print(f"  Got {len(df_station)} records")
            frames.append(df_station)
        else:
            print(f"  (no data)")

    if frames:
        return pd.concat(frames, ignore_index=True)
    return pd.DataFrame()

That’s it. A loop and apd\.concat. All the hard-won complexity lives one layer down, exactly where SRP says it should.

Part 5: The Transform Step

This is where raw, freshly-fetched data becomes something you can publish. Here’s the design philosophy for this whole section: compose many small,pure functions. Each one takes a DataFrame in and returns anewDataFrame out without side effects. Avoid the one giant do-everything function trap.

(EXTRACT)
    Raw API Data
    ↓
    (TRANSFORM)
1. Type parsing (datetime, numeric)
2. Column renaming (French → English)
3. Categorical mapping (validation status, quality)
4. Derived columns computation (flood alert flags)
5. Column reordering (logical grouping)
6. Sorting & index reset
    ↓
    (LOAD)
    Publication-ready dataset

Why split this into six tiny steps instead of one big function? Each piece is testable and replaceable on its own. When something breaks at 3am on a scheduled run, every function is debuggable in isolation. You can pinpoint exactly which stage produced bad output. No need to pick apart one 200-line function.

Type Parsing and Graceful Coercion

Design logic

Data from a CSV or a JSON API starts out, by default, as strings. Pandas needs real types to sort dates chronologically, do date arithmetic (last\_date \+ timedelta\(days=1\)), or compare numeric values (water\_level \> 6000). Type mismatches are a common error that break batch pipelines. One malformed row, like"N/A", a truncated date, misaligned values, or stray characters and a strict parser throws an exception that kills the whole run.

**Best practice:**both conversion functions below useerrors="coerce", so values pandas can’t parse becomeNaT(Not a Time) orNaN(Not a Number) instead of raising an error. This is calledgraceful coercion.

Code-level walkthrough

convert\_to\_date\(\)andconvert\_to\_numeric\(\)are small helper functions that make sure certain columns have the right type.

• Both start withdf\.copy\(\)so they don’t change the original DataFrame.
• errors="coerce"means bad values become missing values instead of causing a crash.

Forconvert\_to\_date\(\):

1. df\.copy\(\)makes a separate copy, so the original table stays unchanged
2. The loop goes through each column name incolumns - if col in df\.columnschecks that the column actually exists before trying to convert it - pd\.to\_datetime\(\.\.\.\)turns text like"2026\-07\-12"into real pandas date/time values - errors="coerce"means invalid values becomeNaT(missing date) instead of raising an error - \.dt\.normalize\(\)removes the time part and keeps only the date at midnight

Forconvert\_to\_numeric\(\):

1. df\.copy\(\)is used here as well.
2. The loop goes through each column name incolumns - if col in df\.columnschecks that the column actually exists before trying to convert it - The loop callspd\.to\_numeric\(\)turns text like"12\.5"into numbers - errors="coerce"turns bad values intoNaNinstead of crashing

def convert_to_date(df: pd.DataFrame, columns: List[str]) -> pd.DataFrame:
    """Convert specified columns to pandas datetime type."""
    df = df.copy()  # Never modify the original!
    for col in columns:
        if col in df.columns:
            df[col] = pd.to_datetime(df[col], errors="coerce").dt.normalize()
    return df

def convert_to_numeric(df: pd.DataFrame, columns: List[str]) -> pd.DataFrame:
    """Convert specified columns to numeric (float) type."""
    df = df.copy()
    for col in columns:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce")
    return df

Why is this useful?

• You can aggregate the data: group it, summarize it and so on...
• Dates become sortable and filterable as real dates.
• Numbers work correctly in calculations like averages, sums, or comparisons.
• It prevents bugs caused by mixed types, like"12"and12.

Bothpd\.to\_datetimeandpd\.to\_numericare official pandas functions with anerrorsparameter. By default, that parameter is set to"raise", which throws on bad input. The other options are"coerce"(replace with null) or"ignore"(leave untouched). See thepandasto\_datetimedocsandto\_numericdocsfor the full parameter list.

Test it against an example of messy input:

messy_df = pd.DataFrame({
    "date_obs_elab": ["2026-01-15", "2026-01-16", "not a date", None],
    "resultat_obs_elab": [5800.0, "5900", "N/A", None],
})

type_safe_df = convert_to_date(messy_df, ["date_obs_elab"])
type_safe_df = convert_to_numeric(type_safe_df, ["resultat_obs_elab"])

# "not a date"  → NaT
# "N/A"         → NaN
# Pipeline continues safely — nothing crashed.

**Best practice:**don’t let one bad row kill an entire run. Coerce bad data to nulls rather than raising exceptions. Flag or log nulls separately if you need to investigate data quality later.

This is a deliberate trade-off. A choice betweenavailability(the pipeline keeps running) overstrictness(catching every bad row immediately). That’s usually the right call for a scheduled, unattended job.

df\.copy\(\): immutability by convention

Look again at the top of both functions above:df = df\.copy\(\). This single line appears at the start ofevery transform functionin the pipeline, and that’s not an accident.

Python DataFrames are mutable objects, passed by reference. If a function modifiesdfin place without copying first, the caller’s original DataFrame changes too. That’s a classicside effectand it can produce genuinely confusing bugs. Call\.copy\(\)first means each function’s output is a brand-new object. The input the caller passed in staysguaranteed untouched.

Best practice:treat DataFrames asimmutable inputs. Return a new DataFrame rather than modify it one in place. Even if it costs a small amount of memory or CPU, the debugging win is almost always worth it for a pipeline that isn’t operating at extreme scale.

Finally, one auto-detecting convenience function wraps these two low-level functions. It scans text columns, guesses whether they contain dates or numbers, and calls the matching parsing function above.

auto\_convert\_columns\(\)tries toguess which columns are dates or numbersand then fixes these types automatically.

How it works:

It starts with two empty lists:

• datetime\_colsfor date columns
• numeric\_colsfor number columns

It goes through each column in the DataFrame. If a column is already a real datetime or numeric type, it skips it. If the column is text-like (objectorstring), it looks at up to 10 non-empty sample values.

It first tries to read those values as dates:

• if that works, the column is added todatetime\_cols

If not, it tries to read them as numbers:

• if that works, the column is added tonumeric\_cols

At the end, it converts all date columns withconvert\_to\_date\(\). Then it converts all numeric columns withconvert\_to\_numeric\(\)

def auto_convert_columns(df: pd.DataFrame) -> pd.DataFrame:
    """Auto-detect and convert datetime and numeric columns to the correct type."""
    datetime_cols = []
    numeric_cols = []

    for col in df.columns:
        if pd.api.types.is_datetime64_any_dtype(df[col]):
            continue
        if pd.api.types.is_numeric_dtype(df[col]):
            continue

        if pd.api.types.is_object_dtype(df[col]) or pd.api.types.is_string_dtype(df[col]):
            sample = df[col].dropna().head(10)
            if len(sample) == 0:
                continue

            try:
                pd.to_datetime(sample, errors='raise', format='mixed')
                datetime_cols.append(col)
                continue
            except (ValueError, TypeError):
                pass

            try:
                pd.to_numeric(sample, errors='raise')
                numeric_cols.append(col)
                continue
            except (ValueError, TypeError):
                pass

    df = convert_to_date(df, datetime_cols)
    df = convert_to_numeric(df, numeric_cols)
    return df

Notice the innertry/exceptblocks here useerrors='raise'. That’s theoppositeof the coercion strategy above, but it only runs against a small\.head\(10\)sampleof each column.

This is a type-sniffingstep, not the final conversion. It tests*“does this column look like dates, or numbers, or neither?”on a cheap sample. Then, it hands off the actual, forgiving conversion of thewhole*column toconvert\_to\_date\(\)orconvert\_to\_numeric\(\). Two differenterrorsstrategies, two different jobs.

Schema Translation with Bidirectional mappings

Design logic

The Hub’Eau API returns French column names and French categorical values, likecode\_stationor"Donnée validée". A dataset meant for an international audience should ship in English. If you rename columns inline, wherever it’s convenient, the mapping between French and English ends up scattered across the codebase. Then you’d have no way to reverse it if you ever needed to.

The fix:defineone authoritative mappingat the top of the module, and derive everything else from it.

Code-level walkthrough

API\_TO\_ENmaps API field names to English translations.

# Primary mapping: French API columns to English column names
API_TO_EN = {
    "code_site": "location_code",
    "code_station": "station_code",
    "date_obs_elab": "record_date",
    "resultat_obs_elab": "water_level_mm",
    "date_prod": "data_production_date",
    "code_statut": "validation_status_code",
    "libelle_statut": "validation_status",
    "code_methode": "production_method_code",
    "libelle_methode": "production_method",
    "code_qualification": "quality_code",
    "libelle_qualification": "quality_assessment",
    "longitude": "longitude",
    "latitude": "latitude",
    "grandeur_hydro_elab": "hubeau_elab_code",
}

# Reverse mapping: English to French (computed automatically)
EN_TO_API = {v: k for k, v in API_TO_EN.items()}

The second line (EN\_TO\_API) is just a shortcut: it swaps each key and value fromAPI\_TO\_EN. Adict comprehensionbuilds it.Here’s the important design point:EN\_TO\_APIisn’t hand-maintained, it’sderived. If you add, remove, or rename an entry inAPI\_TO\_EN,EN\_TO\_APIupdates automatically the next time the module runs.

There’sexactly one placein the entire codebase where a schema change needs to happen.

Categoricalvalues(not just column names) get the same treatment:

CATEGORICAL_MAPPINGS = {
    "validation_status": {
        "Donnée validée": "validated",
        "Donnée brute": "raw",
        "Donnée pré-validée": "pre-validated",
    },
    "quality_assessment": {
        "Bonne": "good",
        "Non qualifiée": "unqualified",
        "Douteuse": "dubious",
    },
    "production_method": {
        "Calculée": "calculated",
        "Mesurée": "measured",
        "Expertisée": "expert-reviewed",
    },
}

The functions below apply these mappings, in order, to standardize column names and category values.

rename\_to\_english\(\):

1. If the DataFrame is empty, it returns a copy right away.
2. It builds a list of columns that are renamed from API names to English names. - It only renames a column if: - the old name exists, and - the new name does not already exist
3. Then it returns the renamed DataFrame.

rename\_to\_api\_schema\(\):

1. Also returns a copy for empty data.
2. Does the reverse: English names back to API names.
3. Returns the DataFrame.

(Useful if you need to send data back in the API’s original format).

apply\_categorical\_mappings\(\):

1. Makes a copy so the original DataFrame is not changed.
2. For each column inCATEGORICAL\_MAPPINGS, it replaces values using the mapping. - Example:"Donnée validée"becomes"validated". - \.fillna\(df\[col\_name\]\)keeps the original value if a value isn’t found in the mapping.
3. Returns the DataFrame.

def rename_to_english(df: pd.DataFrame) -> pd.DataFrame:
    """Rename API column names to English schema names."""
    if df.empty:
        return df.copy()

    columns_to_rename = {}
    for src, dst in API_TO_EN.items():
        if src in df.columns and dst not in df.columns:
            columns_to_rename[src] = dst

    return df.rename(columns=columns_to_rename)

def rename_to_api_schema(df: pd.DataFrame) -> pd.DataFrame:
    """Rename English column names back to API schema names (defensive/reverse operation)."""
    if df.empty:
        return df.copy()

    columns_to_rename = {k: v for k, v in EN_TO_API.items() if k in df.columns}
    return df.rename(columns=columns_to_rename)

def apply_categorical_mappings(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    for col_name, mapping in CATEGORICAL_MAPPINGS.items():
        if col_name in df.columns:
            df[col_name] = df[col_name].map(mapping).fillna(df[col_name])
    return df

Two defensive habits are worth a note here:

**Robustness to partial inputs:**Both rename functions only rename columns that actually exist in the input (if src in df\.columns).

A rename function that assumes every mapped column is always present will crash the moment it’s called on a partial or differently-shaped DataFrame.

\.map\(mapping\)\.fillna\(df\[col\_name\]\).Series\.mapreplaces every value it finds in the mapping dict and turns any valuenotfound in the dict intoNaN. Chain\.fillna\(df\[col\_name\]\)right after restores theoriginalvalue wherever the mapping didn’t apply. That way, unexpected categorical values pass through unchanged instead of silently becoming null. It’s a subtle but important robustness choice.

See this exact\.map\(\)-then-\.fillna\(\)idiom discussed onStack Overflow: pandas map vs apply performance.

A quick round-trip test proves the bidirectional mapping actually works:

sample_api_df = pd.DataFrame({"code_station": ["F700000109"], ...})
renamed_df = rename_to_english(sample_api_df)          # French → English
reversed_df = rename_to_api_schema(renamed_df)          # English → French
# reversed_df.columns.tolist() == sample_api_df.columns.tolist()  → True

Compute Flood Alerts

add\_derived\_columns\(\)is a one-liner. It just computesflood\_alert. If the water level is greater than the flood threshold set in theConfig, the result isTrue. Otherwise, the result isFalse.

def add_derived_columns(df: pd.DataFrame) -> pd.DataFrame:
    """Add a computed columns based on raw data.
    """
    df = df.copy()

    if "water_level_mm" in df.columns:
        df["flood_alert"] = df["water_level_mm"] > STATION_CONFIG.flood_threshold_mm

    return df

Column Ordering

Here’s a small but user-facing detail: define a preferred column order once, as data, and reuse it everywhere.COLUMN\_ORDERholds that preferred column sequence.

order\_columns\(\)rearranges a DataFrame so those columns come first, while any extra columns stay at the end.

COLUMN_ORDER = [
    # Primary identifiers & measurements
    "station_code", "record_date", "water_level_mm", "flood_alert",
    # Metadata about the observation
    "hubeau_elab_code", "data_production_date",
    "validation_status_code", "validation_status",
    "production_method_code", "production_method",
    "quality_code", "quality_assessment",
    # Geographic info (less important)
    "location_code", "longitude", "latitude",
]

def order_columns(df: pd.DataFrame) -> pd.DataFrame:
    """Reorder columns to preferred order."""
    present_cols = [c for c in COLUMN_ORDER if c in df.columns]
    other_cols = [c for c in df.columns if c not in present_cols]
    return df[present_cols + other_cols]

other\_colsacts as a safety net: any column not explicitly listed inCOLUMN\_ORDERstill gets included at the end. They’re not silently dropped.

Rules, in order of priority:

• Keep identifiers and key fields up front.
• Put the most important, most frequently used, and most stable columns first.
• Group related fields together, so the table reads naturally.
• Push optional or rarely-used fields to the end.

Deduplication

Design logic

Remove duplicates as you go. In an incremental pipeline, date ranges and records can overlap:

• The same day might get re-fetched, because API data arrives late or gets finalized later.
• The same observation might appear on two different pages of a paginated response.

Without deduplication, these scenarios let duplicate rows pile up in the dataset over time. This is also, recall from Part 1, exactly what makes the pipelineidempotent: run it once or run it five times and the resulting dataset stays identical.

The fix requires defining what makes a recordunique. Define a composite key:

key = (station_code, observation_date, water_level_value)

Why this specific combination? Physically, one sensor (station\_code) reports one day’s (observation\_date) daily-maximum reading (water\_level\_mm), and that reading should be unique.

• Records from different stations obviously aren’t duplicates of each other.
• Two readings on different days aren’t duplicates.
• A subtler point: if thesamestation reports thesameday but with adifferentvalue, that counts as a distinct observation, for example a corrected or revised measurement, not a duplicate to silently discard.

Code-level walkthrough

create\_dedup\_key\(\)builds a unique text ID for each row by joining three pieces together with an underscore:

• station code
• date, as YYYY-MM-DD
• water level value

Example key:

"F700000109_2024-01-15_5800.0"

def create_dedup_key(df: pd.DataFrame) -> pd.Series:
    """Create unique deduplication key from station, day, and water level value.

    Key format: "station_code_YYYY-MM-DD_value"
    Example: "F700000109_2024-01-15_5800.0"
    """
    parts = []

    if "code_station" in df.columns:
        parts.append(df["code_station"].astype(str))

    if "date_obs_elab" in df.columns:
        parts.append(df["date_obs_elab"].dt.strftime("%Y-%m-%d"))

    if "resultat_obs_elab" in df.columns:
        parts.append(df["resultat_obs_elab"].astype(str))

    if not parts:
        return pd.Series(index=df.index, dtype="object")

    return pd.Series(
        ["_".join(row) for row in zip(*parts)],
        index=df.index
    )

How the key-building works:partsends up as a list ofSeries, one per key component (station, date, value), each the same length as the DataFrame.

Read the last line from the inside out:

return pd.Series(["_".join(row) for row in zip(*parts)], index=df.index)

zip\(\*parts\)transposes that list of columns into row-wise tuples. It yields\(station\_1, date\_1, value\_1\), then\(station\_2, date\_2, value\_2\), and so on. Thelist comprehensionthen joins each row-tuple with underscores into one string key per row.

This*“list of columns → zip → row tuples”*idiom is a common and efficient way to combine several Series into one derived Series, without writing\.apply\(lambda row: \.\.\., axis=1\). Row-wise\.applyis notoriouslyslow in pandascompared tovectorizedstring operations.

The actual deduplication happens inremove\_duplicates\(\). It removes rows from*“new”(freshly fetched data) that already exist in“existing”*(historic data).

Step by step:

1. If one table is empty, it just returnsnew.
2. It enforces types in both DataFrames withauto\_convert\_columns\(\)first, so dates and numbers compare correctly.
3. It creates a deduplication key for each row in both tables withcreate\_dedup\_key\(\).
4. It checks which keys from the fetched data don’t appear in the existing historic dataset, and keeps only truly new rows.
5. It returns the filtered result, keeping the original columns fromnew.

def remove_duplicates(existing: pd.DataFrame, new: pd.DataFrame) -> pd.DataFrame:
    """Remove rows from 'new' that already exist in 'existing'."""
    # Short-circuit: if either is empty, no work to do
    if existing.empty or new.empty:
        return new.copy()

    # Parse types on both sides for a fair comparison
    existing_std = auto_convert_columns(existing)
    new_std = auto_convert_columns(new)

    # Build the keys
    existing_keys = set(create_dedup_key(existing_std).dropna())
    new_keys = create_dedup_key(new_std)

    # Boolean mask: True where the new row is genuinely new
    mask = ~new_keys.isin(existing_keys)

    # Index back into the ORIGINAL (non-standardized) 'new' to preserve all columns
    result = new.iloc[new_keys[mask].index].copy()
    return result

Two performance and robustness details are worth to note:

First,existing\_keysis aset, not alist.Testing membership (in/\.isin\(\)) against a PythonsetisO(1)on average, because it uses a fast hash lookup to answer*is this item already here?*directly. Testing against alistisO(n): it has to scan item by item and it gets slower as the existing dataset grows.

For a dataset with tens of thousands of rows, checked on every single pipeline run, that difference matters and that’s a textbook example of choosing the right data structure for the job. See the general discussion oflist vs. set lookup performance in Pythonand thisguidanceon\.isin\(\)performance for filtering.

**Simple rule:**use asetwhen you care about fast membership checks. Use alistwhen you care about order or duplicates.

Second,new\.iloc\[new\_keys\[mask\]\.index\]indexes back into theoriginal, non-validatednewDataFrame, notnew\_std.

Why?auto\_convert\_columns\(\)only ran to getconsistent types for comparison. The caller still wants theoriginalraw values and schema back for everything that survives deduplication.

Don’t let a side-computationaccidentallybecome your source of truth. Always modify a copy only to decide what to keep or remove. Once you’ve made that decision, apply it to the original data. That way you preserve the real source values for later steps.

In short:

1. First modifications are only for comparison.
2. Use that comparison to filter the original data.
3. Return the original rows unchanged, so later stages can process them.

**Best practice:**keep your deduplication key simple and stable (immutable) and always short-circuit (withif existing\.empty or new\.empty: return new\.copy\(\)) before doing any heavier work. This cheap guard clause skips the expensive deduplication logic when there’s no data to process.

Put it All Together inpostprocess\(\)

All the pieces above are small and testable on their own. The last step ofTransformglues them together,in order, into the single pipeline function below:

def postprocess(df: pd.DataFrame) -> pd.DataFrame:
    """Apply all post-processing transformations."""
    if df.empty:
        return df

    df = df.copy()

    print("  1. Converting types...")
    df = auto_convert_columns(df)

    print("  2. Renaming columns (French → English)...")
    df = rename_to_english(df)

    print("  3. Mapping categorical values...")
    df = apply_categorical_mappings(df)

    print("  4. Adding derived columns...")
    df = add_derived_columns(df)

    print("  5. Reordering columns...")
    df = order_columns(df)

    print("  6. Sorting and resetting index...")
    df = df.sort_values(["record_date", "station_code"]).reset_index(drop=True)

    return df

Notice the shape of this function: it’s basically a linear script with six numbered, printed steps. Each one calls a previously-defined pure function. There’s no newlogichere, onlysequencing. That’s intentional.

**Best practice:**log progress clearly at each stage. When a scheduled job fails at 3am, a clear step-by-step log is the difference between a two-minute diagnosis and an hour of guessing. A good option is to use alogger. Here, this tutorial sticks with classic console printing (print\(f" 1\. Converting types\.\.\."\)).

Part 6: The Load Step

Design Logic

Loadis the final stage: take the processed data and move it to its destination. Typical destinations include data warehouses, data lakes, and databases.

Before publishing, the pipeline prepares two things: the output folder itself, and a metadata file that describes the dataset.

Code-level walkthrough

create\_output\_dir\(\)makes sure an output folder exists, and creates one if it doesn’t:

• Ifuse\_mockisTrue, it usesKAGGLE\_CONFIG\.mock\_output\_dir, the directory for fake data.
• Otherwise, it uses the real output directory set inKAGGLE\_CONFIG\.output\_dir.

def create_output_dir(use_mock: bool = False) -> None:
    """Create output directory (and parents) if it does not already exist."""
    output_dir = KAGGLE_CONFIG.mock_output_dir if use_mock else KAGGLE_CONFIG.output_dir
    output_dir.mkdir(parents=True, exist_ok=True)
    # exist_ok=True: idempotent, safe to call multiple times

exist\_ok=Trueis a small but important detail. Without it,Path\.mkdir\(\)raisesFileExistsErrorif the directory already exists, which it will on every run after the first.

Settingexist\_ok=Truemakes directory creationidempotent: calling it 100 times has the same effect as calling it once. On the filesystem, this is the “safe to re-run” idea behind the deduplication logic from Part 5.

The Kaggle API follows theData Package specification. Itrequiresa descriptivedataset\-metadata\.jsonfile alongside the CSV. This matters for the search and discoverability of theParis flood dataset:

def create_metadata(df: pd.DataFrame, config: KaggleConfig) -> Dict:
    """Generate Kaggle dataset metadata from DataFrame and config."""
    if df.empty or "record_date" not in df.columns:
        first_date = "unknown"
        last_date = "unknown"
    else:
        first_date = df["record_date"].min().strftime("%Y-%m-%d")
        last_date = df["record_date"].max().strftime("%Y-%m-%d")

    return {
        "title": config.title,
        "id": config.dataset_slug,
        "licenses": [{"name": config.license_name}],
        "keywords": config.keywords,
        "temporalCoverage": {"startDate": first_date, "endDate": last_date},
        "geospatialCoverage": config.geospatial_coverage,
        "updateFrequency": config.update_frequency,
    }

Metadata comes from theKaggleConfigdataclass. Note thattemporalCoverageis computedfrom the data itself(df\["record\_date"\]\.min\(\)/\.max\(\)) rather than hardcoded. Every time the pipeline runs, the metadata’s date range automatically reflects reality, with zero manual bookkeeping.

Finally,publish\_to\_kaggle\(\)uploads the updated dataset to Kaggle. It calls out to theKaggle CLIas an external command.

What it does:

1. Gets the current time and turns it into a text label likeWeekly update: 2026\-07\-13 14:30:00
2. Builds a Kaggle command that says: - publish a new dataset version - use files fromKAGGLE\_CONFIG\.output\_dir - attach the message - zip the directory contents
3. Prints the command so you can see what will run
4. Runs the command withsubprocess\.run\(\.\.\.\)

If the upload works:

• it printsSuccessfully published to Kaggle\.

If it fails:

• it prints the error and then raises the error again so the failure isn’t hidden.

def publish_to_kaggle() -> None:
    """Publish the updated dataset to Kaggle using the Kaggle CLI."""
    timestamp = pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
    message = f"Weekly update: {timestamp}"

    cmd = [
        "kaggle", "datasets", "version",
        "-p", str(KAGGLE_CONFIG.output_dir),
        "-m", message,
        "--dir-mode", "zip",
    ]

    print("Publishing to Kaggle...")
    print("Command:", " ".join(cmd))

    try:
        subprocess.run(cmd, check=True)
        print("Successfully published to Kaggle.")
    except subprocess.CalledProcessError as e:
        print(f"Error publishing to Kaggle: {e}")
        raise

publish\_to\_kagglecalls the Kaggle CLI (Command Line Interface) through Python’ssubprocessmodule. This is the standard way to run an external command-line tool from a Python pipeline.

Two choices here are worth adopting as general habits, not just for Kaggle:

• cmdis built as alist**, never as a concatenated string.**Pass a list of arguments tosubprocess\.runskips invoking a shell entirely. That sidestepsshell-injectionrisks and correctly handles arguments containing spaces or special characters. See theofficialsubprocesssecurity considerations.
• check=Truemeans that if the Kaggle CLI fails, by exiting with a non-zero status,subprocess\.runraisesCalledProcessErrorinstead of silently returning. Withoutcheck=True, a failed publish would look exactly like a successful one to the rest of the code.

Part 7: Assemble the Full Pipeline

The Global Rehearsal (Mock Mode)

The whole pipeline is built and tested in isolation. It’s time to wire the blocks into one linear function. First, run it entirely against mock data. Nothing gets published this way. For Demonstration purposes, a CSV filled with artificial data will land in a localmock\_outputfolder.

def run_etl_pipeline(use_mock: bool = True) -> pd.DataFrame:
    """Run the complete ETL pipeline: Extract → Transform → Load to CSV."""

    # STEP 1: LOAD EXISTING DATA
    if use_mock:
        loaded_df = pd.DataFrame(mock_api_response)
        loaded_df['date_obs_elab'] = pd.to_datetime(loaded_df['date_obs_elab'])
        loaded_df = rename_to_english(loaded_df)
    else:
        loaded_df = load_csv(KAGGLE_CONFIG.input_csv)

    # STEP 2: DETERMINE UPDATE RANGE
    should_update, start_date = determine_update_range(loaded_df)
    if not should_update:
        return loaded_df   # already current: nothing more to do

    # STEP 3: EXTRACT (FETCH DATA)
    fetched_data = fetch_all_data(start_date, use_mock=use_mock)

    # STEP 4: TRANSFORM I - DEDUPLICATE
    deduped_fetched_data = remove_duplicates(loaded_df, fetched_data)

    # STEP 5: TRANSFORM II - COMBINE (MERGE WITH EXISTING)
    new_df_english = rename_to_english(deduped_fetched_data)
    merged_historical_and_new = pd.concat([loaded_df, new_df_english], ignore_index=True)

    # STEP 6: TRANSFORM III - POST-PROCESS
    processed_records = postprocess(merged_historical_and_new)

    # STEP 7: EXPORT (LOAD TO CSV)
    create_output_dir(use_mock=use_mock)
    output_path = KAGGLE_CONFIG.mock_output_filename if use_mock else KAGGLE_CONFIG.output_csv_path
    processed_records.to_csv(output_path, index=False, sep=",")

    return processed_records

Every one of the seven steps above corresponds to a function you already built, and tested earlier in this tutorial.**To Assemble them is almost mechanical.**That’s always the payoff of composing small, single-responsibility functions:

loaded_df                 = load_csv(...)                          # 1. load existing data
should_update, start_date = determine_update_range(...)            # 2. check what's needed
fetched_data               = fetch_all_data(start_date)             # 3. fetch new records
deduped_fetched_data       = remove_duplicates(existing, new_raw)   # 4. deduplicate
merged_historical_and_new  = pd.concat([existing, new_clean])       # 5. merge
processed_records          = postprocess(merged_historical_and_new) # 6. postprocess
processed_records.to_csv(...)                                       # 7. save
write_metadata() + publish_to_kaggle()                               # 8. publish

Each line’s intent is obvious just from reading it left to right. That is exactly the point of good decomposition.

main\(\): Pipeline Orchestration

main\(\)is theconductor. It wires every previously-built component together in the right order, just likepostprocess\(\)did one level down.

def main() -> None:
    """Execute the full Paris Flood Monitoring ETL pipeline.

    EXTRACT
    1. Load the existing dataset from the Kaggle input mount
    2. Determine whether an update is needed (and from what date)
    3. Fetch new data from the Hub'Eau API

    TRANSFORM
    4. Deduplicate against the existing dataset
    5. Combine and post-process (type parsing, translation, derived cols...)

    LOAD
    6. Write the updated CSV and metadata file
    7. Publish the new dataset version to Kaggle

    Exits early if the dataset is already up to date.
    """
    final_dataset = run_etl_pipeline(use_mock=False)
    final_dataset.to_csv(KAGGLE_CONFIG.output_csv_path, index=False)
    write_metadata(final_dataset)

    print("\n[FINAL STEP] Publishing to Kaggle...")
    publish_to_kaggle()

    print("Running post-run validation")
    validate_and_analyze(final_dataset)

if __name__ == "__main__":
    main()

Theif \_\_name\_\_ == "\_\_main\_\_":Guard

if __name__ == "__main__":
    main()

This is one of the most common idioms in Python, and it’s worth understanding exactly what it does. Per theofficial Python documentationon\_\_main\_\_:

• Run the file directly (python script\.py) - → the special variable\_\_name\_\_is set to"\_\_main\_\_" - → the condition isTrue - →main\(\)executes.
• Importthe file as a module elsewhere (import script) - →\_\_name\_\_is set to the module’s name instead - → the condition isFalse - →main\(\)doesnotrun automatically.

Best practice:always guard your entry point this way. It makes the module safelyimportablefor testing individual functions, or for reuse in another script, without triggering the full pipeline (including a real publish to Kaggle!) just by importing it.

Part 8: Go Live and Switch to the Real API

Everything above runs safely against mock data. To point the pipeline at the real Hub’Eau API instead:

1. Note theuse\_mock=Falsesetting above. It threads throughAPIConfig\.use\_mock, and the calls torun\_etl\_pipeline\(use\_mock=False\)andmain\(\).
2. Make sureKAGGLE\_CONFIG\.dataset\_slugandinput\_csvpoint atyour ownKaggle dataset copy before you use it. You can only publish new versions of a dataset you own. Fork both thenotebookand thedataset, then update the slug in theKaggleConfigsection.
3. Install and authenticate theKaggle CLIif you plan to callpublish\_to\_kaggle\(\)outside of a Google Colab or Kaggle notebook.

Everything else needszero changes.

Post-Run Validation

validate\_and\_analyze\(\)closes the loop, and it’s more than just a nice-to-have. It’s a welcome sanity check that runsafterthe pipeline finishes. The output is a human-readable report covering shape, data types, null counts, summary statistics onwater\_level\_mm, how many flood-alert records turned up, the temporal coverage and per-station record counts.

It doesn’t change any data. It exists so that whoever, or whatever monitoring system, reads the run’s log output can immediately see whether this week’s numbers look sane. No need to analyze the CSV by hand. Try it!

Summary

You’ve just built acompleteETL pipeline that you can run again and again without breaking anything. The skeleton is here, and that same pattern repeats for almost any scheduled data job: swap out the API, the field mapping, and the destination.

For more tutorials like this one, check out myGitHuborKaggle profile.

Thanks for reading!

References

• Hub’Eau API: France’s official open water-data platform
• Theupdater\.pyscriptruns live once a week and updates the originaldataset.
• The source dataset generator is available onGitHub.
• The completeJupyter notebookto follow this tutorial and code all along (backup available).

Learn to code for free. freeCodeCamp’s open source curriculum has helped more than 40,000 people get jobs as developers.Get started