Harvard Art Museums Data Pipeline

Skill by ara.so — Data Skills collection.

This skill enables you to build production-ready data engineering pipelines using the Harvard Art Museums API. It covers ETL workflows, relational database design, SQL analytics, and interactive Streamlit dashboards for artifact data visualization.

What This Project Does

The Harvard-Artifacts-Collection-Data-Engineering-Analytics-App demonstrates a complete data pipeline:

• Extract: Fetches artifact data from Harvard Art Museums API with pagination and rate limiting

• Transform: Converts nested JSON into normalized relational tables

• Load: Batch inserts data into MySQL/TiDB Cloud databases

• Analyze: Executes analytical SQL queries for insights

• Visualize: Renders interactive dashboards with Plotly and Streamlit

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

Installation

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

Dependencies

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_db_user

DB_PASSWORD=your_db_password

DB_NAME=harvard_artifacts

Get Harvard API Key

• Visit Harvard Art Museums API

• Register for a free API key

• Add to .env file

Database Setup

import mysql.connector

from dotenv import load_dotenv

import os

load_dotenv()

# Database connection

def get_db_connection():

    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')

    )

# Create tables

def setup_database():

    conn = get_db_connection()

    cursor = conn.cursor()

    # 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),

            technique VARCHAR(255),

            medium VARCHAR(500),

            dated VARCHAR(255),

            url TEXT,

            totalpageviews INT,

            totaluniquepageviews INT

        )

    """)

    # Artifact media table

    cursor.execute("""

        CREATE TABLE IF NOT EXISTS artifactmedia (

            id INT AUTO_INCREMENT PRIMARY KEY,

            artifact_id INT,

            iiifbaseuri VARCHAR(500),

            baseimageurl TEXT,

            primaryimageurl TEXT,

            FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)

        )

    """)

    # Artifact colors table

    cursor.execute("""

        CREATE TABLE IF NOT EXISTS artifactcolors (

            id INT AUTO_INCREMENT PRIMARY KEY,

            artifact_id INT,

            color VARCHAR(50),

            spectrum VARCHAR(50),

            hue VARCHAR(50),

            percent DECIMAL(5,2),

            FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)

        )

    """)

    conn.commit()

    cursor.close()

    conn.close()

API Integration

Basic API Request

import requests

import os

from dotenv import load_dotenv

load_dotenv()

def fetch_artifacts(page=1, size=100):

    """Fetch artifacts from Harvard Art Museums API"""

    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 artifacts with images

    }

    response = requests.get(base_url, params=params)

    response.raise_for_status()

    return response.json()

Paginated Data Collection

def collect_all_artifacts(max_records=1000):

    """Collect artifacts with pagination handling"""

    all_artifacts = []

    page = 1

    size = 100

    while len(all_artifacts) < max_records:

        try:

            data = fetch_artifacts(page=page, size=size)

            records = data.get('records', [])

            if not records:

                break

            all_artifacts.extend(records)

            # Check if more pages available

            if data['info']['next'] is None:

                break

            page += 1

            # Rate limiting

            import time

            time.sleep(0.5)

        except Exception as e:

            print(f"Error fetching page {page}: {e}")

            break

    return all_artifacts[:max_records]

ETL Pipeline

Extract and Transform

import pandas as pd

def transform_artifact_metadata(artifacts):

    """Transform artifacts into metadata DataFrame"""

    metadata = []

    for artifact in artifacts:

        metadata.append({

            'id': artifact.get('id'),

            'title': artifact.get('title', 'Unknown')[:500],

            'culture': artifact.get('culture', 'Unknown')[:255],

            'century': artifact.get('century', 'Unknown')[:100],

            'classification': artifact.get('classification', 'Unknown')[:255],

            'department': artifact.get('department', 'Unknown')[:255],

            'technique': artifact.get('technique', 'Unknown')[:255],

            'medium': artifact.get('medium', 'Unknown')[:500],

            'dated': artifact.get('dated', 'Unknown')[:255],

            'url': artifact.get('url', ''),

            'totalpageviews': artifact.get('totalpageviews', 0),

            'totaluniquepageviews': artifact.get('totaluniquepageviews', 0)

        })

    return pd.DataFrame(metadata)

def transform_artifact_media(artifacts):

    """Transform artifacts into media DataFrame"""

    media = []

    for artifact in artifacts:

        artifact_id = artifact.get('id')

        images = artifact.get('images', [])

        if images:

            primary_image = images[0]

            media.append({

                'artifact_id': artifact_id,

                'iiifbaseuri': primary_image.get('iiifbaseuri', ''),

                'baseimageurl': primary_image.get('baseimageurl', ''),

                'primaryimageurl': artifact.get('primaryimageurl', '')

            })

    return pd.DataFrame(media)

def transform_artifact_colors(artifacts):

    """Transform artifacts into colors DataFrame"""

    colors = []

    for artifact in artifacts:

        artifact_id = artifact.get('id')

        color_list = artifact.get('colors', [])

        for color in color_list:

            colors.append({

                'artifact_id': artifact_id,

                'color': color.get('color', ''),

                'spectrum': color.get('spectrum', ''),

                'hue': color.get('hue', ''),

                'percent': color.get('percent', 0.0)

            })

    return pd.DataFrame(colors)

Load into Database

def load_metadata(df, conn):

    """Batch insert metadata into database"""

    cursor = conn.cursor()

    insert_query = """

        INSERT INTO artifactmetadata

        (id, title, culture, century, classification, department,

         technique, medium, dated, url, totalpageviews, totaluniquepageviews)

        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)

        ON DUPLICATE KEY UPDATE

        title=VALUES(title), culture=VALUES(culture)

    """

    data = [tuple(row) for row in df.values]

    cursor.executemany(insert_query, data)

    conn.commit()

    cursor.close()

def run_etl_pipeline(max_records=1000):

    """Execute complete ETL pipeline"""

    # Extract

    print("Extracting artifacts from API...")

    artifacts = collect_all_artifacts(max_records)

    # Transform

    print("Transforming data...")

    df_metadata = transform_artifact_metadata(artifacts)

    df_media = transform_artifact_media(artifacts)

    df_colors = transform_artifact_colors(artifacts)

    # Load

    print("Loading into database...")

    conn = get_db_connection()

    load_metadata(df_metadata, conn)

    load_metadata(df_media, conn)  # Similar function for media

    load_metadata(df_colors, conn)  # Similar function for colors

    conn.close()

    print(f"ETL complete: {len(artifacts)} artifacts processed")

SQL Analytics

Sample Analytical Queries

# Query 1: Artifacts by culture

QUERY_BY_CULTURE = """

    SELECT culture, COUNT(*) as count

    FROM artifactmetadata

    WHERE culture != 'Unknown'

    GROUP BY culture

    ORDER BY count DESC

    LIMIT 20

