harvard-artifacts-data-engineering-analytics

name	harvard-artifacts-data-engineering-analytics
description	Build end-to-end ETL pipelines and analytics dashboards using the Harvard Art Museums API with Python, SQL, and Streamlit
triggers	["how do I build a data pipeline with Harvard Art Museums API","create an ETL pipeline for museum artifacts data","set up analytics dashboard for Harvard art collection","extract and transform Harvard museum data into SQL","build Streamlit app for art artifacts analytics","visualize Harvard Art Museums data with SQL queries","implement artifact collection data engineering pipeline","analyze museum artifact metadata with Python and SQL"]

Harvard Artifacts Data Engineering Analytics

Skill by ara.so — Data Skills collection

This project provides an end-to-end data engineering and analytics application for the Harvard Art Museums API. It demonstrates real-world ETL pipelines, SQL database design, analytical queries, and interactive visualization using Streamlit.

What This Project Does

The application implements a complete data pipeline:

• Extract: Fetches artifact data from Harvard Art Museums API with pagination and rate limiting
• Transform: Processes nested JSON into relational database tables (metadata, media, colors)
• Load: Batch inserts transformed data into MySQL/TiDB Cloud
• Analyze: Executes 20+ predefined SQL queries for insights
• Visualize: Renders interactive dashboards with Plotly charts in Streamlit

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

# Set up environment variables
export HARVARD_API_KEY="your_api_key_here"
export DB_HOST="your_database_host"
export DB_USER="your_database_user"
export DB_PASSWORD="your_database_password"
export DB_NAME="harvard_artifacts"

Configuration

API Key Setup

Obtain an API key from Harvard Art Museums API:

import os
import requests

API_KEY = os.getenv('HARVARD_API_KEY')
BASE_URL = "https://api.harvardartmuseums.org/object"

# Test API connection
response = requests.get(f"{BASE_URL}?apikey={API_KEY}&size=1")
if response.status_code == 200:
    print("API connection successful")

Database Configuration

import mysql.connector
import os

db_config = {
    'host': os.getenv('DB_HOST'),
    'user': os.getenv('DB_USER'),
    'password': os.getenv('DB_PASSWORD'),
    'database': os.getenv('DB_NAME'),
    'port': int(os.getenv('DB_PORT', 3306))
}

conn = mysql.connector.connect(**db_config)
cursor = conn.cursor()

Database Schema

The project uses three main tables:

-- Artifact metadata table
CREATE TABLE artifactmetadata (
    id INT PRIMARY KEY,
    title VARCHAR(500),
    culture VARCHAR(255),
    century VARCHAR(100),
    classification VARCHAR(255),
    department VARCHAR(255),
    dated VARCHAR(255),
    technique VARCHAR(500),
    medium VARCHAR(500),
    period VARCHAR(255),
    provenance TEXT,
    creditline TEXT,
    accession_number VARCHAR(255),
    division VARCHAR(255)
);

-- Artifact media table
CREATE TABLE artifactmedia (
    artifact_id INT,
    baseimageurl VARCHAR(500),
    primaryimageurl VARCHAR(500),
    has_image BOOLEAN,
    total_images INT,
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);

-- Artifact colors table
CREATE TABLE artifactcolors (
    artifact_id INT,
    color VARCHAR(50),
    percentage FLOAT,
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);

ETL Pipeline Implementation

Extract Phase

import requests
import time

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

def extract_all_artifacts(api_key, max_pages=10):
    """Extract multiple pages of artifact data"""
    all_artifacts = []
    
    for page in range(1, max_pages + 1):
        data = fetch_artifacts(api_key, page=page)
        artifacts = data.get('records', [])
        all_artifacts.extend(artifacts)
        
        if not artifacts:
            break
    
    return all_artifacts

Transform Phase

import pandas as pd

def transform_artifacts(raw_artifacts):
    """Transform raw API data into structured dataframes"""
    metadata_list = []
    media_list = []
    colors_list = []
    
    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'),
            'technique': artifact.get('technique'),
            'medium': artifact.get('medium'),
            'period': artifact.get('period'),
            'provenance': artifact.get('provenance'),
            'creditline': artifact.get('creditline'),
            'accession_number': artifact.get('accessionyear'),
            'division': artifact.get('division')
        }
        metadata_list.append(metadata)
        
        # Extract media information
        media = {
            'artifact_id': artifact.get('id'),
            'baseimageurl': artifact.get('baseimageurl'),
            'primaryimageurl': artifact.get('primaryimageurl'),
            'has_image': 1 if artifact.get('primaryimageurl') else 0,
            'total_images': artifact.get('totalpageviews', 0)
        }
        media_list.append(media)
        
        # Extract color data
        colors = artifact.get('colors', [])
        for color in colors:
            color_data = {
                'artifact_id': artifact.get('id'),
                'color': color.get('color'),
                'percentage': color.get('percent')
            }
            colors_list.append(color_data)
    
    return (
        pd.DataFrame(metadata_list),
        pd.DataFrame(media_list),
        pd.DataFrame(colors_list)
    )

Load Phase

