Python ETL: Build Pipelines You Can Maintain

Python is a popular choice for moving data between business systems: pulling records from an API, cleaning a spreadsheet export, combining sources and loading the result into a database or warehouse. A first script often works within an afternoon. The trouble comes months later, when the script has grown into hundreds of lines, fails silently on bad data, creates duplicates when re-run and depends on the one person who wrote it. Python ETL jobs that last are built differently from the start: separate extract, transform and load steps, repeatable loads, clear logging, sensible retries, tests and proper scheduling. This guide explains how to structure Python ETL so it stays maintainable as data and requirements grow.

What a Python ETL job does

ETL stands for extract, transform and load:

• Extract: read data from sources such as APIs, databases, files or SaaS exports
• Transform: clean, validate, reshape, combine and enrich the data
• Load: write the result to a target, such as a database, warehouse or another application

Python suits this work because of its readable syntax and wide range of libraries for HTTP requests, databases, file formats and data manipulation.

Keep the three steps separate

The most useful structural habit is to keep extract, transform and load in separate functions or modules. This makes each step easier to test, lets you rerun one step without the others and helps you see where a failure happened. Transformation functions, in particular, should take data in and return data out without reading from or writing to external systems, so they can be tested with simple inputs.

Worked example: a small, structured ETL job

The simplified, untested Python sketch below reads orders from a CSV file, cleans them and loads them into a database table using an upsert so the job can be safely re-run. File names, columns and the table are placeholders.

# Simplified, untested sketch
import csv
import logging
import sqlite3

log = logging.getLogger("orders_etl")


def extract(path):
    with open(path, newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def transform(rows):
    clean = []
    for row in rows:
        email = (row.get("email") or "").strip().lower()
        if not email:
            log.warning("Skipping order without email: %s", row.get("order_id"))
            continue
        clean.append((row["order_id"], email, float(row["total"])))
    return clean


def load(rows, db_path):
    with sqlite3.connect(db_path) as conn:
        conn.executemany(
            "INSERT INTO orders (order_id, email, total) VALUES (?, ?, ?) "
            "ON CONFLICT(order_id) DO UPDATE SET email = excluded.email, total = excluded.total",
            rows,
        )


def run(path, db_path):
    rows = transform(extract(path))
    load(rows, db_path)
    log.info("Loaded %d orders", len(rows))

In a real job, the transform step would validate more fields, invalid rows would be written to a review file and configuration would come from settings rather than function arguments.

Make loads repeatable

Jobs fail and get re-run. A load that simply inserts rows will create duplicates on every retry. Make loads idempotent:

• Use upserts keyed on a stable identifier from the source
• Or load into a staging table and swap or merge in one step
• Record which batches or files have been processed
• Avoid partial loads that leave the target half-updated

If running the same job twice produces the same result as running it once, recovery becomes simple.

Validate data before loading

Bad data loaded into a target spreads into reports and other systems. Check data during the transform step:

• Required fields are present
• Values have the expected types and formats
• Identifiers are unique where they should be
• Totals or counts are within expected ranges

Decide what to do with failures: skip and log, send to a review file or stop the job. Our guide to pandas data cleaning covers common cleaning tasks such as standardising formats and handling missing values.

Log what matters

Use Python's logging module rather than print statements. Log the start and end of each run, record counts at each step, skipped or invalid records and any errors with enough context to investigate. Send logs somewhere the team can search them, not only to the console of the machine that ran the job.

Retry carefully

External APIs and databases fail temporarily. Retry transient errors, such as timeouts or rate limit responses, with increasing delays between attempts and a maximum number of tries. Do not retry errors that will never succeed, such as invalid credentials or malformed requests; report them clearly instead.

Extract only what you need

Pulling every record from a source on every run is simple but slow, and it puts unnecessary load on source systems and APIs. Where sources support it, extract only records created or changed since the last successful run, using a reliable timestamp or sequence. Store the position reached after each run, and only move it forward once the load succeeds. Schedule occasional full reloads to catch anything incremental runs might miss, such as deleted records.

Expect source changes

Sources change without warning: a column is renamed, an API adds a field, a date format changes. Check incoming data against the structure you expect and fail clearly when it differs, rather than loading misaligned data. Keep mappings between source fields and target columns in one place, so updates are quick when a source changes.

Handle credentials securely

Never hard-code passwords, API keys or tokens in scripts. Read them from environment variables or a secrets manager, and keep them out of version control and logs.

Test the important parts

Write tests for transformation functions using small, representative inputs, including edge cases such as empty fields, unusual formats and duplicates. Test the load step against a test database. Run tests automatically whenever the code changes, before anything is deployed.

Schedule and orchestrate

Simple jobs can run on a scheduler such as cron. As the number of jobs and dependencies grows, an orchestrator helps. Apache Airflow, for example, organises tasks into DAGs with dependencies between tasks, and its documentation shows how default arguments, such as retries, can be applied across the tasks in a DAG (Apache Airflow: DAGs). Whatever you use, make sure failures alert someone and that runs can be retried.

Manage dependencies and environments

Pin library versions so a new release does not change behaviour unexpectedly, and update them deliberately with tests. Use virtual environments or containers so the job runs the same way on every machine. Document how to set up and run the job locally.

Monitor results, not just runs

A job can finish successfully and still load wrong or incomplete data. Monitor row counts, freshness of the latest data and key totals in the target. Alert when they look unusual. Our guide to ETL pipeline development covers designing larger pipelines that combine many sources.

Document the job

Record what the job does, its sources and targets, schedule, owner, how to rerun it and what to do when it fails. Keep this next to the code, and update it whenever the job changes, so whoever is on call can act without reading every line.

Frequently Asked Questions

What is Python ETL?

It is the use of Python code to extract data from sources, transform it by cleaning and reshaping, and load it into a target such as a database or warehouse.

How do you make a Python ETL job safe to re-run?

Make loads idempotent with upserts or staging-and-merge patterns, and record which batches have been processed, so retries do not create duplicates.

Should ETL jobs use print statements for logging?

No. Use the logging module, record counts and errors with context and send logs somewhere the team can search them.

When does a Python ETL job need an orchestrator?

When there are many jobs with dependencies, retries and schedules to manage; tools such as Apache Airflow organise tasks into DAGs.

Build data jobs that keep working

We design, build and maintain Python ETL jobs with testing, logging and monitoring. See our CRM and API integration services, or talk to us about Python ETL.

About the Author

Manav Sharma is the founder of Insyrge and a CRM solutions architect. He works across Zoho, Salesforce, HubSpot, Odoo and more, designing the CRM setups, integrations and automations that keep sales and operations teams working from the same data.