""" load_to_postgres.py Pawsome Provisions Pet Food Facility -------------------------------------- Creates the PostgreSQL schema for the Pawsome Provisions BI pipeline and bulk-loads three CSV datasets: 1. inventory_seed.csv → inventory table (8 SKUs, stock levels, thresholds) 2. sales_june_2025.csv → sales table (June 2025 order lines) 3. reorder_log.csv → reorder_log table (output from inventory_depletion.py) Tables are created with IF NOT EXISTS — safe to run multiple times. Existing data is not deleted; duplicate runs will append rows. Requirements: pip install psycopg2-binary Environment: Written and tested on Ubuntu 22.04. PostgreSQL running via Docker (see docker-compose.yml). Usage: python3 load_to_postgres.py """ import csv import os import psycopg2 from psycopg2 import sql from datetime import datetime # ── Connection Settings ─────────────────────────────────────────────────────── # Adjust these to match your docker-compose.yml environment variables. DB_CONFIG = { "host": "localhost", "port": 5432, "dbname": "pawsome", "user": "postgres", "password": "postgres", } # ── File Paths ──────────────────────────────────────────────────────────────── # Assumes CSVs are in the same directory as this script. # Adjust paths if your folder structure differs. BASE_DIR = os.path.dirname(os.path.abspath(__file__)) INVENTORY_SEED_CSV = os.path.join(BASE_DIR, "inventory_seed.csv") SALES_CSV = os.path.join(BASE_DIR, "sales_june_2025.csv") REORDER_LOG_CSV = os.path.join(BASE_DIR, "reorder_log.csv") # ── Schema Definitions ──────────────────────────────────────────────────────── CREATE_INVENTORY_TABLE = """ CREATE TABLE IF NOT EXISTS inventory ( sku VARCHAR(10) PRIMARY KEY, product_name VARCHAR(100) NOT NULL, category VARCHAR(20) NOT NULL, current_stock INTEGER NOT NULL, reorder_threshold INTEGER NOT NULL, max_capacity INTEGER NOT NULL, last_restocked DATE ); """ CREATE_SALES_TABLE = """ CREATE TABLE IF NOT EXISTS sales ( sale_id SERIAL PRIMARY KEY, sku VARCHAR(10) REFERENCES inventory(sku), quantity INTEGER NOT NULL, unit_price NUMERIC(6,2) NOT NULL, order_date DATE NOT NULL, region VARCHAR(20) ); """ CREATE_REORDER_LOG_TABLE = """ CREATE TABLE IF NOT EXISTS reorder_log ( log_id SERIAL PRIMARY KEY, sku VARCHAR(10) REFERENCES inventory(sku), triggered_date DATE NOT NULL, quantity_ordered INTEGER NOT NULL, expected_delivery DATE NOT NULL, fulfilled BOOLEAN DEFAULT false ); """ # ── Helpers ─────────────────────────────────────────────────────────────────── def read_csv(filepath: str) -> list: """Read a CSV file and return a list of row dictionaries.""" if not os.path.exists(filepath): raise FileNotFoundError(f"CSV not found: {filepath}") with open(filepath, newline="", encoding="utf-8") as f: return list(csv.DictReader(f)) def parse_bool(value: str) -> bool: """Convert common string representations of booleans to Python bool.""" return value.strip().lower() in ("true", "1", "yes") def parse_date(value: str): """Return a date object or None for empty strings.""" value = value.strip() if not value: return None return datetime.strptime(value, "%Y-%m-%d").date() # ── Table Creation ──────────────────────────────────────────────────────────── def create_tables(cursor): """Create all three tables if they do not already exist.""" print("Creating tables (IF NOT EXISTS)...") # inventory must be created before sales and reorder_log # because both reference inventory.sku as a foreign key cursor.execute(CREATE_INVENTORY_TABLE) print(" ✔ inventory") cursor.execute(CREATE_SALES_TABLE) print(" ✔ sales") cursor.execute(CREATE_REORDER_LOG_TABLE) print(" ✔ reorder_log") # ── Data Loading ────────────────────────────────────────────────────────────── def load_inventory(cursor, filepath: str): """ Load inventory_seed.csv into the inventory table. Expected columns: sku, product_name, category, current_stock, reorder_threshold, max_capacity, last_restocked """ rows = read_csv(filepath) insert_sql = """ INSERT INTO inventory (sku, product_name, category, current_stock, reorder_threshold, max_capacity, last_restocked) VALUES (%s, %s, %s, %s, %s, %s, %s) ON CONFLICT (sku) DO NOTHING; """ count = 0 for row in rows: cursor.execute(insert_sql, ( row["sku"].strip(), row["product_name"].strip(), row["category"].strip(), int(row["current_stock"]), int(row["reorder_threshold"]), int(row["max_capacity"]), parse_date(row.get("last_restocked", "")), )) count += 1 print(f" ✔ inventory — {count} rows processed " f"({cursor.rowcount} inserted, duplicates skipped)") def load_sales(cursor, filepath: str): """ Load sales_june_2025.csv into the sales table. Expected columns: sku, quantity, unit_price, order_date, region """ rows = read_csv(filepath) insert_sql = """ INSERT INTO sales (sku, quantity, unit_price, order_date, region) VALUES (%s, %s, %s, %s, %s); """ count = 0 for row in rows: cursor.execute(insert_sql, ( row["sku"].strip(), int(row["quantity"]), float(row["unit_price"]), parse_date(row["order_date"]), row.get("region", "").strip() or None, )) count += 1 print(f" ✔ sales — {count} rows inserted") def load_reorder_log(cursor, filepath: str): """ Load reorder_log.csv into the reorder_log table. This file is generated by inventory_depletion.py. Expected columns: sku, triggered_date, quantity_ordered, expected_delivery, fulfilled """ rows = read_csv(filepath) insert_sql = """ INSERT INTO reorder_log (sku, triggered_date, quantity_ordered, expected_delivery, fulfilled) VALUES (%s, %s, %s, %s, %s); """ count = 0 for row in rows: cursor.execute(insert_sql, ( row["sku"].strip(), parse_date(row["triggered_date"]), int(row["quantity_ordered"]), parse_date(row["expected_delivery"]), parse_bool(row.get("fulfilled", "false")), )) count += 1 print(f" ✔ reorder_log — {count} rows inserted") # ── Entry Point ─────────────────────────────────────────────────────────────── def main(): print("=" * 60) print(" PAWSOME PROVISIONS — PostgreSQL Data Loader") print("=" * 60) print(f"\nConnecting to PostgreSQL at {DB_CONFIG['host']}:{DB_CONFIG['port']} " f"→ database '{DB_CONFIG['dbname']}'...") try: conn = psycopg2.connect(**DB_CONFIG) conn.autocommit = False cursor = conn.cursor() print(" ✔ Connected\n") # Step 1 — Create schema create_tables(cursor) # Step 2 — Load data in dependency order # inventory must be loaded before sales and reorder_log # because of the foreign key constraint on sku print("\nLoading data...") load_inventory(cursor, INVENTORY_SEED_CSV) load_sales(cursor, SALES_CSV) load_reorder_log(cursor, REORDER_LOG_CSV) # Step 3 — Commit everything as one transaction conn.commit() print("\n ✔ All data committed successfully") except FileNotFoundError as e: print(f"\n ✘ File error: {e}") print(" Check that all three CSVs are in the same directory as this script.") except psycopg2.OperationalError as e: print(f"\n ✘ Could not connect to PostgreSQL: {e}") print(" Is your Docker container running? Try: docker compose up -d") except psycopg2.Error as e: print(f"\n ✘ Database error: {e}") if conn: conn.rollback() print(" Transaction rolled back — no data was written.") finally: if cursor: cursor.close() if conn: conn.close() print("\n ✔ Connection closed") print("\n" + "=" * 60) print(" Done. Open Metabase at http://localhost:3000") print("=" * 60) if __name__ == "__main__": main()