def load_to_database(metadata_df, media_df, colors_df, db_config):
    """Batch insert dataframes into MySQL database"""
    import mysql.connector
    
    conn = mysql.connector.connect(**db_config)
    cursor = conn.cursor()
    
    # Load metadata
    for _, row in metadata_df.iterrows():
        query = """
        INSERT INTO artifactmetadata 
        (id, title, culture, century, classification, department, dated, 
         technique, medium, period, provenance, creditline, accession_number, division)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE title=VALUES(title)
        """
        cursor.execute(query, tuple(row))
    
    # Load media
    for _, row in media_df.iterrows():
        query = """
        INSERT INTO artifactmedia 
        (artifact_id, baseimageurl, primaryimageurl, has_image, total_images)
        VALUES (%s, %s, %s, %s, %s)
        """
        cursor.execute(query, tuple(row))
    
    # Load colors
    for _, row in colors_df.iterrows():
        query = """
        INSERT INTO artifactcolors 
        (artifact_id, color, percentage)
        VALUES (%s, %s, %s)
        """
        cursor.execute(query, tuple(row))
    
    conn.commit()
    cursor.close()
    conn.close()

Streamlit Application

Main App Structure

import streamlit as st
import pandas as pd
import plotly.express as px
import mysql.connector
import os

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

# Sidebar configuration
st.sidebar.title("Harvard Art Museums Analytics")
st.sidebar.markdown("### Configuration")

# Database connection
@st.cache_resource
def get_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')
    )

def execute_query(query):
    """Execute SQL query and return results as DataFrame"""
    conn = get_database_connection()
    df = pd.read_sql(query, conn)
    return df

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

# ETL Pipeline Section
if st.sidebar.button("Run ETL Pipeline"):
    with st.spinner("Fetching data from API..."):
        api_key = os.getenv('HARVARD_API_KEY')
        artifacts = extract_all_artifacts(api_key, max_pages=5)
        st.success(f"Extracted {len(artifacts)} artifacts")
    
    with st.spinner("Transforming data..."):
        metadata_df, media_df, colors_df = transform_artifacts(artifacts)
        st.success("Data transformation complete")
    
    with st.spinner("Loading to database..."):
        db_config = {
            'host': os.getenv('DB_HOST'),
            'user': os.getenv('DB_USER'),
            'password': os.getenv('DB_PASSWORD'),
            'database': os.getenv('DB_NAME')
        }
        load_to_database(metadata_df, media_df, colors_df, db_config)
        st.success("Data loaded successfully!")

Analytics Queries

# Predefined analytical queries
QUERIES = {
    "Artifacts by Culture": """
        SELECT culture, COUNT(*) as artifact_count
        FROM artifactmetadata
        WHERE culture IS NOT NULL
        GROUP BY culture
        ORDER BY artifact_count DESC
        LIMIT 15
    """,
    
    "Artifacts by Century": """
        SELECT century, COUNT(*) as count
        FROM artifactmetadata
        WHERE century IS NOT NULL
        GROUP BY century
        ORDER BY count DESC
    """,
    
    "Top Colors Used": """
        SELECT color, COUNT(*) as frequency, AVG(percentage) as avg_percentage
        FROM artifactcolors
        GROUP BY color
        ORDER BY frequency DESC
        LIMIT 10
    """,
    
    "Media Availability": """
        SELECT 
            has_image,
            COUNT(*) as count,
            ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 2) as percentage
        FROM artifactmedia
        GROUP BY has_image
    """,
    
    "Artifacts by Department": """
        SELECT department, COUNT(*) as total_artifacts
        FROM artifactmetadata
        WHERE department IS NOT NULL
        GROUP BY department
        ORDER BY total_artifacts DESC
    """
}

# Query selector
query_name = st.selectbox("Select Analysis", list(QUERIES.keys()))

if st.button("Run Query"):
    df = execute_query(QUERIES[query_name])
    
    # Display 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=query_name)
        st.plotly_chart(fig, use_container_width=True)

Common Patterns

Pattern 1: Incremental Data Loading

def get_latest_artifact_id(cursor):
    """Get the most recent artifact ID in database"""
    cursor.execute("SELECT MAX(id) FROM artifactmetadata")
    result = cursor.fetchone()
    return result[0] if result[0] else 0

def incremental_etl(api_key, db_config):
    """Load only new artifacts since last ETL run"""
    conn = mysql.connector.connect(**db_config)
    cursor = conn.cursor()
    
    latest_id = get_latest_artifact_id(cursor)
    
    # Fetch only newer artifacts
    artifacts = fetch_artifacts(api_key, size=100)
    new_artifacts = [a for a in artifacts if a.get('id', 0) > latest_id]
    
    if new_artifacts:
        metadata_df, media_df, colors_df = transform_artifacts(new_artifacts)
        load_to_database(metadata_df, media_df, colors_df, db_config)
    
    cursor.close()
    conn.close()

Pattern 2: Error Handling and Logging

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def safe_fetch_artifacts(api_key, page=1, max_retries=3):
    """Fetch with retry logic"""
    for attempt in range(max_retries):
        try:
            data = fetch_artifacts(api_key, page)
            logger.info(f"Successfully fetched page {page}")
            return data
        except requests.RequestException as e:
            logger.error(f"Attempt {attempt + 1} failed: {e}")
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff

Running the Application

# Start the Streamlit app
streamlit run app.py

# Access the dashboard at http://localhost:8501

Troubleshooting

API Rate Limiting: If you encounter 429 errors, increase the sleep time between requests:

time.sleep(1)  # Increase from 0.5 to 1 second

Database Connection Issues: Verify environment variables are set:

import os
print(f"DB Host: {os.getenv('DB_HOST')}")
print(f"DB User: {os.getenv('DB_USER')}")

Missing Data Fields: Handle None values in transformations:

metadata = {
    'title': artifact.get('title', 'Unknown'),
    'culture': artifact.get('culture') or 'Unknown'
}

Memory Issues with Large Datasets: Use batch processing:

BATCH_SIZE = 1000
for i in range(0, len(metadata_df), BATCH_SIZE):
    batch = metadata_df.iloc[i:i+BATCH_SIZE]
    load_batch(batch, db_config)