harvard-art-museums-etl-pipeline

---

name: harvard-art-museums-etl-pipeline description: Build ETL pipelines and analytics dashboards using Harvard Art Museums API with Python, SQL, and Streamlit triggers:

• how do I extract data from Harvard Art Museums API
• build an ETL pipeline for museum artifacts
• create a Streamlit dashboard for art collection data
• query Harvard Art Museums API with pagination
• design SQL schema for artifact metadata
• visualize museum collection analytics with Plotly
• transform nested JSON art data into relational tables
• set up data engineering pipeline for cultural heritage data

---

Harvard Art Museums ETL Pipeline

Skill by ara.so — Data Skills collection.

This project provides an end-to-end data engineering solution for extracting, transforming, and analyzing artifacts from the Harvard Art Museums API. It demonstrates real-world ETL patterns, SQL analytics, and interactive visualization using Streamlit.

What It Does

• Extracts artifact data from Harvard Art Museums API with pagination and rate limiting
• Transforms nested JSON into normalized relational tables (metadata, media, colors)
• Loads data into MySQL/TiDB Cloud with proper foreign key relationships
• Analyzes data using predefined SQL queries for cultural insights
• Visualizes results through interactive Plotly charts in a Streamlit dashboard

Installation

# Clone the repository
git clone https://github.com/Manali0711/Harvard-Artifacts-Collection-Data-Engineering-Analytics-App.git
cd Harvard-Artifacts-Collection-Data-Engineering-Analytics-App

# Install dependencies
pip install -r requirements.txt

# Required packages
pip install streamlit pandas requests mysql-connector-python plotly python-dotenv

Configuration

API Key Setup

Get your Harvard Art Museums API key from: https://www.harvardartmuseums.org/collections/api

# Store in environment variables
# .env file
HARVARD_API_KEY=your_api_key_here
DB_HOST=your_database_host
DB_USER=your_database_user
DB_PASSWORD=your_database_password
DB_NAME=harvard_artifacts

Database Connection

import mysql.connector
import os
from dotenv import load_dotenv

load_dotenv()

def get_db_connection():
    """Create MySQL database connection"""
    return mysql.connector.connect(
        host=os.getenv('DB_HOST'),
        user=os.getenv('DB_USER'),
        password=os.getenv('DB_PASSWORD'),
        database=os.getenv('DB_NAME')
    )

Database Schema

Create Tables

-- Artifact Metadata Table
CREATE TABLE artifactmetadata (
    id INT PRIMARY KEY,
    title VARCHAR(500),
    culture VARCHAR(255),
    period VARCHAR(255),
    century VARCHAR(100),
    classification VARCHAR(255),
    department VARCHAR(255),
    dated VARCHAR(255),
    accession_year INT,
    primary_image_url TEXT,
    url TEXT
);

-- Artifact Media Table
CREATE TABLE artifactmedia (
    media_id INT AUTO_INCREMENT PRIMARY KEY,
    artifact_id INT,
    media_type VARCHAR(100),
    base_url TEXT,
    image_url TEXT,
    description TEXT,
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id) ON DELETE CASCADE
);

-- Artifact Colors Table
CREATE TABLE artifactcolors (
    color_id INT AUTO_INCREMENT PRIMARY KEY,
    artifact_id INT,
    color VARCHAR(50),
    hex_code VARCHAR(10),
    percentage DECIMAL(5,2),
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id) ON DELETE CASCADE
);

Key API Patterns

Extract: Fetch Artifacts with Pagination

import requests
import os
from typing import List, Dict

def fetch_artifacts(api_key: str, num_pages: int = 5, page_size: int = 100) -> List[Dict]:
    """
    Fetch artifacts from Harvard Art Museums API with pagination
    
    Args:
        api_key: Harvard API key from environment
        num_pages: Number of pages to fetch
        page_size: Records per page (max 100)
    
    Returns:
        List of artifact dictionaries
    """
    base_url = "https://api.harvardartmuseums.org/object"
    all_artifacts = []
    
    for page in range(1, num_pages + 1):
        params = {
            'apikey': api_key,
            'size': page_size,
            'page': page,
            'hasimage': 1  # Only artifacts with images
        }
        
        try:
            response = requests.get(base_url, params=params, timeout=30)
            response.raise_for_status()
            data = response.json()
            
            records = data.get('records', [])
            all_artifacts.extend(records)
            
            print(f"Fetched page {page}: {len(records)} artifacts")
            
        except requests.exceptions.RequestException as e:
            print(f"Error fetching page {page}: {e}")
            break
    
    return all_artifacts

