harvard-art-museums-etl-analytics

---

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

• build an ETL pipeline for Harvard Art Museums data
• create analytics dashboard with museum artifact data
• extract and transform Harvard museum API data
• set up data engineering pipeline for art collections
• visualize Harvard Art Museums data with Streamlit
• query and analyze museum artifacts with SQL
• implement data pipeline for cultural heritage data
• process Harvard Art Museums API with Python

---

Harvard Art Museums ETL Analytics

Skill by ara.so — Data Skills collection.

This skill enables AI coding agents to build end-to-end data engineering and analytics applications using the Harvard Art Museums API. The project demonstrates real-world ETL pipelines, SQL analytics, and interactive data visualization using Streamlit.

What This Project Does

The Harvard-Artifacts-Collection-Data-Engineering-Analytics-App provides:

• API Integration: Collect artifact data from Harvard Art Museums API with pagination and rate limiting
• ETL Pipeline: Extract, transform, and load artifact metadata, media details, and color data
• SQL Storage: Store structured data in MySQL/TiDB Cloud with proper relational schema
• Analytics Engine: Execute 20+ predefined analytical SQL queries
• Interactive Dashboards: Visualize query results using Streamlit and Plotly

The architecture follows: API → ETL → SQL → Analytics → Visualization

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

Environment Variables

Create a .env file in the project root:

# Harvard Art Museums API
HARVARD_API_KEY=your_api_key_here

# Database Configuration
DB_HOST=your_database_host
DB_PORT=3306
DB_USER=your_database_user
DB_PASSWORD=your_database_password
DB_NAME=harvard_artifacts

Database Setup

import mysql.connector
from mysql.connector import Error

def create_database_schema():
    """Initialize database schema for Harvard artifacts"""
    connection = mysql.connector.connect(
        host=os.getenv('DB_HOST'),
        port=os.getenv('DB_PORT'),
        user=os.getenv('DB_USER'),
        password=os.getenv('DB_PASSWORD')
    )
    
    cursor = connection.cursor()
    
    # Create database
    cursor.execute(f"CREATE DATABASE IF NOT EXISTS {os.getenv('DB_NAME')}")
    cursor.execute(f"USE {os.getenv('DB_NAME')}")
    
    # Create artifact metadata table
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS artifactmetadata (
            id INT PRIMARY KEY,
            title VARCHAR(500),
            culture VARCHAR(255),
            century VARCHAR(100),
            classification VARCHAR(255),
            department VARCHAR(255),
            dated VARCHAR(255),
            period VARCHAR(255),
            technique VARCHAR(500),
            description TEXT,
            url VARCHAR(500),
            dated_end INT,
            dated_begin INT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    
    # Create artifact media table
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS artifactmedia (
            media_id INT AUTO_INCREMENT PRIMARY KEY,
            artifact_id INT,
            image_url VARCHAR(500),
            media_type VARCHAR(100),
            FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
        )
    """)
    
    # Create artifact colors table
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS artifactcolors (
            color_id INT AUTO_INCREMENT PRIMARY KEY,
            artifact_id INT,
            color_hex VARCHAR(10),
            color_percentage FLOAT,
            FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
        )
    """)
    
    connection.commit()
    cursor.close()
    connection.close()

ETL Pipeline Implementation

Extract: Fetch Data from API

import requests
import os
from typing import List, Dict

def fetch_artifacts(page: int = 1, size: int = 100) -> Dict:
    """
    Fetch artifacts from Harvard Art Museums API with pagination
    
    Args:
        page: Page number (1-indexed)
        size: Number of records per page (max 100)
    
    Returns:
        JSON response containing artifact records
    """
    api_key = os.getenv('HARVARD_API_KEY')
    base_url = "https://api.harvardartmuseums.org/object"
    
    params = {
        'apikey': api_key,
        'page': page,
        'size': size,
        'hasimage': 1  # Only fetch artifacts with images
    }
    
    response = requests.get(base_url, params=params)
    response.raise_for_status()
    
    return response.json()

def extract_all_artifacts(max_records: int = 1000) -> List[Dict]:
    """
    Extract multiple pages of artifacts with rate limiting
    
    Args:
        max_records: Maximum number of records to fetch
    
    Returns:
        List of artifact dictionaries
    """
    import time
    
    artifacts = []
    page = 1
    page_size = 100
    
    while len(artifacts) < max_records:
        try:
            data = fetch_artifacts(page=page, size=page_size)
            
            if 'records' not in data or len(data['records']) == 0:
                break
            
            artifacts.extend(data['records'])
            page += 1
            
            # Rate limiting: Harvard API allows 2500 requests/day
            time.sleep(0.5)
            
            print(f"Fetched {len(artifacts)} artifacts...")
            
        except requests.exceptions.RequestException as e:
            print(f"Error fetching page {page}: {e}")
            break
    
    return artifacts[:max_records]

