harvard-artifacts-etl-analytics

name	harvard-artifacts-etl-analytics
description	Build ETL pipelines and analytics dashboards for Harvard Art Museums API data with Python, SQL, and Streamlit
triggers	["how do I extract data from Harvard Art Museums API","create an ETL pipeline for museum artifact data","build a Streamlit dashboard for art collection analytics","query Harvard artifacts database with SQL","visualize museum collection data with Plotly","set up data engineering pipeline for art museum data","transform Harvard API JSON into relational database","analyze artifact metadata by culture and century"]

Harvard Artifacts ETL Analytics Skill

Skill by ara.so — Data Skills collection.

This skill enables AI coding agents to help developers build end-to-end ETL pipelines and analytics applications using the Harvard Art Museums API. The project demonstrates real-world data engineering patterns including API integration, data transformation, SQL database design, and interactive visualization with Streamlit.

What This Project Does

The Harvard Artifacts Collection application:

• Extracts artifact data from Harvard Art Museums API with pagination and rate limiting
• Transforms nested JSON into normalized relational tables
• Loads data into MySQL/TiDB Cloud databases
• Provides 20+ predefined analytical SQL queries
• Visualizes results with interactive Plotly charts in Streamlit

Architecture Flow: 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:

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 Connection
DB_HOST=your_database_host
DB_PORT=3306
DB_USER=your_db_user
DB_PASSWORD=your_db_password
DB_NAME=harvard_artifacts

Get Harvard API Key

Register at: https://www.harvardartmuseums.org/collections/api

Database Setup

-- Create database
CREATE DATABASE harvard_artifacts;
USE harvard_artifacts;

-- Artifact metadata table
CREATE TABLE artifactmetadata (
    id INT PRIMARY KEY,
    title VARCHAR(500),
    culture VARCHAR(200),
    century VARCHAR(100),
    classification VARCHAR(200),
    department VARCHAR(200),
    dated VARCHAR(200),
    division VARCHAR(200),
    medium VARCHAR(500),
    technique VARCHAR(500),
    period VARCHAR(200),
    accessionyear INT,
    totalpageviews INT,
    totaluniquepageviews INT
);

-- Artifact media table
CREATE TABLE artifactmedia (
    id INT AUTO_INCREMENT PRIMARY KEY,
    artifact_id INT,
    iiifbaseuri VARCHAR(500),
    baseimageurl VARCHAR(500),
    primaryimageurl VARCHAR(500),
    imagecopyright TEXT,
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);

-- Artifact colors table
CREATE TABLE artifactcolors (
    id INT AUTO_INCREMENT PRIMARY KEY,
    artifact_id INT,
    color VARCHAR(50),
    spectrum VARCHAR(50),
    hue VARCHAR(50),
    percent FLOAT,
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);

Running the Application

streamlit run app.py

The Streamlit dashboard will open at http://localhost:8501

Key Code Patterns

API Data Extraction

import requests
import os
from dotenv import load_dotenv

load_dotenv()

def fetch_artifacts(api_key, page=1, size=100):
    """
    Fetch artifacts from Harvard Art Museums API with pagination
    """
    url = "https://api.harvardartmuseums.org/object"
    params = {
        "apikey": api_key,
        "page": page,
        "size": size,
        "hasimage": 1  # Only artifacts with images
    }
    
    response = requests.get(url, params=params)
    response.raise_for_status()
    return response.json()

# Usage
api_key = os.getenv("HARVARD_API_KEY")
data = fetch_artifacts(api_key, page=1, size=100)
print(f"Total records: {data['info']['totalrecords']}")
print(f"Total pages: {data['info']['pages']}")

ETL Pipeline with Pagination

import pandas as pd
import time

def extract_all_artifacts(api_key, max_pages=10):
    """
    Extract artifacts across multiple pages with rate limiting
    """
    all_artifacts = []
    
    for page in range(1, max_pages + 1):
        try:
            data = fetch_artifacts(api_key, page=page)
            artifacts = data.get('records', [])
            all_artifacts.extend(artifacts)
            
            print(f"Extracted page {page}/{max_pages}: {len(artifacts)} artifacts")
            time.sleep(0.5)  # Rate limiting
            
        except Exception as e:
            print(f"Error on page {page}: {e}")
            break
    
    return all_artifacts

def transform_artifacts(raw_artifacts):
    """
    Transform nested JSON into structured DataFrames
    """
    metadata_list = []
    media_list = []
    colors_list = []
    
    for artifact in raw_artifacts:
        # Metadata
        metadata_list.append({
            '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'),
            'division': artifact.get('division'),
            'medium': artifact.get('medium'),
            'technique': artifact.get('technique'),
            'period': artifact.get('period'),
            'accessionyear': artifact.get('accessionyear'),
            'totalpageviews': artifact.get('totalpageviews', 0),
            'totaluniquepageviews': artifact.get('totaluniquepageviews', 0)
        })
        
        # Media
        if artifact.get('primaryimageurl'):
            media_list.append({
                'artifact_id': artifact.get('id'),
                'iiifbaseuri': artifact.get('iiifbaseuri'),
                'baseimageurl': artifact.get('baseimageurl'),
                'primaryimageurl': artifact.get('primaryimageurl'),
                'imagecopyright': artifact.get('imagecopyright')
            })
        
        # Colors
        for color in artifact.get('colors', []):
            colors_list.append({
                'artifact_id': artifact.get('id'),
                'color': color.get('color'),
                'spectrum': color.get('spectrum'),
                'hue': color.get('hue'),
                'percent': color.get('percent')
            })
    
    return (
        pd.DataFrame(metadata_list),
        pd.DataFrame(media_list),
        pd.DataFrame(colors_list)
    )

