How to handle errors and retries in a Python ETL pipeline

An ETL pipeline that runs beautifully on your laptop breaks in production for the most ordinary reasons: the API answers 429, the database drops the connection halfway, a file arrives with one corrupted row. Handling errors and retries in an ETL pipeline is what separates a fragile script from a process you can trust. The goal here is clear: tell a temporary failure apart from a real error, retry only when it makes sense, and never lose data silently.

Prerequisites

• Python 3.9 or later installed.
• The requests and sqlalchemy libraries (pip install requests sqlalchemy).
• A simple ETL pipeline already working (extract from an API, transform, load into a database).
• A basic understanding of try/except in Python.

Step 1: Separate transient errors from permanent ones

Not every error deserves another attempt. A timeout or an HTTP 503 is transient: ten seconds from now the request will probably work. An HTTP 401 (wrong credentials) or a KeyError in a transformation, on the other hand, is permanent — retrying a thousand times returns the same error and only delays the pipeline.

Start by writing that rule as code, so you stop guessing:

import requests

HTTP_RETRY = {408, 429, 500, 502, 503, 504}

def e_transitorio(erro):
    if isinstance(erro, requests.exceptions.HTTPError):
        resposta = erro.response
        return resposta is not None and resposta.status_code in HTTP_RETRY
    return isinstance(erro, (requests.exceptions.ConnectionError,
                             requests.exceptions.Timeout))

Step 2: Build a retry function with exponential backoff

Retrying immediately, five times in a row, is the best way to annoy an API that is already overloaded. The good practice is exponential backoff: wait 1s, 2s, 4s, 8s... and add a little randomness (jitter) so that several processes do not all retry on the same second.

import logging
import random
import time

log = logging.getLogger("etl")

def com_retry(funcao, tentativas=5, base=1.0, maximo=30.0):
    for tentativa in range(1, tentativas + 1):
        try:
            return funcao()
        except Exception as erro:
            if tentativa == tentativas or not e_transitorio(erro):
                raise
            espera = min(maximo, base * 2 ** (tentativa - 1))
            espera += random.uniform(0, espera * 0.1)
            log.warning("Tentativa %s falhou (%s). Nova tentativa em %.1fs",
                        tentativa, erro, espera)
            time.sleep(espera)

Note two important details: if the error is not transient, the raise happens on the very first attempt; and on the last attempt the error is raised too, instead of being swallowed.

Step 3: Apply the retry to extraction

Extraction is where the pipeline touches the outside world, so that is where retries pay off the most. Wrap the HTTP call in a function with no arguments and hand it to com_retry:

URL = "https://api.exemplo.com/v1/encomendas"

def extrair_pagina(pagina):
    def pedido():
        resposta = requests.get(URL, params={"page": pagina}, timeout=30)
        resposta.raise_for_status()
        return resposta.json()["data"]
    return com_retry(pedido)

Always set a timeout. Without it, a hanging request blocks the pipeline for hours and no retry ever happens. If the API returns a Retry-After header, respect that value instead of your calculated backoff.

Step 4: Quarantine the bad rows

In the transformation the problem is a different one: a single row whose valor field says "N/A" must not bring down a batch of 200,000 records. Instead of blowing up, split the valid from the invalid rows and store the rejects in a dead-letter queue (a file or a quarantine table) together with the reason.

import json

def transformar(linhas):
    validas, quarentena = [], []
    for linha in linhas:
        try:
            validas.append({
                "id": int(linha["id"]),
                "email": linha["email"].strip().lower(),
                "valor": float(linha["valor"]),
            })
        except (KeyError, TypeError, ValueError, AttributeError) as erro:
            quarentena.append({"linha": linha, "erro": str(erro)})
    return validas, quarentena

def guardar_quarentena(registos, caminho="quarentena.jsonl"):
    with open(caminho, "a", encoding="utf-8") as f:
        for registo in registos:
            f.write(json.dumps(registo, ensure_ascii=False) + "\n")

A healthy rule: if the quarantine goes beyond a threshold you define (say 5% of the batch), the pipeline should fail. Many bad rows all of a sudden is not an accident, it is a data contract that changed.

Step 5: Fail in a controlled and visible way

The load should be transactional: either everything lands, or nothing does. And when the pipeline really does fail, it must fail loudly — with a non-zero exit code, so the orchestrator (Airflow, Azure Data Factory, cron) knows something went wrong.

import sys

def executar():
    linhas = extrair_pagina(1)
    validas, quarentena = transformar(linhas)

    if quarentena:
        guardar_quarentena(quarentena)
    if len(quarentena) > 0.05 * max(len(linhas), 1):
        raise ValueError(f"Demasiadas linhas invalidas: {len(quarentena)}")

    with motor.begin() as ligacao:   # commit at the end, rollback on error
        carregar(ligacao, validas)

    log.info("ETL concluido: %s carregadas, %s em quarentena",
             len(validas), len(quarentena))

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    try:
        executar()
    except Exception:
        log.exception("ETL falhou")
        sys.exit(1)

Verify the result

Do not wait for production to find out whether your error handling works. Test it on purpose:

• Point URL at an endpoint that returns 503 and confirm in the logs that the waits grow (1s, 2s, 4s...) and that the error is raised after the last attempt.
• Point it at an endpoint that returns 401 and confirm there are no retries — it fails on the first try, as it should.
• Inject a row whose valor field is "N/A" and check that it shows up in quarentena.jsonl while the remaining rows were loaded.
• Run echo $? after a failure: it should return 1.

Conclusion

With these five pieces — classify the error, retry with backoff, quarantine bad rows, load inside a transaction and fail visibly — your ETL pipeline no longer needs babysitting at three in the morning. The natural next step is making the pipeline idempotent, so a re-run after a failure does not duplicate data. One final tip: once you are comfortable with the hand-written com_retry, try the tenacity library, which does the same thing with a decorator. Here is the question to take away: in your current pipeline, how many of last week's errors were actually transient?