Transform: Clean and Structure Data

import pandas as pd

def transform_artifacts(raw_artifacts: List[Dict]) -> tuple:
    """
    Transform raw API data into structured dataframes
    
    Returns:
        Tuple of (metadata_df, media_df, colors_df)
    """
    metadata_records = []
    media_records = []
    color_records = []
    
    for artifact in raw_artifacts:
        # Extract metadata
        metadata = {
            'id': artifact.get('id'),
            'title': artifact.get('title', ''),
            'culture': artifact.get('culture', ''),
            'century': artifact.get('century', ''),
            'classification': artifact.get('classification', ''),
            'department': artifact.get('department', ''),
            'dated': artifact.get('dated', ''),
            'period': artifact.get('period', ''),
            'technique': artifact.get('technique', ''),
            'description': artifact.get('description', ''),
            'url': artifact.get('url', ''),
            'dated_begin': artifact.get('datebegin'),
            'dated_end': artifact.get('dateend')
        }
        metadata_records.append(metadata)
        
        # Extract media/images
        if 'images' in artifact and artifact['images']:
            for image in artifact['images']:
                media_records.append({
                    'artifact_id': artifact.get('id'),
                    'image_url': image.get('baseimageurl', ''),
                    'media_type': 'image'
                })
        
        # Extract color data
        if 'colors' in artifact and artifact['colors']:
            for color in artifact['colors']:
                color_records.append({
                    'artifact_id': artifact.get('id'),
                    'color_hex': color.get('hex', ''),
                    'color_percentage': color.get('percent', 0.0)
                })
    
    metadata_df = pd.DataFrame(metadata_records)
    media_df = pd.DataFrame(media_records)
    colors_df = pd.DataFrame(color_records)
    
    return metadata_df, media_df, colors_df

Load: Insert into SQL Database

def load_to_database(metadata_df: pd.DataFrame, 
                     media_df: pd.DataFrame, 
                     colors_df: pd.DataFrame):
    """
    Load transformed data into MySQL database using batch inserts
    """
    connection = mysql.connector.connect(
        host=os.getenv('DB_HOST'),
        port=os.getenv('DB_PORT'),
        user=os.getenv('DB_USER'),
        password=os.getenv('DB_PASSWORD'),
        database=os.getenv('DB_NAME')
    )
    
    cursor = connection.cursor()
    
    # Load metadata (batch insert)
    metadata_query = """
        INSERT INTO artifactmetadata 
        (id, title, culture, century, classification, department, 
         dated, period, technique, description, url, dated_begin, dated_end)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE
        title=VALUES(title), culture=VALUES(culture)
    """
    
    metadata_values = [tuple(row) for row in metadata_df.values]
    cursor.executemany(metadata_query, metadata_values)
    
    # Load media
    if not media_df.empty:
        media_query = """
            INSERT INTO artifactmedia (artifact_id, image_url, media_type)
            VALUES (%s, %s, %s)
        """
        media_values = [tuple(row) for row in media_df.values]
        cursor.executemany(media_query, media_values)
    
    # Load colors
    if not colors_df.empty:
        colors_query = """
            INSERT INTO artifactcolors (artifact_id, color_hex, color_percentage)
            VALUES (%s, %s, %s)
        """
        colors_values = [tuple(row) for row in colors_df.values]
        cursor.executemany(colors_query, colors_values)
    
    connection.commit()
    cursor.close()
    connection.close()
    
    print(f"Loaded {len(metadata_df)} artifacts to database")

Analytics Queries

Sample SQL Queries

ANALYTICS_QUERIES = {
    "artifacts_by_culture": """
        SELECT culture, COUNT(*) as artifact_count
        FROM artifactmetadata
        WHERE culture IS NOT NULL AND culture != ''
        GROUP BY culture
        ORDER BY artifact_count DESC
        LIMIT 15
    """,
    
    "artifacts_by_century": """
        SELECT century, COUNT(*) as artifact_count
        FROM artifactmetadata
        WHERE century IS NOT NULL AND century != ''
        GROUP BY century
        ORDER BY artifact_count DESC
    """,
    
    "department_distribution": """
        SELECT department, COUNT(*) as artifact_count
        FROM artifactmetadata
        WHERE department IS NOT NULL
        GROUP BY department
        ORDER BY artifact_count DESC
    """,
    
    "top_colors": """
        SELECT color_hex, COUNT(*) as usage_count,
               AVG(color_percentage) as avg_percentage
        FROM artifactcolors
        GROUP BY color_hex
        ORDER BY usage_count DESC
        LIMIT 20
    """,
    
    "media_availability": """
        SELECT 
            COUNT(DISTINCT m.artifact_id) as artifacts_with_media,
            COUNT(*) as total_media_items,
            AVG(images_per_artifact) as avg_images_per_artifact
        FROM (
            SELECT artifact_id, COUNT(*) as images_per_artifact
            FROM artifactmedia
            GROUP BY artifact_id
        ) as images_count
        JOIN artifactmedia m ON m.artifact_id = images_count.artifact_id
    """,
    
    "temporal_analysis": """
        SELECT 
            FLOOR(dated_begin / 100) * 100 as century_start,
            COUNT(*) as artifact_count
        FROM artifactmetadata
        WHERE dated_begin IS NOT NULL
        GROUP BY century_start
        ORDER BY century_start
    """
}

