Harvard Art Museum Data Pipeline

Skill by ara.so — Data Skills collection.

This project provides an end-to-end data engineering solution for collecting, transforming, storing, and analyzing artifact data from the Harvard Art Museums API. It demonstrates production-ready ETL pipelines, SQL analytics, and interactive visualization using Streamlit.

What It Does

The Harvard Art Museum Data Pipeline:

• Extracts artifact data from the 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 batch inserts for performance

• Analyzes data using predefined SQL queries for business insights

• Visualizes results through interactive Streamlit dashboards with Plotly charts

Architecture

Harvard Art Museums API → Python ETL → MySQL/TiDB → SQL Analytics → Streamlit Dashboard

Key Components:

• API integration with secure key management

• Three-table relational schema: artifactmetadata, artifactmedia, artifactcolors

• 20+ analytical SQL queries

• Real-time interactive visualizations

Installation

Prerequisites

• Python 3.8+

• MySQL or TiDB Cloud account

• Harvard Art Museums API key (obtain from https://harvardartmuseums.org/collections/api)

Setup

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

Required Dependencies

streamlit

pandas

requests

mysql-connector-python

plotly

python-dotenv

Configuration

Database Connection

Configure your MySQL/TiDB connection in your application:

import mysql.connector

import os

def get_db_connection():

    """Create database connection using environment variables"""

    return mysql.connector.connect(

        host=os.getenv('DB_HOST'),

        user=os.getenv('DB_USER'),

        password=os.getenv('DB_PASSWORD'),

        database=os.getenv('DB_NAME')

    )

API Configuration

Set up the Harvard Art Museums API client:

import os

import requests

class HarvardMuseumAPI:

    def __init__(self):

        self.api_key = os.getenv('HARVARD_API_KEY')

        self.base_url = "https://api.harvardartmuseums.org"

    def get_objects(self, page=1, size=100):

        """Fetch objects with pagination"""

        url = f"{self.base_url}/object"

        params = {

            'apikey': self.api_key,

            'page': page,

            'size': size

        }

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

        response.raise_for_status()

        return response.json()

Database Schema

Create Tables

-- Artifact Metadata

CREATE TABLE artifactmetadata (

    id INT PRIMARY KEY,

    title VARCHAR(500),

    culture VARCHAR(255),

    century VARCHAR(100),

    classification VARCHAR(255),

    department VARCHAR(255),

    technique VARCHAR(500),

    medium VARCHAR(500),

    dated VARCHAR(255),

    accession_number VARCHAR(100),

    url VARCHAR(500)

);

-- Artifact Media

CREATE TABLE artifactmedia (

    media_id INT AUTO_INCREMENT PRIMARY KEY,

    artifact_id INT,

    image_url VARCHAR(1000),

    alt_text TEXT,

    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)

);

-- Artifact Colors

CREATE TABLE artifactcolors (

    color_id INT AUTO_INCREMENT PRIMARY KEY,

    artifact_id INT,

    color_name VARCHAR(100),

    hex_code VARCHAR(10),

    percentage DECIMAL(5,2),

    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)

);

ETL Pipeline

Extract Phase

import requests

import time

def extract_artifacts(api_key, num_pages=10, page_size=100):

    """

    Extract artifacts from Harvard Art Museums API with pagination

    """

    base_url = "https://api.harvardartmuseums.org/object"

    all_records = []

    for page in range(1, num_pages + 1):

        params = {

            'apikey': api_key,

            'page': page,

            'size': page_size

        }

        try:

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

            response.raise_for_status()

            data = response.json()

            if 'records' in data:

                all_records.extend(data['records'])

            # Rate limiting

            time.sleep(0.5)

        except requests.exceptions.RequestException as e:

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

            continue

    return all_records

Transform Phase

import pandas as pd

def transform_metadata(records):

    """Transform artifact records into metadata DataFrame"""

    metadata = []

    for record in records:

        metadata.append({

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

            'title': record.get('title', '')[:500],

            'culture': record.get('culture', '')[:255],

            'century': record.get('century', '')[:100],

            'classification': record.get('classification', '')[:255],

            'department': record.get('department', '')[:255],

            'technique': record.get('technique', '')[:500],

            'medium': record.get('medium', '')[:500],

            'dated': record.get('dated', '')[:255],

            'accession_number': record.get('accessionyear', '')[:100],

            'url': record.get('url', '')[:500]

        })

    return pd.DataFrame(metadata)

