Gas Price ETL Pipeline

An automated, modular Data Engineering pipeline built with Python that ingests regional fuel price metrics from the CollectAPI Gas Prices API, standardizes and transforms city-level datasets using Pandas, and loads structured records into a PostgreSQL database using SQLAlchemy and the psycopg2 driver.

---

Project Overview & Objectives

• Data Ingestion: Extract live regional fuel pricing data via CollectAPI REST endpoints.
• Data Transformation: Clean, filter, and standardize schema definitions (column renaming, dropping unused attributes) using Pandas.
• Data Loading: Model target table schemas and execute batch writes to PostgreSQL using SQLAlchemy ORM and Pandas to_sql().
• Software Design: Prototype pipeline logic interactively in Jupyter Notebook before refactoring into a modular, decoupled Python package.

---

Architecture & Code Structure

The project transitions from interactive prototyping (pipeline.ipynb) to production-style modular modules:

├── notebook/
│   └── pipeline-prototyping.ipynb         # etl pipeline prototyping in Jupyter
├── Etl/
│   ├── extract.py                # API request handling and JSON payload parsing
│   ├── transform.py              # Pandas schema cleaning, filtering, and transformation
│   └── load.py                   # PostgreSQL engine creation and batch loading
├── .envs
│   └──.env.example               # Environment variables template
├── images
│   └──db_verification.png        # Db Querying Image(SQL)
├── main.py                       # Pipeline orchestration script
├── .gitignore                    # Dependencies to ignore when shipping the project
├── requirements.txt              # Python dependencies
└── README.md

Quickstart & Setup

1. Environment Setup

Clone the repository and initialize a isolated Python virtual environment:

# Create a virtual environment
python -m venv GPricesVE

# Activate the virtual environment
# Windows (PowerShell / CMD)
GPricesVE\Scripts\activate

# Linux / macOS
source GPricesVE/bin/activate

Install project dependencies:

pip install -r requirements.txt

2. Configuration & Credentials

.env.example

# API Configuration
API_KEY=your_collectapi_key_here

# PostgreSQL Credentials
DB_HOST=
DB_NAME=
DB_USER=
DB_PASSWORD=your_secure_password
DB_PORT=5432

pipeline-prototyping.ipynb

headers = {
    'content-type': "application/json",
    'authorization': "apikeyhere"
    }

   engine = create_engine(
    # Enter your DB Credentials below
    f'postgresql+psycopg2://db_user:db_password@db_host:db_port/db_name')

3. Pipeline Execution

python main.py

Verification & Output

1. Execution Logs

The pipeline processes API responses and commits records directly to PostgreSQL.

2. Database Verification

Verification via SQL queries executed in pgAdmin confirms successful ingestion of target records:

SELECT * FROM gas_prices;