def execute_analytics_query(query_name: str) -> pd.DataFrame:
    """Execute a predefined analytics query and return results"""
    connection = mysql.connector.connect(
        host=os.getenv('DB_HOST'),
        database=os.getenv('DB_NAME'),
        user=os.getenv('DB_USER'),
        password=os.getenv('DB_PASSWORD')
    )
    
    query = ANALYTICS_QUERIES.get(query_name)
    df = pd.read_sql(query, connection)
    connection.close()
    
    return df

Streamlit Application

Main Application Structure

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

def main():
    st.set_page_config(
        page_title="Harvard Art Museums Analytics",
        page_icon="🏛️",
        layout="wide"
    )
    
    st.title("🏛️ Harvard Art Museums - Data Analytics Dashboard")
    st.markdown("---")
    
    # Sidebar navigation
    page = st.sidebar.selectbox(
        "Choose a page",
        ["Data Collection", "Analytics Dashboard", "Query Explorer"]
    )
    
    if page == "Data Collection":
        show_data_collection_page()
    elif page == "Analytics Dashboard":
        show_analytics_dashboard()
    else:
        show_query_explorer()

def show_data_collection_page():
    """Page for triggering ETL pipeline"""
    st.header("📥 Data Collection")
    
    num_records = st.number_input(
        "Number of artifacts to collect",
        min_value=100,
        max_value=5000,
        value=500,
        step=100
    )
    
    if st.button("Start ETL Pipeline"):
        with st.spinner("Extracting data from API..."):
            raw_data = extract_all_artifacts(max_records=num_records)
            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 records")
        
        with st.spinner("Loading into database..."):
            load_to_database(metadata_df, media_df, colors_df)
            st.success("✅ Data loaded successfully!")
        
        # Show sample data
        st.subheader("Sample Data Preview")
        st.dataframe(metadata_df.head(10))

def show_analytics_dashboard():
    """Pre-built analytics visualizations"""
    st.header("📊 Analytics Dashboard")
    
    col1, col2 = st.columns(2)
    
    with col1:
        st.subheader("Artifacts by Culture")
        df = execute_analytics_query("artifacts_by_culture")
        fig = px.bar(df, x='culture', y='artifact_count', 
                     color='artifact_count', color_continuous_scale='viridis')
        st.plotly_chart(fig, use_container_width=True)
    
    with col2:
        st.subheader("Department Distribution")
        df = execute_analytics_query("department_distribution")
        fig = px.pie(df, names='department', values='artifact_count')
        st.plotly_chart(fig, use_container_width=True)
    
    st.subheader("Temporal Distribution")
    df = execute_analytics_query("temporal_analysis")
    fig = px.line(df, x='century_start', y='artifact_count',
                  markers=True, line_shape='spline')
    st.plotly_chart(fig, use_container_width=True)
    
    st.subheader("Top Colors in Collection")
    df = execute_analytics_query("top_colors")
    fig = go.Figure(data=[go.Bar(
        x=df['color_hex'],
        y=df['usage_count'],
        marker_color=['#' + hex for hex in df['color_hex']]
    )])
    st.plotly_chart(fig, use_container_width=True)

def show_query_explorer():
    """Custom query execution interface"""
    st.header("🔍 Query Explorer")
    
    query_choice = st.selectbox(
        "Select a query",
        list(ANALYTICS_QUERIES.keys())
    )
    
    st.code(ANALYTICS_QUERIES[query_choice], language='sql')
    
    if st.button("Execute Query"):
        df = execute_analytics_query(query_choice)
        st.dataframe(df)
        
        # Auto-generate visualization if appropriate
        if len(df.columns) == 2 and df.shape[0] > 0:
            col1_name, col2_name = df.columns
            fig = px.bar(df, x=col1_name, y=col2_name)
            st.plotly_chart(fig, use_container_width=True)

if __name__ == "__main__":
    main()

Running the Application