"""

# Query 2: Most viewed artifacts

QUERY_TOP_VIEWED = """

    SELECT title, culture, totalpageviews

    FROM artifactmetadata

    ORDER BY totalpageviews DESC

    LIMIT 10

"""

# Query 3: Artifacts by century

QUERY_BY_CENTURY = """

    SELECT century, COUNT(*) as count

    FROM artifactmetadata

    WHERE century != 'Unknown'

    GROUP BY century

    ORDER BY count DESC

"""

# Query 4: Color distribution

QUERY_COLOR_DISTRIBUTION = """

    SELECT color, COUNT(*) as count, AVG(percent) as avg_percent

    FROM artifactcolors

    GROUP BY color

    ORDER BY count DESC

    LIMIT 15

"""

# Query 5: Department statistics

QUERY_DEPARTMENT_STATS = """

    SELECT

        department,

        COUNT(*) as total_artifacts,

        AVG(totalpageviews) as avg_views

    FROM artifactmetadata

    WHERE department != 'Unknown'

    GROUP BY department

    ORDER BY total_artifacts DESC

"""

def execute_query(query):

    """Execute SQL query and return DataFrame"""

    conn = get_db_connection()

    df = pd.read_sql(query, conn)

    conn.close()

    return df

Streamlit Dashboard

Main Application

import streamlit as st

import plotly.express as px

def main():

    st.set_page_config(

        page_title="Harvard Artifacts Analytics",

        page_icon="🏛️",

        layout="wide"

    )

    st.title("🏛️ Harvard Art Museums Analytics Dashboard")

    # Sidebar

    st.sidebar.header("Navigation")

    page = st.sidebar.selectbox(

        "Choose a page",

        ["Data Collection", "SQL Analytics", "Visualizations"]

    )

    if page == "Data Collection":

        show_data_collection()

    elif page == "SQL Analytics":

        show_sql_analytics()

    elif page == "Visualizations":

        show_visualizations()

def show_data_collection():

    st.header("📥 Data Collection")

    num_records = st.number_input(

        "Number of records to collect",

        min_value=100,

        max_value=10000,

        value=1000,

        step=100

    )

    if st.button("Start ETL Pipeline"):

        with st.spinner("Running ETL pipeline..."):

            run_etl_pipeline(num_records)

        st.success(f"Successfully collected {num_records} artifacts!")

def show_sql_analytics():

    st.header("📊 SQL Analytics")

    queries = {

        "Artifacts by Culture": QUERY_BY_CULTURE,

        "Most Viewed Artifacts": QUERY_TOP_VIEWED,

        "Artifacts by Century": QUERY_BY_CENTURY,

        "Color Distribution": QUERY_COLOR_DISTRIBUTION,

        "Department Statistics": QUERY_DEPARTMENT_STATS

    }

    selected_query = st.selectbox("Select Query", list(queries.keys()))

    if st.button("Execute Query"):

        df = execute_query(queries[selected_query])

        st.dataframe(df, use_container_width=True)

        # 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)

if __name__ == "__main__":

    main()

Run the Dashboard

streamlit run app.py

Common Patterns

Rate Limiting API Requests

import time

from functools import wraps

def rate_limit(calls_per_second=2):

    """Decorator to rate limit API calls"""

    min_interval = 1.0 / calls_per_second

    last_called = [0.0]

    def decorator(func):

        @wraps(func)

        def wrapper(*args, **kwargs):

            elapsed = time.time() - last_called[0]

            left_to_wait = min_interval - elapsed

            if left_to_wait > 0:

                time.sleep(left_to_wait)

            ret = func(*args, **kwargs)

            last_called[0] = time.time()

            return ret

        return wrapper

    return decorator

@rate_limit(calls_per_second=2)

def fetch_artifact_by_id(artifact_id):

    """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()

