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 from Hub'Eau, France's official open water-data API. Then, you'll clean that data and publish it as a public dataset, just like the live version does.

This tutorial is based on a real pipeline that runs once a week, on a schedule, and it keeps the Paris Flood Dataset updated automatically.

You won't just copy and paste code, though. The real goal is to understand why the 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 this notebook. For most of the tutorial, the pipeline runs on simulated (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 @dataclass instead 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, schedulable main() 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 with functions and basic OOP (Object-Oriented programming) in Python. Don't worry, this guide explains every non-obvious piece as you go.

• Optional: a free Kaggle account and the Kaggle 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 a Jupyter notebook. Download the .ipynb file 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 the notebook package too:

pip install notebook

Part 1: The Design Logic

Before you touch a single line of code, we'll spend three minutes on why the pipeline is shaped this way.

This is the big-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 for Extract, 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:

The Architecture, at a Glance

Notice that "Extract" already touches two different sources: the existing dataset (what you already have) and the new data 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

Idempotence means 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 the entire history 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.

An incremental pipeline avoids this. Instead, it:

• Checks the most recent date already in the dataset

• Requests only the data from that date onward

• Merges the new records into the existing dataset

You'll find this logic built explicitly in determine_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 follows PEP 8's import-ordering convention. Most auto-formatters (isort, ruff) enforce this same grouping.

• Nothing is imported with from 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, including Google's Python Style Guide.

• The pd.set_option(...) calls exist purely for notebook 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 .py script.

Part 3: Manage Configuration with Dataclasses

Why Bother with a Config Layer at All?

Every pipeline has knobs: 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 an if level > 6000: buried three functions deep. Then changing any setting 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 @dataclass decorator is a natural fit for that.