# Start the Streamlit app
streamlit run app.py

# Access at http://localhost:8501

Common Patterns

Rate-Limited API Calls

import time
from functools import wraps

def rate_limit(calls_per_minute=120):
    """Decorator to rate limit API calls"""
    min_interval = 60.0 / calls_per_minute
    
    def decorator(func):
        last_called = [0.0]
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            wait_time = min_interval - elapsed
            
            if wait_time > 0:
                time.sleep(wait_time)
            
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        
        return wrapper
    return decorator

@rate_limit(calls_per_minute=100)
def fetch_artifact_by_id(artifact_id: int) -> Dict:
    """Fetch single artifact with rate limiting"""
    api_key = os.getenv('HARVARD_API_KEY')
    url = f"https://api.harvardartmuseums.org/object/{artifact_id}"
    response = requests.get(url, params={'apikey': api_key})
    return response.json()

Incremental Data Loading

def get_last_loaded_id() -> int:
    """Get the highest artifact ID already in database"""
    connection = mysql.connector.connect(
        host=os.getenv('DB_HOST'),
        database=os.getenv('DB_NAME'),
        user=os.getenv('DB_USER'),
        password=os.getenv('DB_PASSWORD')
    )
    
    cursor = connection.cursor()
    cursor.execute("SELECT MAX(id) FROM artifactmetadata")
    result = cursor.fetchone()[0]
    connection.close()
    
    return result if result else 0

def incremental_etl():
    """Load only new artifacts since last run"""
    last_id = get_last_loaded_id()
    
    # Fetch artifacts with ID > last_id
    new_artifacts = fetch_artifacts_after_id(last_id)
    
    if new_artifacts:
        metadata_df, media_df, colors_df = transform_artifacts(new_artifacts)
        load_to_database(metadata_df, media_df, colors_df)
        print(f"Loaded {len(new_artifacts)} new artifacts")
    else:
        print("No new artifacts to load")

Troubleshooting

API Key Issues

# Verify API key is loaded
import os
from dotenv import load_dotenv

load_dotenv()

api_key = os.getenv('HARVARD_API_KEY')
if not api_key:
    raise ValueError("HARVARD_API_KEY not found in environment variables")

# Test API connection
response = requests.get(
    "https://api.harvardartmuseums.org/object",
    params={'apikey': api_key, 'size': 1}
)
print(f"API Status: {response.status_code}")

Database Connection Errors

def test_database_connection():
    """Test database connectivity"""
    try:
        connection = mysql.connector.connect(
            host=os.getenv('DB_HOST'),
            port=int(os.getenv('DB_PORT', 3306)),
            user=os.getenv('DB_USER'),
            password=os.getenv('DB_PASSWORD'),
            database=os.getenv('DB_NAME'),
            connect_timeout=10
        )
        
        cursor = connection.cursor()
        cursor.execute("SELECT 1")
        result = cursor.fetchone()
        
        cursor.close()
        connection.close()
        
        print("✅ Database connection successful")
        return True
        
    except Error as e:
        print(f"❌ Database connection failed: {e}")
        return False

Handling Missing Data

def safe_transform(artifact: Dict) -> Dict:
    """Transform with null handling"""
    return {
        'id': artifact.get('id'),
        'title': artifact.get('title') or 'Untitled',
        'culture': artifact.get('culture') or 'Unknown',
        'century': artifact.get('century') or 'Unknown',
        'classification': artifact.get('classification') or 'Unclassified',
        'department': artifact.get('department') or 'General Collection',
        'description': (artifact.get('description') or '')[:5000],  # Truncate long text
        'dated_begin': artifact.get('datebegin') if isinstance(artifact.get('datebegin'), int) else None,
        'dated_end': artifact.get('dateend') if isinstance(artifact.get('dateend'), int) else None
    }

Memory Management for Large Datasets

def chunked_etl(total_records: int, chunk_size: int = 500):
    """Process large datasets in chunks to manage memory"""
    num_chunks = (total_records + chunk_size - 1) // chunk_size
    
    for chunk_num in range(num_chunks):
        start_idx = chunk_num * chunk_size
        end_idx = min(start_idx + chunk_size, total_records)
        
        print(f"Processing chunk {chunk_num + 1}/{num_chunks}")
        
        # Extract chunk
        artifacts = extract_all_artifacts_range(start_idx, end_idx)
        
        # Transform and load immediately
        metadata_df, media_df, colors_df = transform_artifacts(artifacts)
        load_to_database(metadata_df, media_df, colors_df)
        
        # Clear memory
        del artifacts, metadata_df, media_df, colors_df

This skill provides comprehensive guidance for building ETL pipelines and analytics applications with the Harvard Art Museums API using Python, SQL, and Streamlit.