py-etl-pipeline

Minimal Python ETL: CSV → PostgreSQL with bulk insert, native upsert, audit logging, retries, optional scheduling, Alembic migrations, Docker Compose, and an optional GraphQL health surface.

---

py-etl-pipeline is a small, production-minded ETL service: extract sales rows from CSV (pandas + normalised headers), transform with validation and coercion, then load into PostgreSQL using either bulk insert or ON CONFLICT DO UPDATE upsert, with row-level audit logging, tenacity retries for transient DB errors, optional scheduled runs, Alembic migrations, and Docker Compose for Postgres + the worker. An optional Strawberry GraphQL ASGI app exposes health, version, and non-secret runtime hints. Canonical repository: github.com/esousa97/py-etl-pipeline.

Demo (quick smoke test)

Create a virtual environment, install dependencies, point at a SQLite file, and run the bundled demo script (no live Postgres required).

Linux / macOS (bash)

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

export DATABASE_URL="sqlite:///./demo_smoke.db"
python scripts/demo_load.py

Windows (PowerShell)

py -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt

$env:DATABASE_URL = "sqlite:///./demo_smoke.db"
python scripts/demo_load.py

To exercise the full CLI entrypoint against PostgreSQL, copy .env.example → .env, set DATABASE_URL (or the discrete POSTGRES_* variables used by Compose), ensure your CSV path via SALES_CSV_PATH, then run python main.py (see docs/development.md).

Features

Area	What you get
Extract	CSV via pandas with automatic header normalisation (e.g. Unit Price → unit_price).
Transform	Type coercion, defaults, and row validation before load.
Load	Bulk insert or PostgreSQL-native upsert with shared chunking/dedup helpers.
Audit	Every load batch recorded in a logs table via load_logging.
Resilience	Tenacity retries with exponential backoff for transient database failures (retry_db).
Scheduling	Optional recurring runs using the schedule library (RUN_SCHEDULED).
Migrations	Alembic with autogenerate wired to SQLAlchemy models.
Docker	docker-compose.yml — Postgres 16 plus an ETL image with health-aware startup.
GraphQL (optional)	Strawberry ASGI app: health, version, and safe ETL runtime hints.

Tech stack

Component	Role
Python 3.11+	Language and runtime
SQLAlchemy 2	ORM, sessions, schema bootstrap
psycopg2	PostgreSQL driver
pandas	CSV extraction
tenacity	Retry/backoff for DB operations
Alembic	Schema migrations
pytest / pytest-cov	Tests and coverage
Ruff	Lint + format
Strawberry + Starlette + Uvicorn	Optional GraphQL HTTP surface

Prerequisites

• Python 3.11+ and pip.
• PostgreSQL for production-style runs (main.py, Compose stack, --run-pg tests). SQLite is enough for demos and much of the unit suite.

Installation and usage

From source (recommended)

git clone https://github.com/esousa97/py-etl-pipeline.git
cd py-etl-pipeline
pip install -r requirements.txt
cp .env.example .env   # then edit credentials and paths

Development install (editable)

pip install -e .

PyPI

There are no PyPI install badges in this README yet because uploads to PyPI run only on a published GitHub Release (see .github/workflows/publish.yml). The same workflow supports Run workflow (manual dispatch): it builds wheels/sdists and uploads artifacts for inspection, without publishing. Configure PyPI trusted publishing (or a token) for the pypi environment for release publishes; then pip install py-etl-pipeline works after the first successful publish.

Dependency review runs as a CI job on every pull request (not on plain pushes to main), using actions/dependency-review-action alongside lint and tests.

Quick Start

One-shot pipeline (bulk)

python main.py

LOAD_MODE defaults to bulk. The pipeline bootstraps missing tables and reads the CSV configured by SALES_CSV_PATH.

Upsert mode

Linux / macOS

export LOAD_MODE=upsert
python main.py

Windows (PowerShell)

$env:LOAD_MODE = "upsert"
python main.py

Scheduled mode

Runs once at startup, then every SCHEDULE_INTERVAL_MINUTES (default 60):

Linux / macOS

export RUN_SCHEDULED=true
python main.py

Windows (PowerShell)

$env:RUN_SCHEDULED = "true"
python main.py

Optional GraphQL probe

uvicorn py_etl_pipeline.graphql_server:app --host 0.0.0.0 --port 8000

Example body against http://localhost:8000/graphql:

{"query": "query { health version }"}

Resilience (automatic retries)

Database-facing entrypoints are wrapped with retry_db (Tenacity): transient errors are retried with backoff before surfacing. Tune behaviour via the Tenacity configuration in src/py_etl_pipeline/retry.py.

Documentation

Document	Contents
LICENSE	MIT License
CONTRIBUTING.md	Contribution guidelines
CODE_OF_CONDUCT.md	Community standards
SECURITY.md	Vulnerability reporting
CHANGELOG.md	Version history
codecov.yml	Codecov defaults
.pre-commit-config.yaml	Local + CI-friendly Ruff hooks
docs/architecture.md	Components, data flow, schema
docs/configuration.md	Environment variables
docs/development.md	Setup, tests, migrations, Docker, scripts

Project layout

Path	Role
src/py_etl_pipeline/config.py	Environment loading and DSN construction
src/py_etl_pipeline/database.py	Engine, session factory, init_db
src/py_etl_pipeline/models.py	SQLAlchemy models (Sale, LogEntry, …)
src/py_etl_pipeline/extract.py	CSV → DataFrame
src/py_etl_pipeline/transform.py	Validation and coercion
src/py_etl_pipeline/load.py	Load strategy dispatcher
src/py_etl_pipeline/load_bulk.py	Bulk insert path
src/py_etl_pipeline/load_upsert.py	Upsert path (PG-native + safeguards)
src/py_etl_pipeline/load_logging.py	Audit log helper
src/py_etl_pipeline/load_utils.py	Chunking, dedup, key checks
src/py_etl_pipeline/pipeline.py	End-to-end orchestration
src/py_etl_pipeline/retry.py	Shared retry decorator
src/py_etl_pipeline/graphql_server.py	Optional Strawberry ASGI app
main.py	CLI entry: bootstrap + run (scheduled or one-shot)
migrations/	Alembic environment and revisions
scripts/	validate_load, diagnose_db, demo_load
tests/	pytest suite (SQLite by default; Postgres behind --run-pg)
.github/workflows/	CI (incl. PR dependency review), CodeQL, PyPI publish
docker-compose.yml / Dockerfile	Local stack and ETL image

Tests

pip install -r requirements.txt
pytest -q

PostgreSQL integration tests are opt-in:

pytest --run-pg

Coverage

Coverage is collected on Python 3.12 in CI and uploaded to Codecov.

pip install -r requirements.txt
pytest --cov=py_etl_pipeline --cov-report=term-missing

Contributing

See CONTRIBUTING.md.

Changelog

See CHANGELOG.md.

License

MIT.

Author

Enoque Sousa

⬆ Back to Top

Made with ❤️ by Enoque Sousa

Project status: Study project