Transform: Normalize Nested JSON

import pandas as pd
from typing import List, Dict, Tuple

def transform_artifacts(raw_data: List[Dict]) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    """
    Transform raw API data into normalized dataframes
    
    Returns:
        Tuple of (metadata_df, media_df, colors_df)
    """
    metadata_list = []
    media_list = []
    colors_list = []
    
    for artifact in raw_data:
        # Extract metadata
        metadata = {
            'id': artifact.get('id'),
            'title': artifact.get('title', '')[:500],
            'culture': artifact.get('culture', '')[:255],
            'period': artifact.get('period', '')[:255],
            'century': artifact.get('century', '')[:100],
            'classification': artifact.get('classification', '')[:255],
            'department': artifact.get('department', '')[:255],
            'dated': artifact.get('dated', '')[:255],
            'accession_year': artifact.get('accessionyear'),
            'primary_image_url': artifact.get('primaryimageurl', ''),
            'url': artifact.get('url', '')
        }
        metadata_list.append(metadata)
        
        # Extract media
        images = artifact.get('images', [])
        for img in images:
            media = {
                'artifact_id': artifact.get('id'),
                'media_type': 'image',
                'base_url': img.get('baseimageurl', ''),
                'image_url': img.get('iiifbaseuri', ''),
                'description': img.get('description', '')[:1000] if img.get('description') else None
            }
            media_list.append(media)
        
        # Extract colors
        colors = artifact.get('colors', [])
        for color in colors:
            color_data = {
                'artifact_id': artifact.get('id'),
                'color': color.get('color', ''),
                'hex_code': color.get('hex', ''),
                'percentage': color.get('percent', 0)
            }
            colors_list.append(color_data)
    
    return (
        pd.DataFrame(metadata_list),
        pd.DataFrame(media_list),
        pd.DataFrame(colors_list)
    )

Load: Batch Insert into Database

from typing import List
import mysql.connector

def load_metadata(df: pd.DataFrame, connection):
    """Load artifact metadata with batch insert"""
    cursor = connection.cursor()
    
    insert_query = """
    INSERT INTO artifactmetadata 
    (id, title, culture, period, century, classification, department, 
     dated, accession_year, primary_image_url, url)
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    ON DUPLICATE KEY UPDATE
    title=VALUES(title), culture=VALUES(culture), period=VALUES(period)
    """
    
    # Convert dataframe to list of tuples
    data = [tuple(row) for row in df.values]
    
    cursor.executemany(insert_query, data)
    connection.commit()
    print(f"Inserted {cursor.rowcount} metadata records")

def load_media(df: pd.DataFrame, connection):
    """Load artifact media"""
    cursor = connection.cursor()
    
    insert_query = """
    INSERT INTO artifactmedia 
    (artifact_id, media_type, base_url, image_url, description)
    VALUES (%s, %s, %s, %s, %s)
    """
    
    data = [tuple(row) for row in df.values]
    cursor.executemany(insert_query, data)
    connection.commit()
    print(f"Inserted {cursor.rowcount} media records")

def load_colors(df: pd.DataFrame, connection):
    """Load artifact colors"""
    cursor = connection.cursor()
    
    insert_query = """
    INSERT INTO artifactcolors 
    (artifact_id, color, hex_code, percentage)
    VALUES (%s, %s, %s, %s)
    """
    
    data = [tuple(row) for row in df.values]
    cursor.executemany(insert_query, data)
    connection.commit()
    print(f"Inserted {cursor.rowcount} color records")

Streamlit Dashboard

Main Application Structure

import streamlit as st
import pandas as pd
import plotly.express as px
from dotenv import load_dotenv
import os

load_dotenv()

st.set_page_config(
    page_title="Harvard Art Museums Analytics",
    page_icon="🎨",
    layout="wide"
)

