my-etl-package 1.0.0

A package for ETL pipeline operations

My ETL Package

This package implements a simple Extract–Transform–Load (ETL) pipeline using Python.
It processes CSV files from an input directory, transforms and integrate them into a single dataset, saves the results to an output directory, and loads the final dataset into a PostgreSQL database.

---

🔑 Pre-requisites

1. Working Python environment
   Make sure you have a working Python environment set up, either locally or in Jupyter.
   If you haven’t already, follow this guide:
   Setting up a Basic Python Development Environment

2. PostgreSQL database
   Ensure you have a PostgreSQL database installed and configured.
   If you don’t have one already, download and install both PostgreSQL and PgAdmin from the official sources below, then use PgAdmin to create your first database:

   • Download PostgreSQL
   • Download PgAdmin

   Once installed, you can follow this step-by-step guide to create a new database in PgAdmin:
   Creating a Database using PgAdmin

---

📦 Installation

Install directly from PyPI:

pip install my-etl-package

Set up your environment variables in a .env file (required for PostgreSQL connection):

DB_HOST=localhost
DB_PORT=5432
DB_NAME=mydatabase
DB_USER=myuser
DB_PASSWORD=mypassword

---

⚙️ Package Contents

After installation, you can inspect the available functions:

import my_etl_package
help(my_etl_package)

Typical contents:

NAME
    my_etl_package

PACKAGE CONTENTS
    utils

FUNCTIONS
    read_csv(file_path: pathlib.Path) -> pandas.DataFrame
    transform_data(dfs: List[pandas.DataFrame]) -> pandas.DataFrame
    write_csv(df: pandas.DataFrame, output_path: pathlib.Path) -> None
    load_to_db(df: pd.DataFrame, table_name: str, engine: sqlalchemy.engine.Engine) -> None

Utilities inside my_etl_package.utils:

FUNCTIONS
    list_csv_files(directory_path: pathlib.Path) -> List[pathlib.Path]
    PostgresConnector().get_db_connection() -> sqlalchemy.engine.Engine

---

📂 Data Directory Structure (Recommended but not Mandatory)

When running locally, organize your data as follows (relative to your current working directory):

pwd/
├── .env              # Environment variables (PostgreSQL credentials)
└── data/
    ├── raw/          # Place input CSV files here
    └── processed/    # Processed output CSVs will be written here

---

🛠️ Usage

1. Use Methods Individually

1.1. List all CSV files in a directory

from pathlib import Path
from my_etl_package.utils import list_csv_files

input_directory = Path().absolute() / "data/raw"
files = list_csv_files(input_directory)
print(files)

[WindowsPath('C:/Users/khhal/Documents/data/raw/103_semester_2_week_1_raw.csv'), WindowsPath('C:/Users/khhal/Documents/data/raw/104_semester_2_week_2_raw.csv')]

1.2. Read a CSV file

from my_etl_package import read_csv

df1 = read_csv(files[0])
df1

	from	message	status	date_sent	student_id	course_code	student_name
0	447440049121	3821656	received	2023-01-25 13:22:10+00:00	3821656	NaN	Khaled Ahmed
1	447440049121	3821656 103	received	2023-01-25 12:26:28+00:00	3821656	103.0	Khaled Ahmed

from my_etl_package import read_csv

df2 = read_csv(files[1])
df2

	from	message	status	date_sent	student_id	course_code	student_name	session
0	4.474400e+11	3821656 104	received	2023-02-02 13:24:02+00:00	3821656.0	104.0	Khaled Ahmed	NaN
1	NaN	NaN	NaN	NaN	NaN	NaN	NaN	NaN

1.3. Transform multiple CSVs

from my_etl_package import transform_data

combined_df = transform_data([df1, df2])
combined_df

	from	message	status	date_sent	student_id	course_code	student_name
0	4.474400e+11	3821656	received	2023-01-25 13:22:10+00:00	3821656.0	NaN	Khaled Ahmed
1	4.474400e+11	3821656 103	received	2023-01-25 12:26:28+00:00	3821656.0	103.0	Khaled Ahmed
2	4.474400e+11	3821656 104	received	2023-02-02 13:24:02+00:00	3821656.0	104.0	Khaled Ahmed

