warehouse_loader.py

import os
import sys
import psycopg2


def get_connection():
    try:
        return psycopg2.connect(
            host=os.getenv("DB_HOST", "localhost"),
            port=os.getenv("DB_PORT", "5432"),
            dbname=os.getenv("DB_NAME", "crypto_db"),
            user=os.getenv("DB_USER", "postgres"),
            password=os.getenv("DB_PASSWORD", "1998")
        )
    except Exception as e:
        print(f"[ERROR] Database connection failed: {e}")
        sys.exit(1)


def load_dim_coin(cursor):
    query = """
        INSERT INTO dim_coin (coin_id, name, symbol)
        SELECT DISTINCT coin_id, name, symbol
        FROM crypto_prices
        WHERE coin_id IS NOT NULL
        ON CONFLICT (coin_id) DO NOTHING;
    """
    cursor.execute(query)
    print("[INFO] dim_coin loaded.")


def load_dim_date(cursor):
    query = """
        INSERT INTO dim_date (date_id)
        SELECT DISTINCT DATE(fetched_at)
        FROM crypto_prices
        WHERE fetched_at IS NOT NULL
        ON CONFLICT (date_id) DO NOTHING;
    """
    cursor.execute(query)
    print("[INFO] dim_date loaded.")


def load_fact_table(cursor):
    query = """
        INSERT INTO fact_crypto_prices (
            coin_id,
            date_id,
            current_price,
            market_cap,
            total_volume,
            price_change_percentage_24h
        )
        SELECT
            coin_id,
            DATE(fetched_at) AS date_id,
            current_price,
            market_cap,
            total_volume,
            price_change_percentage_24h
        FROM crypto_prices
        WHERE coin_id IS NOT NULL
          AND fetched_at IS NOT NULL
        ON CONFLICT DO NOTHING;
    """
    cursor.execute(query)
    print("[INFO] fact_crypto_prices loaded.")


def main():
    print("[INFO] Starting warehouse loading...")

    conn = get_connection()
    cursor = conn.cursor()

    try:
        load_dim_coin(cursor)
        load_dim_date(cursor)
        load_fact_table(cursor)

        conn.commit()
        print("[SUCCESS] Data warehouse loaded successfully.")

    except Exception as e:
        conn.rollback()
        print(f"[ERROR] Warehouse loading failed: {e}")
        sys.exit(1)

    finally:
        cursor.close()
        conn.close()


if __name__ == "__main__":
    main()