Database Loading

import mysql.connector
from mysql.connector import Error

def get_db_connection():
    """
    Create MySQL database connection from environment variables
    """
    return 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')
    )

def load_metadata(df_metadata):
    """
    Batch insert artifact metadata into database
    """
    conn = get_db_connection()
    cursor = conn.cursor()
    
    insert_query = """
    INSERT INTO artifactmetadata 
    (id, title, culture, century, classification, department, dated, 
     division, medium, technique, period, accessionyear, 
     totalpageviews, totaluniquepageviews)
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    ON DUPLICATE KEY UPDATE
    title=VALUES(title), culture=VALUES(culture)
    """
    
    records = df_metadata.to_records(index=False).tolist()
    cursor.executemany(insert_query, records)
    conn.commit()
    
    print(f"Inserted {cursor.rowcount} metadata records")
    cursor.close()
    conn.close()

def load_all_data(df_metadata, df_media, df_colors):
    """
    Load all transformed data into respective tables
    """
    load_metadata(df_metadata)
    # Similar functions for media and colors
    print("ETL pipeline completed successfully")

Streamlit Analytics Dashboard

import streamlit as st
import plotly.express as px

st.set_page_config(page_title="Harvard Artifacts Analytics", layout="wide")

st.title("🎨 Harvard Art Museums Collection Analytics")

# Sidebar for query selection
query_options = {
    "Top 10 Cultures by Artifact Count": """
        SELECT culture, COUNT(*) as count
        FROM artifactmetadata
        WHERE culture IS NOT NULL
        GROUP BY culture
        ORDER BY count DESC
        LIMIT 10
    """,
    "Artifacts by Century": """
        SELECT century, COUNT(*) as count
        FROM artifactmetadata
        WHERE century IS NOT NULL
        GROUP BY century
        ORDER BY count DESC
    """,
    "Most Common Colors": """
        SELECT color, COUNT(*) as usage_count, AVG(percent) as avg_percent
        FROM artifactcolors
        GROUP BY color
        ORDER BY usage_count DESC
        LIMIT 10
    """,
    "Department Distribution": """
        SELECT department, COUNT(*) as artifact_count
        FROM artifactmetadata
        WHERE department IS NOT NULL
        GROUP BY department
        ORDER BY artifact_count DESC
    """
}

selected_query = st.sidebar.selectbox("Select Analysis", list(query_options.keys()))

if st.button("Run Analysis"):
    conn = get_db_connection()
    df_result = pd.read_sql(query_options[selected_query], conn)
    conn.close()
    
    st.subheader(f"Results: {selected_query}")
    st.dataframe(df_result)
    
    # Auto-generate visualization
    if len(df_result.columns) >= 2:
        fig = px.bar(df_result, 
                     x=df_result.columns[0], 
                     y=df_result.columns[1],
                     title=selected_query)
        st.plotly_chart(fig, use_container_width=True)

Complete ETL Workflow

def run_full_etl_pipeline():
    """
    Complete ETL pipeline from API to database
    """
    # Extract
    print("Starting extraction...")
    api_key = os.getenv("HARVARD_API_KEY")
    raw_artifacts = extract_all_artifacts(api_key, max_pages=5)
    
    # Transform
    print("Transforming data...")
    df_metadata, df_media, df_colors = transform_artifacts(raw_artifacts)
    
    # Load
    print("Loading to database...")
    load_all_data(df_metadata, df_media, df_colors)
    
    print(f"Pipeline complete! Processed {len(raw_artifacts)} artifacts")
    return df_metadata, df_media, df_colors

# Execute pipeline
if __name__ == "__main__":
    run_full_etl_pipeline()

Common Analytical Queries

-- Top viewed artifacts
SELECT title, culture, totalpageviews
FROM artifactmetadata
ORDER BY totalpageviews DESC
LIMIT 20;

-- Artifacts with color data
SELECT a.title, a.culture, c.color, c.percent
FROM artifactmetadata a
JOIN artifactcolors c ON a.id = c.artifact_id
WHERE c.percent > 50
ORDER BY c.percent DESC;

-- Media availability by department
SELECT 
    a.department,
    COUNT(*) as total_artifacts,
    COUNT(m.id) as with_media,
    ROUND(COUNT(m.id) * 100.0 / COUNT(*), 2) as media_percentage
FROM artifactmetadata a
LEFT JOIN artifactmedia m ON a.id = m.artifact_id
GROUP BY a.department
ORDER BY total_artifacts DESC;

Troubleshooting

API Rate Limits:

• Add time.sleep(0.5) between requests
• Implement exponential backoff for 429 errors

Database Connection Issues:

try:
    conn = get_db_connection()
    conn.ping(reconnect=True, attempts=3, delay=5)
except Error as e:
    print(f"Database error: {e}")

Missing Data Fields:

# Safe field access
culture = artifact.get('culture', 'Unknown')
colors = artifact.get('colors', [])

Streamlit Caching:

@st.cache_data(ttl=3600)
def load_cached_data(query):
    conn = get_db_connection()
    df = pd.read_sql(query, conn)
    conn.close()
    return df

This skill provides everything needed to build production-ready ETL pipelines and analytics dashboards for museum collection data using modern Python data engineering tools.