python-etl-pipeline — Production-Grade Python ETL & CI/CD Pipeline

A modular, strictly typed Extract, Transform, Load (ETL) data pipeline written in Python, engineered with explicit schema validation, a custom domain exception hierarchy, environment-driven 12-Factor configuration, multi-stage non-root Docker containerization, and a 3-stage GitHub Actions CI/CD quality gate.

---

Architecture & Pipeline Flow

Data Sources (Built-in / CSV / JSON)
        │
        ▼
┌─────────────────────────────────────────────────────────────────┐
│ 1. EXTRACT (etl/pipeline.py :: extract_data)                    │
│    • Reads source via PipelineConfig (ETL_SOURCE_FORMAT/PATH)   │
│    • Rejects empty datasets or malformed JSON/CSV payloads      │
│    • Emits ExtractionError with chained root cause              │
└───────────────────────────────┬─────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│ 2. VALIDATE & TRANSFORM (validate_data + transform_data)        │
│    • Enforces required schema columns: {"id", "name", "value"}  │
│    • Rejects null/NaN cells per column via ValidationError      │
│    • Computes derived columns (value_doubled, name_upper)       │
└───────────────────────────────┬─────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│ 3. LOAD (etl/pipeline.py :: load_data)                          │
│    • Computes aggregate metrics (record count, total_value)     │
│    • Persists atomic JSON summary artifact if ETL_OUTPUT_PATH   │
│    • Emits structured telemetry logs with timestamps            │
└─────────────────────────────────────────────────────────────────┘

---

Key Engineering Highlights

• 12-Factor Environment Configuration (etl/config.py): Immutable PipelineConfig dataclass resolved from environment variables with validation on log levels, source formats, and positive chunk sizes.
• Explicit Exception Hierarchy (etl/exceptions.py): All failure modes inherit from a common ETLError base class (ExtractionError, ValidationError, TransformationError, LoadError) so upstream orchestrators (Airflow, Prefect, AWS Step Functions) can catch and retry specific failure stages cleanly.
• Strict Static Analysis & Formatting Gate: Enforced on every push via GitHub Actions:
  • black --check (line-length = 100)
  • isort --check-only (profile = "black")
  • flake8
  • mypy static type checking
• Matrix Testing & Coverage (tests/): Verified across Python 3.10, 3.11, and 3.12 with pytest and pytest-cov, exercising happy paths, CSV/JSON extraction, schema mismatches, null injection, and I/O failure recovery.
• Hardened Multi-Stage Dockerfile (Dockerfile): Compiles wheels in a builder stage, copies artifacts into a slim runtime image, executes under an unprivileged etluser account, and runs a container HEALTHCHECK.

---

Project Structure

python-etl-pipeline/
├── etl/
│   ├── __init__.py             # Public package exports
│   ├── config.py               # Environment-driven PipelineConfig dataclass
│   ├── exceptions.py           # ETLError -> Extraction/Validation/Transformation/LoadError
│   └── pipeline.py             # extract_data, validate_data, transform_data, load_data, run_pipeline
├── tests/
│   ├── test_config.py          # Configuration defaults, overrides, and invalid-input tests
│   └── test_pipeline.py        # End-to-end extraction, validation, transformation & load tests
├── .github/workflows/
│   └── ci-cd.yml               # 3-stage Lint -> Matrix Pytest (3.10-3.12) -> Docker Build/Test
├── Dockerfile                  # Multi-stage, non-root production container with HEALTHCHECK
├── pyproject.toml              # Black, isort, mypy, and coverage configuration
├── setup.cfg                   # Flake8 configuration
├── requirements.txt            # Runtime dependencies (pandas, numpy)
├── requirements-dev.txt        # Dev & CI dependencies (pytest, pytest-cov, black, isort, flake8, mypy)
└── LICENSE                     # MIT License

---

Quick Start

1. Local Setup

git clone https://github.com/saiprasanth-git/python-etl-pipeline.git
cd python-etl-pipeline

python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements-dev.txt

2. Run the ETL Pipeline

python -m etl.pipeline

Sample structured log output:

2026-08-11 16:55:00,102 [INFO] etl.pipeline: Starting ETL pipeline
2026-08-11 16:55:00,108 [INFO] etl.pipeline: Extracted 5 records
2026-08-11 16:55:00,111 [INFO] etl.pipeline: Transformed 5 records
2026-08-11 16:55:00,112 [INFO] etl.pipeline: Loaded 5 records (total_value=1000)
2026-08-11 16:55:00,112 [INFO] etl.pipeline: Pipeline complete: {'status': 'success', 'records': 5, 'total_value': 1000}

3. Configuration Reference

All settings are optional and read from environment variables:

Variable	Default	Description
ETL_SOURCE_PATH	(none)	Path to a local .csv or .json input dataset
ETL_SOURCE_FORMAT	builtin	Input adapter: builtin, csv, or json
ETL_OUTPUT_PATH	(none)	Optional file path to write JSON execution summary
ETL_LOG_LEVEL	INFO	Standard Python logging level (DEBUG, INFO, WARNING, ERROR)
ETL_CHUNK_SIZE	1000	Batch chunk size (validated >= 1)

Example running against a custom CSV file and exporting a JSON summary:

export ETL_SOURCE_PATH=./data/customers.csv
export ETL_SOURCE_FORMAT=csv
export ETL_OUTPUT_PATH=./output/result.json
python -m etl.pipeline

---

Testing, Linting & Docker Verification

Run Unit Tests with Coverage

PYTHONPATH=. pytest tests/ --cov=etl --cov-report=term-missing -v

Run the Full Static Analysis Suite

black --check etl tests
isort --check-only etl tests
flake8 etl tests
mypy etl

Build & Verify Production Docker Container

docker build -t python-etl-pipeline:latest .
docker run --rm python-etl-pipeline:latest
docker run --rm -e PYTHONPATH=. python-etl-pipeline:latest python -m pytest tests/

---

License

Distributed under the MIT License. See LICENSE for details.