Error Handling and Retries

from requests.exceptions import RequestException

def fetch_with_retry(url, params, max_retries=3):

    """Fetch with exponential backoff retry"""

    for attempt in range(max_retries):

        try:

            response = requests.get(url, params=params, timeout=10)

            response.raise_for_status()

            return response.json()

        except RequestException as e:

            if attempt == max_retries - 1:

                raise

            wait_time = 2 ** attempt

            print(f"Retry {attempt + 1}/{max_retries} after {wait_time}s")

            time.sleep(wait_time)

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:

        conn = get_db_connection()

        cursor = conn.cursor()

        cursor.execute("SELECT 1")

        result = cursor.fetchone()

        cursor.close()

        conn.close()

        print("Database connection successful")

        return True

    except Exception as e:

        print(f"Database connection failed: {e}")

        return False

Memory Issues with Large Datasets

def batch_process_artifacts(total_records, batch_size=100):

    """Process artifacts in batches to manage memory"""

    num_batches = (total_records + batch_size - 1) // batch_size

    for batch_num in range(num_batches):

        page = batch_num + 1

        artifacts = fetch_artifacts(page=page, size=batch_size)

        # Process batch

        df = transform_artifact_metadata(artifacts['records'])

        conn = get_db_connection()

        load_metadata(df, conn)

        conn.close()

        print(f"Processed batch {batch_num + 1}/{num_batches}")

Streamlit Caching for Performance

@st.cache_data(ttl=3600)

def cached_query_execution(query):

    """Cache query results for 1 hour"""

    return execute_query(query)

# Use in dashboard

df = cached_query_execution(QUERY_BY_CULTURE)

st.dataframe(df)