def transform_media(records):

    """Extract media information from artifacts"""

    media = []

    for record in records:

        artifact_id = record.get('id')

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

        for image in images:

            media.append({

                'artifact_id': artifact_id,

                'image_url': image.get('baseimageurl', '')[:1000],

                'alt_text': image.get('alttext', '')

            })

    return pd.DataFrame(media)

def transform_colors(records):

    """Extract color data from artifacts"""

    colors = []

    for record in records:

        artifact_id = record.get('id')

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

        for color in color_list:

            colors.append({

                'artifact_id': artifact_id,

                'color_name': color.get('color', '')[:100],

                'hex_code': color.get('hex', '')[:10],

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

            })

    return pd.DataFrame(colors)

Load Phase

def load_to_database(df, table_name, connection):

    """

    Batch insert DataFrame into MySQL table

    """

    cursor = connection.cursor()

    # Generate INSERT statement

    columns = ', '.join(df.columns)

    placeholders = ', '.join(['%s'] * len(df.columns))

    insert_query = f"INSERT IGNORE INTO {table_name} ({columns}) VALUES ({placeholders})"

    # Batch insert

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

    cursor.executemany(insert_query, data_tuples)

    connection.commit()

    print(f"Inserted {cursor.rowcount} rows into {table_name}")

    cursor.close()

Complete ETL Workflow

def run_etl_pipeline(api_key, db_connection, num_pages=10):

    """

    Execute complete ETL pipeline

    """

    # Extract

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

    records = extract_artifacts(api_key, num_pages=num_pages)

    # Transform

    print("Transforming data...")

    metadata_df = transform_metadata(records)

    media_df = transform_media(records)

    colors_df = transform_colors(records)

    # Load

    print("Loading data to database...")

    load_to_database(metadata_df, 'artifactmetadata', db_connection)

    load_to_database(media_df, 'artifactmedia', db_connection)

    load_to_database(colors_df, 'artifactcolors', db_connection)

    print("ETL pipeline completed successfully!")

SQL Analytics Queries

Common Analytics Patterns

# Top cultures by artifact count

query_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_centuries = """

SELECT century, COUNT(*) as count

FROM artifactmetadata

WHERE century IS NOT NULL

GROUP BY century

ORDER BY count DESC;

"""

# Media availability analysis

query_media = """

SELECT

    CASE

        WHEN EXISTS (SELECT 1 FROM artifactmedia WHERE artifactmedia.artifact_id = artifactmetadata.id)

        THEN 'Has Media'

        ELSE 'No Media'

    END as media_status,

    COUNT(*) as count

FROM artifactmetadata

GROUP BY media_status;

"""

# Color distribution

query_colors = """

SELECT color_name, COUNT(*) as usage_count, AVG(percentage) as avg_percentage

FROM artifactcolors

GROUP BY color_name

ORDER BY usage_count DESC

LIMIT 15;

"""

# Department breakdown

query_departments = """

SELECT department, COUNT(*) as artifact_count

FROM artifactmetadata

WHERE department IS NOT NULL

GROUP BY department

ORDER BY artifact_count DESC;

"""

Streamlit Dashboard

Main Application Structure

import streamlit as st

import pandas as pd

import plotly.express as px

def main():

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

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

    st.markdown("---")

    # Sidebar configuration

    with st.sidebar:

        st.header("Configuration")

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

            run_etl_with_progress()

    # Main content tabs

    tab1, tab2, tab3 = st.tabs(["📊 Analytics", "🗄️ Data Explorer", "📈 Visualizations"])

    with tab1:

        show_analytics_dashboard()

    with tab2:

        show_data_explorer()

    with tab3:

        show_visualizations()

def run_etl_with_progress():

    """Run ETL with progress bar"""

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

        connection = get_db_connection()

        api_key = os.getenv('HARVARD_API_KEY')

        run_etl_pipeline(api_key, connection, num_pages=5)

        st.success("ETL completed successfully!")

