Release v1 of Date Dimension Population Script

build_dim_date.py

'''
File:      build_dim_date.py
Project:   fz
Created:   2024-02-03
Author:    Adge Denkers
           https://github.com/adgedenkers/
License:   MIT License
'''

import psycopg2
import keyring
from datetime import date, timedelta
import calendar
import holidays

# Define holidays for the United States
us_holidays = holidays.UnitedStates()

def create_date_range(start_date, end_date):
    """Generate a range of dates from start_date to end_date."""
    for n in range(int((end_date - start_date).days)):
        yield start_date + timedelta(n)

def get_special_day(single_date):
    """Determine special days like '3rd Monday', '1st Thursday', etc."""
    week_number, weekday = divmod(single_date.day - 1, 7)
    weekday_name = calendar.day_name[single_date.weekday()]
    return f"{week_number + 1} {weekday_name}" if weekday == 0 else None

def get_date_parts(single_date):
    """Break down a date into its constituent parts."""
    return {
        'date': single_date,
        'year': single_date.year,
        'quarter': (single_date.month - 1) // 3 + 1,
        'month': single_date.month,
        'month_name': single_date.strftime("%B"),
        'week_of_year': single_date.isocalendar()[1],
        'day_of_month': single_date.day,
        'day_of_week': single_date.isoweekday(),
        'day_name': calendar.day_name[single_date.weekday()],
        'special_day': get_special_day(single_date),
        'is_weekend': single_date.weekday() >= 5,
        'is_holiday': single_date in us_holidays,
        'holiday_name': us_holidays.get(single_date, ''),
        'federal_holiday': single_date in us_holidays
    }

def insert_date_into_db(date_parts, connection):
    """Insert a record into the dim_date table."""
    with connection.cursor() as cursor:
        try:
            cursor.execute("""
                INSERT INTO dim_date (
                    date, year, quarter, month, month_name, week_of_year, 
                    day_of_month, day_of_week, day_name, is_weekend, is_holiday, holiday_name, federal_holiday
                ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            """, (
                date_parts['date'], date_parts['year'], date_parts['quarter'], 
                date_parts['month'], date_parts['month_name'], date_parts['week_of_year'],
                date_parts['day_of_month'], date_parts['day_of_week'], date_parts['day_name'],
                date_parts['is_weekend'], date_parts['is_holiday'], date_parts['holiday_name'],
                date_parts['federal_holiday']
            ))
            connection.commit()
            return True
        except Exception as e:
            print(f"Error inserting {date_parts['date']}: {e}")
            connection.rollback()
            return False

def main():
    """Main function to insert date data into a PostgreSQL database."""
    conn_params = {
        'host': 'localhost',
        'port': 5432,
        'database': 'postgres',
        'user': 'username',
        'password': keyring.get_password("db-pass-postgres", "username")
    }

    conn = None
    try:
        conn = psycopg2.connect(**conn_params)

        start_date = date(2023, 1, 1)
        end_date = date(2050, 1, 1)

        for single_date in create_date_range(start_date, end_date):
            date_parts = get_date_parts(single_date)
            if not insert_date_into_db(date_parts, conn):
                break  # Stop if insertion fails

        print("All dates have been inserted.")
    except Exception as e:
        print(f"Database connection failed: {e}")
    finally:
        if conn:
            conn.close()

if __name__ == "__main__":
    main()