1.4. Write processed DataFrame to CSV

from pathlib import Path
from my_etl_package import write_csv

output_path = Path().absolute() / "data/processed"
output_path.mkdir(exist_ok=True)
output_file = output_path / "processed.csv"
write_csv(combined_df, output_file)

1.5. Load DataFrame into PostgreSQL

from dotenv import load_dotenv
from my_etl_package.utils import PostgresConnector
from my_etl_package import load_to_db


# Load environment variables
load_dotenv()

# If .env is in a different location, specify the path:
# load_dotenv('./some_other_location/.env')

table_name = "etl_pipeline_processed"
load_to_db(combined_df, table_name)

INFO:my_etl_package.utils.connect_db:PostgresConnector initialized with loaded credentials.
INFO:my_etl_package.utils.connect_db:Database engine generated for: postgresql://postgres:Khal8891@localhost:5434/postgres.

2. Run the Full ETL Pipeline

Here’s an end-to-end pipeline script:

In etl_pipeline.py (name as you wish) in the current working directory ->

import logging
from pathlib import Path
from dotenv import load_dotenv
from my_etl_package.utils import list_csv_files, PostgresConnector
from my_etl_package import read_csv, transform_data, write_csv, load_to_db

# Configure logging
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)

# Load environment variables
load_dotenv()
logging.info("Environment variables loaded.")

# Set up workspace
base = Path().absolute() / "data"
input_directory = base / "raw"
output_directory = base / "processed"
output_filename = "etl_pipeline_processed.csv"
output_path = output_directory / output_filename

# Create output directory if it doesn't exist
output_directory.mkdir(exist_ok=True)
logging.info(f"Output directory set to: {output_directory}")


def main():
    logging.info("Starting ETL pipeline...")

    # Configuration
    table_name = "etl_pipeline_processed"
    logging.info(f"Using table: {table_name}")

    # Extract
    logging.info(f"Looking for CSV files in: {input_directory}")
    file_paths = list_csv_files(input_directory)
    logging.info(f"Found {len(file_paths)} CSV files.")

    # Read
    dfs = (read_csv(f) for f in file_paths)

    # Transform
    logging.info("Transforming data...")
    combined_df = transform_data(dfs)
    logging.info("Data transformation complete.")

    # Load - write to CSV
    logging.info("Writing processed data to CSV...")
    write_csv(combined_df, output_path)
    logging.info("Data written to CSV.")

    # Load - load into Postgres
    logging.info("Loading data into PostgreSQL...")
    load_to_db(combined_df, table_name)
    logging.info("Data successfully loaded into PostgreSQL.")

    logging.info("ETL pipeline finished.")

main()

INFO:root:Environment variables loaded.
INFO:root:Output directory set to: C:\Users\khhal\Documents\data\processed
INFO:root:Starting ETL pipeline...
INFO:root:Using table: etl_pipeline_processed
INFO:root:Looking for CSV files in: C:\Users\khhal\Documents\data\raw
INFO:root:Found 2 CSV files.
INFO:root:Transforming data...
INFO:root:Data transformation complete.
INFO:root:Writing processed data to CSV...
INFO:root:Data written to CSV.
INFO:root:Loading data into PostgreSQL...
INFO:my_etl_package.utils.connect_db:PostgresConnector initialized with loaded credentials.
INFO:my_etl_package.utils.connect_db:Database engine generated for: postgresql://postgres:Khal8891@localhost:5434/postgres.
INFO:root:Data successfully loaded into PostgreSQL.
INFO:root:ETL pipeline finished.

Run it:

python etl_pipeline.py

---

✅ Features

• 🔎 Automatically detects all CSV files in data/raw/
• 🛠️ Cleans and transforms raw datasets
• 💾 Stores processed results in data/processed/
• 🗄️ Loads final output into a PostgreSQL table

---

📝 Notes

• Ensure PostgreSQL is running and accessible with the credentials in your .env file.
• Place your input CSV files in the same input directory before running the pipeline.

---