def show_analytics_dashboard():

    """Display predefined analytics queries"""

    st.subheader("Analytical Insights")

    queries = {

        "Top Cultures": query_cultures,

        "Artifacts by Century": query_centuries,

        "Media Availability": query_media,

        "Color Distribution": query_colors,

        "Department Breakdown": query_departments

    }

    query_choice = st.selectbox("Select Analysis", list(queries.keys()))

    if st.button("Run Query"):

        connection = get_db_connection()

        df = pd.read_sql(queries[query_choice], connection)

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

            st.plotly_chart(fig, use_container_width=True)

def show_data_explorer():

    """Interactive data exploration"""

    st.subheader("Data Explorer")

    table = st.selectbox("Select Table",

                        ["artifactmetadata", "artifactmedia", "artifactcolors"])

    connection = get_db_connection()

    df = pd.read_sql(f"SELECT * FROM {table} LIMIT 100", connection)

    st.dataframe(df)

    st.caption(f"Showing first 100 rows from {table}")

if __name__ == "__main__":

    main()

Running the Dashboard

streamlit run app.py

Common Patterns

Pagination Handler

def fetch_all_pages(api_key, max_pages=None):

    """

    Fetch all available pages from API

    """

    page = 1

    all_records = []

    while True:

        data = get_objects(api_key, page=page)

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

        if not records:

            break

        all_records.extend(records)

        info = data.get('info', {})

        if page >= info.get('pages', 0):

            break

        if max_pages and page >= max_pages:

            break

        page += 1

        time.sleep(0.5)  # Rate limiting

    return all_records

Error Handling

def safe_etl_execution():

    """ETL with comprehensive error handling"""

    try:

        connection = get_db_connection()

        api_key = os.getenv('HARVARD_API_KEY')

        if not api_key:

            raise ValueError("HARVARD_API_KEY not set")

        run_etl_pipeline(api_key, connection)

    except mysql.connector.Error as db_error:

        print(f"Database error: {db_error}")

        # Implement retry logic or alerting

    except requests.exceptions.RequestException as api_error:

        print(f"API error: {api_error}")

        # Log and retry

    finally:

        if connection.is_connected():

            connection.close()

Troubleshooting

API Rate Limiting

If you encounter rate limit errors:

import time

from functools import wraps

def rate_limited(max_per_second=2):

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

    min_interval = 1.0 / max_per_second

    def decorator(func):

        last_called = [0.0]

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

            result = func(*args, **kwargs)

            last_called[0] = time.time()

            return result

        return wrapper

    return decorator

@rate_limited(max_per_second=2)

def fetch_data(url, params):

    return requests.get(url, params=params)

Database Connection Issues

def get_db_connection_with_retry(max_retries=3):

    """Database connection with retry logic"""

    for attempt in range(max_retries):

        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

            )

            return connection

        except mysql.connector.Error as e:

            if attempt == max_retries - 1:

                raise

            time.sleep(2 ** attempt)  # Exponential backoff

Memory Optimization for Large Datasets

def batch_process_records(records, batch_size=1000):

    """Process large datasets in batches"""

    for i in range(0, len(records), batch_size):

        batch = records[i:i + batch_size]

        metadata_df = transform_metadata(batch)

        media_df = transform_media(batch)

        colors_df = transform_colors(batch)

        connection = get_db_connection()

        load_to_database(metadata_df, 'artifactmetadata', connection)

        load_to_database(media_df, 'artifactmedia', connection)

        load_to_database(colors_df, 'artifactcolors', connection)

        connection.close()

        print(f"Processed batch {i//batch_size + 1}")

Handling Missing Data

def clean_record(record):

    """Clean and validate record data"""

    return {

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

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

        'culture': (record.get('culture') or '')[:255],

        'century': (record.get('century') or '')[:100],

        # Use empty string instead of None for NOT NULL fields

        'classification': (record.get('classification') or '')[:255],

        'department': (record.get('department') or '')[:255],

    }

This skill provides everything needed to build production-ready data pipelines using the Harvard Art Museums API with proper ETL practices, SQL analytics, and interactive dashboards.