def main():
    st.title("🎨 Harvard Art Museums Analytics Dashboard")
    
    # Sidebar navigation
    page = st.sidebar.selectbox(
        "Choose a page",
        ["ETL Pipeline", "SQL Analytics", "Visualizations"]
    )
    
    if page == "ETL Pipeline":
        show_etl_page()
    elif page == "SQL Analytics":
        show_analytics_page()
    else:
        show_visualizations_page()

if __name__ == "__main__":
    main()

ETL Pipeline Page

def show_etl_page():
    st.header("ETL Pipeline")
    
    api_key = st.text_input("Harvard API Key", type="password", 
                            value=os.getenv('HARVARD_API_KEY', ''))
    
    col1, col2 = st.columns(2)
    with col1:
        num_pages = st.number_input("Number of pages", min_value=1, max_value=100, value=5)
    with col2:
        page_size = st.number_input("Page size", min_value=10, max_value=100, value=100)
    
    if st.button("🚀 Run ETL Pipeline"):
        with st.spinner("Extracting data from API..."):
            raw_data = fetch_artifacts(api_key, num_pages, page_size)
            st.success(f"✅ Extracted {len(raw_data)} artifacts")
        
        with st.spinner("Transforming data..."):
            metadata_df, media_df, colors_df = transform_artifacts(raw_data)
            st.success(f"✅ Transformed into {len(metadata_df)} metadata, {len(media_df)} media, {len(colors_df)} color records")
        
        with st.spinner("Loading into database..."):
            conn = get_db_connection()
            load_metadata(metadata_df, conn)
            load_media(media_df, conn)
            load_colors(colors_df, conn)
            conn.close()
            st.success("✅ Data loaded successfully!")
        
        # Show preview
        st.subheader("Data Preview")
        st.dataframe(metadata_df.head(10))

Analytics Queries

Common SQL Query Patterns

# Top 10 cultures by artifact count
QUERY_TOP_CULTURES = """
SELECT culture, COUNT(*) as artifact_count
FROM artifactmetadata
WHERE culture IS NOT NULL AND culture != ''
GROUP BY culture
ORDER BY artifact_count DESC
LIMIT 10;
"""

# Artifacts by century
QUERY_BY_CENTURY = """
SELECT century, COUNT(*) as count
FROM artifactmetadata
WHERE century IS NOT NULL
GROUP BY century
ORDER BY count DESC;
"""

# Department distribution
QUERY_BY_DEPARTMENT = """
SELECT department, COUNT(*) as count
FROM artifactmetadata
WHERE department IS NOT NULL
GROUP BY department
ORDER BY count DESC;
"""

# Most common colors
QUERY_TOP_COLORS = """
SELECT color, COUNT(*) as frequency, AVG(percentage) as avg_percentage
FROM artifactcolors
GROUP BY color
ORDER BY frequency DESC
LIMIT 15;
"""

# Artifacts with most images
QUERY_MOST_IMAGES = """
SELECT m.title, m.culture, COUNT(med.media_id) as image_count
FROM artifactmetadata m
JOIN artifactmedia med ON m.id = med.artifact_id
GROUP BY m.id, m.title, m.culture
ORDER BY image_count DESC
LIMIT 10;
"""

def execute_query(query: str, connection) -> pd.DataFrame:
    """Execute SQL query and return results as DataFrame"""
    return pd.read_sql(query, connection)

Analytics Dashboard Page

def show_analytics_page():
    st.header("📊 SQL Analytics")
    
    queries = {
        "Top 10 Cultures": QUERY_TOP_CULTURES,
        "Artifacts by Century": QUERY_BY_CENTURY,
        "Department Distribution": QUERY_BY_DEPARTMENT,
        "Most Common Colors": QUERY_TOP_COLORS,
        "Artifacts with Most Images": QUERY_MOST_IMAGES
    }
    
    selected_query = st.selectbox("Select Analysis", list(queries.keys()))
    
    if st.button("Run Query"):
        conn = get_db_connection()
        df = execute_query(queries[selected_query], conn)
        conn.close()
        
        st.subheader("Query Results")
        st.dataframe(df)
        
        # Auto-generate visualization
        if len(df.columns) >= 2:
            fig = px.bar(df, x=df.columns[0], y=df.columns[1],
                        title=selected_query)
            st.plotly_chart(fig, use_container_width=True)

Visualization Patterns