Compare that to a configuration written as a plain dict:

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 has a single responsibility: API details, station rules, and publishing destination. Each class becomes one module-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 misconfigured APIConfig(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 at station_codes. It isn't written as station_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 does Real Python's guide to optional arguments. The fix is field(default_factory=...). It calls a fresh factory function (here, a lambda) 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_path and metadata_path are marked field(init=False), so you can't set them directly through the constructor. Instead, __post_init__ computes them from other fields (output_dir and output_filename).

Use this pattern for derived, computed values: compute them once, in one place, instead of recomputing output_dir / output_filename every time you need the path elsewhere in the codebase.

See Real Python's data classes guide or the data classes chapter of O'Reilly's Fluent Python for more on this pattern.

Tip: the auto-generated __repr__ gives you a readable printout for free. Call print(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', ...)

Part 4: The Extraction Step

The Extract phase pulls data from source systems and reads it into memory. Here, you genuinely have two sources to extract from: the existing dataset (what you already published last time) and the new (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 a FileNotFoundError.

Here's the trick: load_csv() follows the Null Object pattern. Instead of raising an error, it returns an empty DataFrame. 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=False tells pandas to read the file carefully, so it avoids mixed-type guesses. parse_dates=True tries 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 a consistent type from every code path in a function. A function that sometimes returns a DataFrame and sometimes None forces every caller to add a None check before using the result.

Return a DataFrame, 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 covers yesterday.

• 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)

Why yesterday and not today? 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 last fully completed day.

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

   • existing is a pandas DataFrame that 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

   • If last_day >= yesterday, it returns:

     • False = no update needed

     • None = no start date needed

8. If data is behind

   • It sets next_day to 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 for two possible column names: date_obs_elab, the raw API name, or the already-renamed English name record_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 becomes NaT (Not a Time) instead of raising an exception.

The return type is Tuple[bool, Optional[str]]. This tuple bundles 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: use Tuple return types (or, for more fields, a small dataclass or NamedTuple) to bundle related results together. Document clearly what each position means. If you return different types from different code paths without documentation, you'll hit a common source of confusion and bugs: "why is this None sometimes 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 a mock generator. 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 from start_date.

How it works:

• It converts start_date into a real date.

• It loops for num_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 the Single 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() owns all the 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 (with concurrent.futures or asyncio, for example), the orchestrator is the only place 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.

• If use_mock=True, it uses fake data instead of calling the real API.

  1. It calls generate_mock_api_data()

  2. Turns the result into a DataFrame

  3. Converts the date column into real pandas dates

  4. Returns that DataFrame

• If use_mock=False, it does the real API request:

  1. Creates a reusable HTTP session

  2. Starts from start_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 with date_debut_obs_elab=cursor: "give me records from this date on."

2. If the request fails outright (requests.RequestException), log it and break. break stops 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 the latest date 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 to last_page_date + 1 day and loop again for the next page.

7. As a safety net: if the page returned fewer records than max_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 classic pagination termination heuristic. You don't always need a next_page token from the API. If a full page is size=20000 and you get back only 4213 records, there's nothing left to fetch.

Three specific, deliberate choices, called out:

session = requests.Session()

A Session object 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 call requests.get() more than once against the same host in a loop, reach for a Session.

except requests.RequestException as e:

RequestException is the base class for every exception requests can raise: timeouts, connection errors, HTTP errors from raise_for_status() and more. Catching the base class here means any network hiccup gets handled the same forgiving way: log it, stop fetching this station, move on.

timeout=API_CONFIG.timeout_seconds

By default, requests calls never time out. Without an explicit timeout, a hung server can freeze your entire pipeline indefinitely. The requests advanced usage docs states that requests to external servers should have a timeout attached.

Best practice: wrap every external I/O call in try/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 a pd.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 a new DataFrame 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 use errors="coerce", so values pandas can't parse become NaT (Not a Time) or NaN (Not a Number) instead of raising an error. This is called graceful coercion.

Code-level walkthrough

convert_to_date() and convert_to_numeric() are small helper functions that make sure certain columns have the right type.

• Both start with df.copy() so they don’t change the original DataFrame.

• errors="coerce" means bad values become missing values instead of causing a crash.

For convert_to_date():

1. df.copy() makes a separate copy, so the original table stays unchanged

2. The loop goes through each column name in columns

   • if col in df.columns checks 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 become NaT (missing date) instead of raising an error

   • .dt.normalize() removes the time part and keeps only the date at midnight

For convert_to_numeric():

1. df.copy() is used here as well.

2. The loop goes through each column name in columns

   • if col in df.columns checks that the column actually exists before trying to convert it

   • The loop calls pd.to_numeric() turns text like "12.5" into numbers

   • errors="coerce" turns bad values into NaN instead 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" and 12.

Both pd.to_datetime and pd.to_numeric are official pandas functions with an errors parameter. 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 the pandas to_datetime docs and to_numeric docs for 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 between availability (the pipeline keeps running) over strictness (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 of every transform function in the pipeline, and that's not an accident.

Python DataFrames are mutable objects, passed by reference. If a function modifies df in place without copying first, the caller's original DataFrame changes too. That's a classic side effect and 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 stays guaranteed untouched.

Best practice: treat DataFrames as immutable 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 to guess which columns are dates or numbers and then fixes these types automatically.

How it works:

It starts with two empty lists:

• datetime_cols for date columns

• numeric_cols for 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 (object or string), 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 to datetime_cols

If not, it tries to read them as numbers:

• if that works, the column is added to numeric_cols

At the end, it converts all date columns with convert_to_date(). Then it converts all numeric columns with convert_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 inner try/except blocks here use errors='raise'. That's the opposite of the coercion strategy above, but it only runs against a small .head(10) sample of each column.

This is a type-sniffing step, 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 the whole column to convert_to_date() or convert_to_numeric(). Two different errors strategies, two different jobs.

Schema Translation with Bidirectional mappings

Design logic

The Hub'Eau API returns French column names and French categorical values, like code_station or "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: define one authoritative mapping at the top of the module, and derive everything else from it.

Code-level walkthrough

API_TO_EN maps 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 from API_TO_EN. A dict comprehension builds it. Here's the important design point: EN_TO_API isn't hand-maintained, it's derived. If you add, remove, or rename an entry in API_TO_EN, EN_TO_API updates automatically the next time the module runs.

There's exactly one place in the entire codebase where a schema change needs to happen.

Categorical values (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 in CATEGORICAL_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.map replaces every value it finds in the mapping dict and turns any value not found in the dict into NaN. Chain .fillna(df[col_name]) right after restores the original value 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 on Stack 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 computes flood_alert. If the water level is greater than the flood threshold set in the Config, the result is True. Otherwise, the result is False.

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_ORDER holds 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_cols acts as a safety net: any column not explicitly listed in COLUMN_ORDER still 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 pipeline idempotent: run it once or run it five times and the resulting dataset stays identical.

The fix requires defining what makes a record unique. 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 the same station reports the same day but with a different value, 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: parts ends up as a list of Series, 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. The list comprehension then 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 .apply is notoriously slow in pandas compared to vectorized string operations.

The actual deduplication happens in remove_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 returns new.

2. It enforces types in both DataFrames with auto_convert_columns() first, so dates and numbers compare correctly.

3. It creates a deduplication key for each row in both tables with create_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 from new.

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_keys is a set, not a list. Testing membership (in / .isin()) against a Python set is O(1) on average, because it uses a fast hash lookup to answer is this item already here? directly. Testing against a list is O(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 of list vs. set lookup performance in Python and this guidance on .isin() performance for filtering.

Simple rule: use a set when you care about fast membership checks. Use a list when you care about order or duplicates.

Second, new.iloc[new_keys[mask].index] indexes back into the original, non-validated new DataFrame, not new_std.

Why? auto_convert_columns() only ran to get consistent types for comparison. The caller still wants the original raw values and schema back for everything that survives deduplication.

Don't let a side-computation accidentally become 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 (with if 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 in postprocess()

All the pieces above are small and testable on their own. The last step of Transform glues 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 new logic here, only sequencing. 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 a logger. Here, this tutorial sticks with classic console printing (print(f" 1. Converting types...")).

Part 6: The Load Step

Design Logic

Load is 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:

• If use_mock is True, it uses KAGGLE_CONFIG.mock_output_dir, the directory for fake data.

• Otherwise, it uses the real output directory set in KAGGLE_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=True is a small but important detail. Without it, Path.mkdir() raises FileExistsError if the directory already exists, which it will on every run after the first.

Setting exist_ok=True makes directory creation idempotent: 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 the Data Package specification. It requires a descriptive dataset-metadata.json file alongside the CSV. This matters for the search and discoverability of the Paris 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 the KaggleConfig dataclass. Note that temporalCoverage is computed from 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 the Kaggle CLI as an external command.

What it does:

1. Gets the current time and turns it into a text label like Weekly update: 2026-07-13 14:30:00

2. Builds a Kaggle command that says:

   • publish a new dataset version

   • use files from KAGGLE_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 with subprocess.run(...)

If the upload works:

• it prints Successfully 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_kaggle calls the Kaggle CLI (Command Line Interface) through Python's subprocess module. 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:

• cmd is built as a list, never as a concatenated string. Pass a list of arguments to subprocess.run skips invoking a shell entirely. That sidesteps shell-injection risks and correctly handles arguments containing spaces or special characters. See the official subprocess security considerations.

• check=True means that if the Kaggle CLI fails, by exiting with a non-zero status, subprocess.run raises CalledProcessError instead of silently returning. Without check=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 local mock_output folder.

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 the conductor. It wires every previously-built component together in the right order, just like postprocess() 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()

The if __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 the official Python documentation on __main__:

• Run the file directly (python script.py)

  • → the special variable __name__ is set to "__main__"

  • → the condition is True

  • → main() executes.

• Import the file as a module elsewhere (import script)

  • → __name__ is set to the module's name instead

  • → the condition is False

  • → main() does not run automatically.

Best practice: always guard your entry point this way. It makes the module safely importable for 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 the use_mock=False setting above. It threads through APIConfig.use_mock, and the calls to run_etl_pipeline(use_mock=False) and main().

2. Make sure KAGGLE_CONFIG.dataset_slug and input_csv point at your own Kaggle dataset copy before you use it. You can only publish new versions of a dataset you own. Fork both the notebook and the dataset, then update the slug in the KaggleConfig section.

3. Install and authenticate the Kaggle CLI if you plan to call publish_to_kaggle() outside of a Google Colab or Kaggle notebook.

Everything else needs zero 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 runs after the pipeline finishes. The output is a human-readable report covering shape, data types, null counts, summary statistics on water_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 a complete ETL 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 my GitHub or Kaggle profile.

Thanks for reading!

References

• Hub'Eau API: France's official open water-data platform

• The updater .py script runs live once a week and updates the original dataset.

• The source dataset generator is available on GitHub.

• The complete Jupyter notebook to follow this tutorial and code all along (backup available).

But managing disk space can be a quite a challenge, as WSL uses virtual hard disks that do not automatically free up unused space.

This tutorial will guide you through the process of manually compacting your WSL virtual hard disks. We’ll automate this task using a PowerShell script, ensuring that your WSL environment remains efficient and clutter-free.

Reclaim Your Space

WSL uses a virtualization platform to install Linux distributions on your Windows system. Each distribution you add gets its own Virtual Hard Disk (VHD), which uses the ext4 file system (common in Linux). It’s saved on your Windows drive as an ext4.vhdx file.

Key issues here:

• Inefficient storage: by default, VHD files do not reclaim unused space. This means that when you delete a file in WSL, the associated disk space isn’t immediately freed up.

• Disk space consumption: due to that inefficient storage, the VHD files can grow large thanks to that accumulated data, especially if you’re a WSL heavy user.

• Need for maintenance: you may not know that you need to compact your VHD files in order to reclaim disk space.

If you notice that your free disk space is shrinking even after deleting files and apps, WSL might be the reason. This tutorial will help you keep your WSL and Windows environment running smoothly.

Table of Contents

• Part 1: How to Manually Compact Your Virtual Hard Disk

  • Prerequisites

  • Step 1: Verify your WSL version and status

  • Step 2: List all installed distributions verbosely

  • Step 3: Locate your linux Virtual Hard Drive (VHDX) path

  • Step 4: Shut down all WSL instances

  • Step 5: Compact the Linux virtual hard drive using DiskPart

  • Step 6: Restart WSL and verify

• Part 2: How to Make Your Life Easier with Automation

  • Prerequisites

  • Step 1: Find out installed WSL2 distributions

  • Step 2: Select a distro to compact

  • Step 3: Locate the ext4.vhdx File

  • Step 4: The confirmation prompt

  • Step 5: Shut Down WSL and compact

  • Step 6: Run a DiskPart script

Part 1: How to Manually Compact Your Virtual Hard Disk

Let's start by going through the process manually. This section will guide you through checking your WSL version and associated Linux distributions, finding VHD files, shutting down WSL, and compacting the virtual disk.

Prerequisites

• Windows 10 (20H1/2004+) or Windows 11 with WSL2 installed

• The PowerShell or Command Prompt running as Administrator (from the Windows menu, right click the icon and choose run as Administrator).

Step 1: Verify your WSL version and status

First, make sure you’re running on WSL version 2 (commonly referred as WSL2). The first version is outdated and WSL2 provides significant improvements. Open PowerShell (as Admin) or Command Prompt (as Admin) and run:

wsl -v

wsl --status

These commands display the WSL client version and whether your default distro is using WSL 2. Here’s the output of the wsl -v command:

And here’s the output of the wsl --status command.

Step 2: List all installed distributions verbosely

To see a detailed list of your WSL distributions (including which version each uses), run:

wsl.exe --list --verbose

Above you can see the output of the WSL --list --verbose command.

Look for your distro name (for example, “Ubuntu”) and note its WSL version. If it shows “Version 2”, you can proceed with compaction.

Step 3: Locate your linux Virtual Hard Drive (VHDX) path

Each WSL distro’s files live in a VHDX file on your Windows drive. To find the path for any Linux distribution, use this PowerShell snippet:

(Get-ChildItem `

-Path HKCU:\Software\Microsoft\Windows\CurrentVersion\Lxss `

| Where-Object { $_.GetValue("DistributionName") -eq 'YOUR_DISTRO_NAME' }

).GetValue("BasePath") + "\ext4.vhdx"

Where you replace YOUR_DISTRO_NAME with yours (Ubuntu, Debian, Kali-linux..). Here’s the output of the command shown above in PowerShell (the filepath has been anonymized):

This command reads the registry key for your linux distribution, then appends “\ext4.vhdx” to build the full file path.

Make sure you copy the whole line. We will need it in later stages.

Step 4: Shut down all WSL instances

Before you can compact any virtual drive, make sure WSL is completely shut down. In PowerShell or Command Prompt (still as Administrator), run:

wsl.exe --shutdown

Step 5: Compact the Linux virtual hard drive using DiskPart

You successfully gathered all the needed information (about your system, the available distros, and their VHDX filepath) to proceed with the main task. In this step, you actually proceed with the compaction.

1. Launch DiskPart in the same elevated (admin) shell:

diskpart

DiskPart will open in a new window. It's a Windows command-line tool for managing disk partitions. Be cautious when using it, as incorrect actions can cause serious data loss.

1. In the DiskPart prompt, select the VHDX file you found earlier. Replace the path as displayed below with your actual path (the line you copied before):

select vdisk file="C:\Users\username\AppData\path\to\ext4.vhdx"

The above is the output of the select vdisk command (some data has been anonymized).

1. Attach the virtual drive in read-only mode:

Compaction only needs to scan the empty blocks in the file, not write to the Linux filesystem inside. Read-only mode guarantees that DiskPart only inspect the blocks for zero‐trimming without any chance of damaging or altering your Linux filesystem.

attach vdisk readonly

You can see in the screenshot above that the virtual hard drive has been successfully attached.

1. Compact the disk:

This frees up the disk space by shrinking the physical size of the .vhdx file to match the actual used data inside.

compact vdisk

This operation might take a while. When you see the “DiskPart successfully compacted the virtual disk file” message, proceed with the next step. In the image below, the virtual hard drive has been successfully compacted.

1. Detach the virtual drive:

detach vdisk

There you go – the virtual hard drive has been successfully detached.

This command releases any locks on the virtual drive and effectively dismounts it. If you don't use this command, the file remains "in use," preventing WSL (or you) from accessing it until you reboot or manually force it closed.

6. Exit DiskPart:

exit

Step 6: Restart WSL and verify

Back in PowerShell or Command Prompt, you can relaunch your distro:

wsl -d YOUR_DISTRO_NAME

You can even try the Unix df -h command in your WSL prompt to check your new available disk spaces.

Congrats, you just achieved a maintenance task that can free up lots of gigabytes of storage over time. Now, it’s time to automate.

Part 2: How to Make Your Life Easier with Automation

Since it's often hard to remember exactly where your WSL distro is located and you probably won't use it very often, this PowerShell script will automate the entire process we covered in part 1. Here's a preview of the steps you'll follow:

• Detect installed WSL distributions.

• Select one (and handle the cases there are more than one).

• Locate the corresponding ext4.vhdx file.

• Shut down WSL and use DiskPart to compact the virtual disk.

Prerequisites

• Windows 10 (20H1/2004+) or Windows 11 with WSL 2 enabled.

• PowerShell or Command Prompt (as Administrator).

You’ll also need a code editor. The Windows notepad is enough for completing this task. You can also use an IDE (Integrated Development Environment) like VS Code or an ISE (Integrated Scripting Environment) like PowerShell ISE (included with Windows).

To test the script, download it from GitHub. Open an elevated PowerShell or Command Prompt and navigate to the script’s folder. With just the command below, you will be able to run it and free up some disk space:

powershell.exe -NoProfile -ExecutionPolicy Bypass -File .\wsl_compactor.ps1

Step 1: Find out installed WSL2 distributions

One of the main challenges is to find the Linux distributions available on the host system. Let’s check the first block and see what’s it about:

$lxssKey = 'HKCU:\Software\Microsoft\Windows\CurrentVersion\Lxss'
\(distros = Get-ChildItem \)lxssKey | ForEach-Object {
    \(p = Get-ItemProperty \)_.PSPath
    [PSCustomObject]@{
        Name     = $p.DistributionName
        BasePath = $p.BasePath
    }
}

WSL records each distribution under this windows registry key:

HKCU:\Software\Microsoft\Windows\CurrentVersion\Lxss

Each subkey has two important values:

• DistributionName (for example, Ubuntu)

• BasePath (This is where the distribution files are stored. It’s the directory that contains the ext4.vhdx file.)

The script uses Get-ChildItem and Get-ItemProperty to enumerate these subkeys and build a list of available Linux distributions.

if ($distros.Count -eq 0) {
Throw-And-Exit "No WSL distros found in the registry."
}

If no distributions are found, the script terminates and prints this error message on the terminal: "No WSL distros found in the registry.”

Step 2: Select a distro to compact

Here, the process has two steps:

• If multiple distros are found, it displays all the distros with a numbered menu and prompts you to choose one:

if ($distros.Count -gt 1) {
    Write-Host "Multiple distros detected. Please choose one:`n"

    for (\(i = 0; \)i -lt \(distros.Count; \)i++) {
        Write-Host "[\((\)i+1)] \((\)distros[$i].Name)"
    }
    \(selected = \)distros[[int]$choice - 1]
}

The computed menu will look like this:

Multiple distros detected. Please choose one:

[1] Ubuntu 20.04

[2] Debian

[3] Alpine

• If only one distribution is found on the host system, the script selects it automatically:

else {
\(selected = \)distros[0]
}

When setting up a distribution, whether chosen manually by the user or selected automatically, the important information is the path to each distribution's virtual hard disk. This path is saved in two main variables: 'distro' (which identifies the specific distribution) and 'basePath' (which shows where its virtual disk is located).

\(distro = \)selected.Name
\(basePath = \)selected.BasePath

Write-Host "`nSelected distro: $distro" -ForegroundColor DarkYellow
Write-Host "BasePath: $basePath"

The lines above display an output that looks like this:

Selected distro: Ubuntu (or any other distro)
BasePath: C:\Users\<User_name>\AppData\Local\Packages\…

Like for all other steps, it’s important to consider the case of something going wrong, by throwing an error and exiting the program:

if (-not (Test-Path $basePath)) {
Throw-And-Exit "BasePath '$basePath' does not exist on disk."
}

Step 3: Locate the ext4.vhdx File

In the first step, we collected the information we need about the available distributions and where they are stored on the Windows system. By choosing an entry (either manually or automatically), we can find the correct file. Sometimes, the ext4 file is located between the base path and a LocalState folder. This script manages both situations. It builds the usual locations where the file can be found.

They look like this:

• $BasePath\ext4.vhdx

• $BasePath\LocalState\ext4.vhdx

This can translate into something like this on your system (option 1):

C:\Users\Alice\AppData\Local\Packages\CanonicalGroupLimited.Ubuntu20.04onWindows_79rhkp1fndgsc\ext4.vhdx

or like this (option 2):

C:\Users\Alice\AppData\Local\Packages\CanonicalGroupLimited.Ubuntu20.04onWindows_79rhkp1fndgsc\LocalState\ext4.vhdx

(You might find out that your WSL2 distro is located in some other directory than “Packages” – but don’t worry, your BasePath will match the correct folders).

The idea is to build the two possible path options:

$possible = @(
Join-Path $basePath 'ext4.vhdx'
Join-Path $basePath 'LocalState\ext4.vhdx'
)

And pick the first one that actually contains the file:

\(vhdx = \)possible | Where-Object { Test-Path $_ } | Select-Object -First 1

Again, we throw an error message if no suitable file is found:

if (-not $vhdx) {
Throw-And-Exit "No ext4.vhdx found under '$basePath'."
}

Step 4: The confirmation prompt

Disk management tools require caution and you need to understand the potential consequences of your actions. A confirmation prompt is always a good safeguard to prevent accidental data loss or unwanted system changes.

Before proceeding, the script shows you:

• a Distro name

• its BasePath

• the VHDX file path

Write-Host "`nAbout to compact this WSL distro:" -ForegroundColor Magenta
Write-Host " Distro : $distro"
Write-Host " BasePath : $basePath"
Write-Host " VHDX file: $vhdx`n"

It then prompts “Are you sure you want to proceed? (Y/N)”:

Write-Host "Are you sure you want to proceed? (Y/N) " -ForegroundColor DarkCyan -NoNewline

# Then read the response
$answer = Read-Host

You’re then prompted to Type Y (case-insensitive) to continue or anything else to cancel.

if ($answer.ToUpper() -ne 'Y') {
    Write-Warning "Operation canceled"
    exit
}

For the two steps above, I had to use a trick to print the question in color but a simple option (without colors) could be:

if ((Read-Host 'Are you sure you want to proceed? (Y/N)').ToUpper() -ne 'Y') { 
    Write-Warning 'Operation canceled'
    exit 
}

Step 5: Shut down WSL and compact

Before proceeding to the DiskPart utility, it’s important to stop all running WSL instances. Pass the shutdown command directly in PowerShell.

Write-Host "Shutting down WSL…" -ForegroundColor Cyan
wsl.exe –shutdown

A common mistake is to forget to launch PowerShell or the Command Prompt with Administrator rights. You can prevent this case with a message:

if ($LASTEXITCODE -ne 0) {
     Throw-And-Exit "Failed to shut down WSL (exit code $LASTEXITCODE). Are you running as Administrator?"
}

Step 6: Run a DiskPart script

Building the script:

The process is the same as in the manual part, but this time, we ‘inject’ the ready-to-go DiskPart commands into the script.

$dpScript = @"
select vdisk file="$vhdx"
attach vdisk readonly
compact vdisk
detach vdisk
exit
"@

Before launching, there are two steps you need to take:

1. The PowerShell script writes the lines above to a temporary file:

$tempFile = [IO.Path]::GetTempFileName()
Set-Content -LiteralPath \(tempFile -Value \)dpScript -Encoding ASCII

This is the equivalent to the commands passed in the manual part:

select vdisk file="C:\…\ext4.vhdx" # full path to the vdisk file

attach vdisk readonly

compact vdisk

detach vdisk

exit

1. Compacting can take a while, especially if you’ve never de-cluttered your virtual drive before. It’s wise to show a warning before proceeding:

Write-Host "Running DiskPart to compact the VHDX. Be patient, this might take a while..." -ForegroundColor Cyan

Invoke DiskPart:

# Run DiskPart with the script saved to the temporary file and process each output line as it arrives
diskpart /s $tempFile | ForEach-Object {
    # Grab any "NN percent" type message from the line
    if ($_ -match '(\d+)\s+percent') {
        # Only print when the percentage actually changes
        Write-Host "\((\)Matches[1])% completed"
    }
    else {
        # Just echo all over line-types, verbatim
        Write-Host $_
    }
}

Several points to note here:

• It runs diskpart /s $tempFile: DiskPart reads and executes commands from the temporary file into the PowerShell loop for on-the-fly processing.

• For a better user-experience: the snippet below does the trick of filtering out repeated status values by comparing \(pct with the sentinel \)lastPct, and only writing new lines when they differ.

How?

Before entering the loop, we initialize:

$lastPct with -1
$lastPct = -1 # We initiate a sentinel value

We are having a guaranteed “first” value that no real percent (0–100) will equal. That way, as soon as you see the first 0 percent, 10 percent, or whatever, it differs from -1.

Then:

if ($_ -match '(\d+)\s+percent') {
    # Print only when the percentage changes
    Write-Host "\((\)Matches[1])% completed"
}

This guarantees that on the very first percentage update (say “0 percent” or “10 percent”), \(pct –ne \)lastPct will be true, so it emits the first line. Afterwards, $lastPct holds the last real percentage, and it only prints again when a new, different progress percentage comes in.

The output looks more clean:

10% completed
20% completed
…

Otherwise, it’ll flood the screen with dozens of identical “20 percent completed” (for example) updates.

Of course we handle the case of other values (non percent lines) normally:

else {
    # non-percent lines: print verbatim
    if ($_ -match '\S') {
        Write-Host $_
    }
}

Once the process is done, don’t forget to clean up the tempfile.

Remove-Item $tempFile -ErrorAction SilentlyContinue

By the end of the process, you should see something like this

Leaving DiskPart...

Okay, that’s it for the scripting! If you collected all the snippets so far, just save them with a .ps1 file extension, or download the full example from this GitHub repository.

Usage:

You now have a complete understanding of what's happening. Ready to get started? In Windows, open the Command Prompt or PowerShell as an administrator.

• Navigate to the directory containing your .ps1 script.

• Execute the script with the following command, replacing <File_name_here> with your actual file name:

powershell.exe -NoProfile -ExecutionPolicy Bypass -File .\<File_name_here>.ps1

The -NoProfile -ExecutionPolicy Bypass parameters launch PowerShell in a clean, unrestricted environment that ignores user-specific settings and allows script execution without security restrictions. Don’t worry, in this case, it’s okay to do that.

Wait…wait…wait...

Well bravo! You’ve just reclaimed all that unused space inside your WSL2 (almost) hands free!

Now, you can take it a step further by modifying this script to run completely automatically, without the confirmation prompt (step 4). You can also schedule it as a regular maintenance task using a task scheduler:

schtasks /create /tn "Schedule_name" /tr "powershell.exe -ExecutionPolicy Bypass -File C:\path\to\script.ps1" /sc monthly /d 15 /st 09:00

This is an example for a monthly execution where /d 15 means 15th of each month and /st 09:00 is a start time set at 9 am.

That's it! Remember, regular maintenance, whether you do it manually or automatically, is essential to prevent unnecessary disk space usage and ensure a smooth experience with WSL.

Thanks for reading!

You can find a list of my current projects on GitHub.

If you haven’t heard about what’s going on:

GitHub is struggling to contain an ongoing attack that’s flooding the site with with millions of code repositories. These repositories contain obfuscated malware that steals passwords and cryptocurrency from developer devices. … The result is millions of forks with names identical to the original one.

– Dan Goodin, Ars technica

Because search engines and GitHub’s own search rankings favor recent activity, these cloned repositories often float to the top – then they lure unsuspecting developers into pulling code that may contain malware.

One of my repositories has been targeted by such an attack, prompting me to monitor it closely. This guide offers tips to spot malicious repository clones before they catch you off guard.

Table of Contents

1. What is a Repository Confusion Attack?

   • Supply Chain Attacks

2. 🛡️ Basic Mitigation Strategies

   • Verify the contributors profiles

   • Search for clone repositories

   • Examine the commit pattern

   • Examine the commit history

   • Examine the commit contents

   • Compare the concerned files

   • Some information about the malware

3. Action Time

4. Conclusion

5. More Resources

What is a Repository Confusion Attack?

A repository confusion attack involves:

• Cloning legitimate repositories.

• Injecting malicious code into the clone.

• Uploading the clone.

• Spreading through various unaware actors.

Supply Chain Attacks

If you search for repository confusion on the internet, you'll find out it's a type of supply chain attack.

A supply chain attack is an indirect threat where hackers try infiltrating a system by targeting a trusted third-party or software component, rather than attacking the primary target directly.

It's not the first time this has happened. Before GitHub was targeted, PyPI was attacked in 2023 with fake packages posing as legitimate. These packages lured negligent pip users into downloading malicious payloads (containing in most cases infostealer malware).

🛡️ Basic Mitigation Strategies

Before using any repository, make sure you follow these steps and take these precautions.

Verify the contributors profiles

That's a first check: if you see a rather empty GitHub profile – one without reputation that contains just one repository but with a lot of daily commits to it – well, that's a bit suspicious.

In the fake repository, the original author will be listed as a contributor, too. Check that profile. You should be able to find the legitimate repository and do some comparisons.

In the above screenshot you can see solotech143, my evil doppelgänger (he’s been taken down since).

Search for clone repositories

You can do a GitHub search by repository name and sort the results by most recent first. Malicious repositories tend to appear at the top of the search results because they are updated more frequently. The original repository might be hidden deeper in the search results.

It’s like clone wars.

This is where it’s dangerous: users generally click on the first few search results, and in that type of attack, you’re almost guaranteed to see the attacker’s fake repository at the top of the results. The attacker achieves that by giving the fake repository regular fresh commits (and sometimes even a few stars!).

In my case, the original repository is a submission for the HackaViz 2025 competition. Hackathons offer a good attack surface because, beyond the fact they draw niche communities, they are also time sensitive.

Now, let’s move forward a year and imagine Hackaviz 2026 is starting soon. The attacker has easily outranked the untouched original submission. Which repository is most likely to be visited when future competitors – unaware of the scam – will look for the previous submissions?

Examine the commit pattern

Here’s when things take a weird turn. Malicious clones are run by automated agents, so the commit history fits a pattern that is rather unusual for a human. Of course, you can automate for many legitimate reasons but… this will always follow a clear goal and there will always be a human-touch at some point. In this case, commits are not adding up.

Let's see how that looks in the screenshots below:

Regular like a clock...

... and hyperactive!

Examine the commit history

You can’t! And that's the weird part. You're just able to see the last and the initial commit. So why is it hiding all of them? Do you like it when someone hide things from you?

For July 10th, we should be able to see 11 commits, where are the ten others?

Well, you can only check the first and last commit. That is not a lot for a repository that has more than 2000 commits registered.

Examine the commit contents

Well, since I can always check the last commit, I checked some of them. They share the same pattern: the bot is constantly looping over the README file doing the same modifications. As you can see in the screenshot below, it’s updating the file with links to an infected release.

Above you can see an AI agent stuck in the Readme loop of change.

Human edits are more varied. In a human-driven project, you will see a large mix of commits: feature commits, exploratory experiments, bug fixes, styling tweaks, and sometimes reverts. A bot clone will often just overwrite files, bump versions, or re-inject the same malicious payload repeatedly with no real contribution to the codebase.

Compare the concerned files

This is where common sense comes handy. So, you have two README's:

1. The first consists of AI-generated content that is cluttered with emojis and low-value information. It is designed solely to entice you into clicking the download link of the release.

2. The other follows best practices for creating a good README file. It is accurate and well-structured and functions as a valuable helper and explainer to the code. It also goes deep into the most important aspects of the project. This is usually a good sign that a repository is organic and genuine.

Some information about the malware

What do we have so far? Well, a suspicious link in a phishy, AI-generated README file that is consistent with a very suspicious pattern in the commit history.

Now, let’s have a closer look at that dubious release and let’s see what an online antivirus scanner might reveal about it.

The malware is packed only in the miniature-fortnight-v1.7.6.zip release.

Above you can see the result of a scan with an online scanner.

The .zip file contains only four files:

• config.txt

• launch.bat

• lua51.dll

• luajit.exe

These files are totally unrelated to the source project (a Python data science project with Jupyter notebooks combined to a React app using three.js).

I will not go into the detail in this article. But for the curious ones, it's an infostealer malware (a malware that will exfiltrate your credentials and other precious information about your configuration) similar to the one described in detail here.

Action Time

If you discover a potentially malicious repository, here are some steps you can take:

1. Document some evidence.

2. Notify the original repository maintainers.

3. Report the malicious clone to GitHub.

Reporting a repository or a profile on GitHub is easy and fast. Go to the user’s profile page, click “Block or report” in the left sidebar and choose “Report abuse” in the pop-up. You will have to complete a short contact form with some details about the behavior before submitting. If needed, you can find more information on GitHub.

Conclusion

This is a description of just one attack, from the perspective of someone who found out that one of his repository had been targeted. There are likely cases of more sophisticated attacks. But the clone repository flood we can see on GitHuB is definitely massive low quality automation. Quantity over quality.
To be honest, I'm quite surprised algorithms crafted at GitHub didn't manage to spot this one.

This also raises questions related to AI.

• What happens when LLMs are trained on malicious content? That’s a more general question about AI poisoning.

• A human might easily spot the patterns and the low quality content for now. But..

  • Imagine you are using coding agents, many of them. Will the agents pick-up the malicious clone instead of the original one? How to distinguish the repositories from an automaton's perspective?

  • The attackers will refine their tactics, making the clones more human-like and therefore luring us more easily into their traps.

• This is really a situation that makes me wonder about the early days of Google. Back then, the company had to fight huge amounts of spam due to keyword stuffing and manipulative SEO tactics. Will big tech companies have to go through a Florida update moment to face the rise of AI generated spam ?

More Resources

• A detailed description of the attack

• Complete safety recommendations

Stay Informed, Stay Secure!

A cheat-sheet is also available on my GitHub. Feel free to contribute to it!