Create Interactive Charts

import plotly.express as px
import plotly.graph_objects as go

def create_culture_chart(df: pd.DataFrame):
    """Bar chart for culture distribution"""
    fig = px.bar(
        df, 
        x='culture', 
        y='artifact_count',
        title='Top Cultures in Collection',
        labels={'artifact_count': 'Number of Artifacts', 'culture': 'Culture'},
        color='artifact_count',
        color_continuous_scale='Viridis'
    )
    fig.update_layout(xaxis_tickangle=-45)
    return fig

def create_century_timeline(df: pd.DataFrame):
    """Timeline visualization for centuries"""
    fig = px.line(
        df, 
        x='century', 
        y='count',
        title='Artifact Distribution Across Centuries',
        markers=True
    )
    return fig

def create_color_pie_chart(df: pd.DataFrame):
    """Pie chart for color distribution"""
    fig = px.pie(
        df, 
        values='frequency', 
        names='color',
        title='Most Common Colors in Artifacts',
        hole=0.3
    )
    return fig

Complete ETL Workflow

from dotenv import load_dotenv
import os

def run_complete_etl():
    """Complete ETL pipeline execution"""
    load_dotenv()
    
    # 1. Extract
    print("Starting extraction...")
    api_key = os.getenv('HARVARD_API_KEY')
    raw_data = fetch_artifacts(api_key, num_pages=10, page_size=100)
    print(f"Extracted {len(raw_data)} artifacts")
    
    # 2. Transform
    print("Transforming data...")
    metadata_df, media_df, colors_df = transform_artifacts(raw_data)
    
    # Data quality checks
    print(f"Metadata records: {len(metadata_df)}")
    print(f"Media records: {len(media_df)}")
    print(f"Color records: {len(colors_df)}")
    print(f"Null values in metadata:\n{metadata_df.isnull().sum()}")
    
    # 3. Load
    print("Loading to database...")
    conn = get_db_connection()
    
    try:
        load_metadata(metadata_df, conn)
        load_media(media_df, conn)
        load_colors(colors_df, conn)
        print("ETL pipeline completed successfully!")
    except Exception as e:
        print(f"Error during load: {e}")
        conn.rollback()
    finally:
        conn.close()

if __name__ == "__main__":
    run_complete_etl()

Troubleshooting

API Rate Limiting

import time
from requests.exceptions import HTTPError

def fetch_with_retry(url, params, max_retries=3):
    """Fetch with exponential backoff for rate limiting"""
    for attempt in range(max_retries):
        try:
            response = requests.get(url, params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except HTTPError as e:
            if e.response.status_code == 429:  # Too Many Requests
                wait_time = (2 ** attempt) * 5  # Exponential backoff
                print(f"Rate limited. Waiting {wait_time}s...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded")

Database Connection Issues

import mysql.connector
from mysql.connector import Error

def safe_db_connection():
    """Create database connection with error handling"""
    try:
        connection = mysql.connector.connect(
            host=os.getenv('DB_HOST'),
            user=os.getenv('DB_USER'),
            password=os.getenv('DB_PASSWORD'),
            database=os.getenv('DB_NAME'),
            connect_timeout=10,
            autocommit=False
        )
        if connection.is_connected():
            print("Database connection successful")
            return connection
    except Error as e:
        print(f"Database connection error: {e}")
        return None

Handling Missing Data

def clean_artifact_data(df: pd.DataFrame) -> pd.DataFrame:
    """Clean and validate artifact data"""
    # Replace empty strings with None
    df = df.replace('', None)
    
    # Handle missing titles
    df['title'] = df['title'].fillna('Untitled')
    
    # Ensure numeric fields
    df['accession_year'] = pd.to_numeric(df['accession_year'], errors='coerce')
    
    # Truncate long text fields
    df['title'] = df['title'].str[:500]
    df['culture'] = df['culture'].str[:255]
    
    return df

Running the Application

# Start Streamlit dashboard
streamlit run app.py

# Run ETL pipeline only
python etl_pipeline.py

# Run with custom configuration
HARVARD_API_KEY=your_key DB_HOST=localhost streamlit run app.py

This skill enables AI agents to help developers build complete ETL pipelines for cultural heritage data, from API extraction through SQL analytics to interactive visualization.