Harvard Art Museums ETL Pipeline

Skill by ara.so — Data Skills collection.

This project provides an end-to-end data engineering solution for extracting, transforming, and analyzing artifacts from the Harvard Art Museums API. It demonstrates real-world ETL patterns, SQL analytics, and interactive visualization using Streamlit.

What It Does

• Extracts artifact data from 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 proper foreign key relationships

• Analyzes data using predefined SQL queries for cultural insights

• Visualizes results through interactive Plotly charts in a Streamlit dashboard

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

API Key Setup

Get your Harvard Art Museums API key from: https://www.harvardartmuseums.org/collections/api

# Store in environment variables

# .env file

HARVARD_API_KEY=your_api_key_here

DB_HOST=your_database_host

DB_USER=your_database_user

DB_PASSWORD=your_database_password

DB_NAME=harvard_artifacts

Database Connection

import mysql.connector

import os

from dotenv import load_dotenv

load_dotenv()

def get_db_connection():

    """Create MySQL 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')

    )

Database Schema

Create Tables

-- Artifact Metadata Table

CREATE TABLE artifactmetadata (

    id INT PRIMARY KEY,

    title VARCHAR(500),

    culture VARCHAR(255),

    period VARCHAR(255),

    century VARCHAR(100),

    classification VARCHAR(255),

    department VARCHAR(255),

    dated VARCHAR(255),

    accession_year INT,

    primary_image_url TEXT,

    url TEXT

);

-- Artifact Media Table

CREATE TABLE artifactmedia (

    media_id INT AUTO_INCREMENT PRIMARY KEY,

    artifact_id INT,

    media_type VARCHAR(100),

    base_url TEXT,

    image_url TEXT,

    description TEXT,

    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id) ON DELETE CASCADE

);

-- Artifact Colors Table

CREATE TABLE artifactcolors (

    color_id INT AUTO_INCREMENT PRIMARY KEY,

    artifact_id INT,

    color VARCHAR(50),

    hex_code VARCHAR(10),

    percentage DECIMAL(5,2),

    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id) ON DELETE CASCADE

);

Key API Patterns

Extract: Fetch Artifacts with Pagination

import requests

import os

from typing import List, Dict

def fetch_artifacts(api_key: str, num_pages: int = 5, page_size: int = 100) -> List[Dict]:

    """

    Fetch artifacts from Harvard Art Museums API with pagination

    Args:

        api_key: Harvard API key from environment

        num_pages: Number of pages to fetch

        page_size: Records per page (max 100)

    Returns:

        List of artifact dictionaries

    """

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

    all_artifacts = []

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

        params = {

            'apikey': api_key,

            'size': page_size,

            'page': page,

            'hasimage': 1  # Only artifacts with images

        }

        try:

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

            response.raise_for_status()

            data = response.json()

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

            all_artifacts.extend(records)

            print(f"Fetched page {page}: {len(records)} artifacts")

        except requests.exceptions.RequestException as e:

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

            break

    return all_artifacts

Transform: Normalize Nested JSON

import pandas as pd

from typing import List, Dict, Tuple

def transform_artifacts(raw_data: List[Dict]) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:

    """

    Transform raw API data into normalized dataframes

    Returns:

        Tuple of (metadata_df, media_df, colors_df)

    """

    metadata_list = []

    media_list = []

    colors_list = []

    for artifact in raw_data:

        # Extract metadata

        metadata = {

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

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

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

            'period': artifact.get('period', '')[:255],

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

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

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

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

            'accession_year': artifact.get('accessionyear'),

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

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

        }

        metadata_list.append(metadata)

        # Extract media

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

        for img in images:

            media = {

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

                'media_type': 'image',

                'base_url': img.get('baseimageurl', ''),

                'image_url': img.get('iiifbaseuri', ''),

                'description': img.get('description', '')[:1000] if img.get('description') else None

            }

            media_list.append(media)

        # Extract colors

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

        for color in colors:

            color_data = {

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

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

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

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

            }

            colors_list.append(color_data)

    return (

        pd.DataFrame(metadata_list),

        pd.DataFrame(media_list),

        pd.DataFrame(colors_list)

    )

Load: Batch Insert into Database

from typing import List

import mysql.connector

def load_metadata(df: pd.DataFrame, connection):

    """Load artifact metadata with batch insert"""

    cursor = connection.cursor()

    insert_query = """

    INSERT INTO artifactmetadata

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

     dated, accession_year, primary_image_url, url)

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

    ON DUPLICATE KEY UPDATE

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

    """

    # Convert dataframe to list of tuples

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

    cursor.executemany(insert_query, data)

    connection.commit()

    print(f"Inserted {cursor.rowcount} metadata records")

def load_media(df: pd.DataFrame, connection):

    """Load artifact media"""

    cursor = connection.cursor()

    insert_query = """

    INSERT INTO artifactmedia

    (artifact_id, media_type, base_url, image_url, description)

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

    """

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

    cursor.executemany(insert_query, data)

    connection.commit()

    print(f"Inserted {cursor.rowcount} media records")

def load_colors(df: pd.DataFrame, connection):

    """Load artifact colors"""

    cursor = connection.cursor()

    insert_query = """

    INSERT INTO artifactcolors

    (artifact_id, color, hex_code, percentage)

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

    """

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

    cursor.executemany(insert_query, data)

    connection.commit()

    print(f"Inserted {cursor.rowcount} color records")

Streamlit Dashboard

Main Application Structure

import streamlit as st

import pandas as pd

import plotly.express as px

from dotenv import load_dotenv

import os

load_dotenv()

st.set_page_config(

    page_title="Harvard Art Museums Analytics",

    page_icon="🎨",

    layout="wide"

)

def main():

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

    # Sidebar navigation

    page = st.sidebar.selectbox(

        "Choose a page",

        ["ETL Pipeline", "SQL Analytics", "Visualizations"]

    )

    if page == "ETL Pipeline":

        show_etl_page()

    elif page == "SQL Analytics":

        show_analytics_page()

    else:

        show_visualizations_page()

if __name__ == "__main__":

    main()

ETL Pipeline Page

def show_etl_page():

    st.header("ETL Pipeline")

    api_key = st.text_input("Harvard API Key", type="password",

                            value=os.getenv('HARVARD_API_KEY', ''))

    col1, col2 = st.columns(2)

    with col1:

        num_pages = st.number_input("Number of pages", min_value=1, max_value=100, value=5)

    with col2:

        page_size = st.number_input("Page size", min_value=10, max_value=100, value=100)

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

        with st.spinner("Extracting data from API..."):

            raw_data = fetch_artifacts(api_key, num_pages, page_size)

            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, {len(media_df)} media, {len(colors_df)} color records")

        with st.spinner("Loading into database..."):

            conn = get_db_connection()

            load_metadata(metadata_df, conn)

            load_media(media_df, conn)

            load_colors(colors_df, conn)

            conn.close()

            st.success("✅ Data loaded successfully!")

        # Show preview

        st.subheader("Data Preview")

        st.dataframe(metadata_df.head(10))

Analytics Queries

Common SQL Query Patterns

# Top 10 cultures by artifact count

QUERY_TOP_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_BY_CENTURY = """

SELECT century, COUNT(*) as count

FROM artifactmetadata

WHERE century IS NOT NULL

GROUP BY century

ORDER BY count DESC;

"""

# Department distribution

QUERY_BY_DEPARTMENT = """

SELECT department, COUNT(*) as count

FROM artifactmetadata

WHERE department IS NOT NULL

GROUP BY department

ORDER BY count DESC;

"""

# Most common colors

QUERY_TOP_COLORS = """

SELECT color, COUNT(*) as frequency, AVG(percentage) as avg_percentage

FROM artifactcolors

GROUP BY color

ORDER BY frequency DESC

LIMIT 15;

"""

# Artifacts with most images

QUERY_MOST_IMAGES = """

SELECT m.title, m.culture, COUNT(med.media_id) as image_count

FROM artifactmetadata m

JOIN artifactmedia med ON m.id = med.artifact_id

GROUP BY m.id, m.title, m.culture

ORDER BY image_count DESC

LIMIT 10;

"""

def execute_query(query: str, connection) -> pd.DataFrame:

    """Execute SQL query and return results as DataFrame"""

    return pd.read_sql(query, connection)

Analytics Dashboard Page

def show_analytics_page():

    st.header("📊 SQL Analytics")

    queries = {

        "Top 10 Cultures": QUERY_TOP_CULTURES,

        "Artifacts by Century": QUERY_BY_CENTURY,

        "Department Distribution": QUERY_BY_DEPARTMENT,

        "Most Common Colors": QUERY_TOP_COLORS,

        "Artifacts with Most Images": QUERY_MOST_IMAGES

    }

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

    if st.button("Run Query"):

        conn = get_db_connection()

        df = execute_query(queries[selected_query], conn)

        conn.close()

        st.subheader("Query 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=selected_query)

            st.plotly_chart(fig, use_container_width=True)

Visualization Patterns

Create Interactive Charts

import plotly.express as px

import plotly.graph_objects as go

def create_culture_chart(df: pd.DataFrame):

    """Bar chart for culture distribution"""

    fig = px.bar(

        df,

        x='culture',

        y='artifact_count',

        title='Top Cultures in Collection',

        labels={'artifact_count': 'Number of Artifacts', 'culture': 'Culture'},

        color='artifact_count',

        color_continuous_scale='Viridis'

    )

    fig.update_layout(xaxis_tickangle=-45)

    return fig

def create_century_timeline(df: pd.DataFrame):

    """Timeline visualization for centuries"""

    fig = px.line(

        df,

        x='century',

        y='count',

        title='Artifact Distribution Across Centuries',

        markers=True

    )

    return fig

def create_color_pie_chart(df: pd.DataFrame):

    """Pie chart for color distribution"""

    fig = px.pie(

        df,

        values='frequency',

        names='color',

        title='Most Common Colors in Artifacts',

        hole=0.3

    )

    return fig

Complete ETL Workflow

from dotenv import load_dotenv

import os

def run_complete_etl():

    """Complete ETL pipeline execution"""

    load_dotenv()

    # 1. Extract

    print("Starting extraction...")

    api_key = os.getenv('HARVARD_API_KEY')

    raw_data = fetch_artifacts(api_key, num_pages=10, page_size=100)

    print(f"Extracted {len(raw_data)} artifacts")

    # 2. Transform

    print("Transforming data...")

    metadata_df, media_df, colors_df = transform_artifacts(raw_data)

    # Data quality checks

    print(f"Metadata records: {len(metadata_df)}")

    print(f"Media records: {len(media_df)}")

    print(f"Color records: {len(colors_df)}")

    print(f"Null values in metadata:\n{metadata_df.isnull().sum()}")

    # 3. Load

    print("Loading to database...")

    conn = get_db_connection()

    try:

        load_metadata(metadata_df, conn)

        load_media(media_df, conn)

        load_colors(colors_df, conn)

        print("ETL pipeline completed successfully!")

    except Exception as e:

        print(f"Error during load: {e}")

        conn.rollback()

    finally:

        conn.close()

if __name__ == "__main__":

    run_complete_etl()

Troubleshooting

API Rate Limiting

import time

from requests.exceptions import HTTPError

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

    """Fetch with exponential backoff for rate limiting"""

    for attempt in range(max_retries):

        try:

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

            response.raise_for_status()

            return response.json()

        except HTTPError as e:

            if e.response.status_code == 429:  # Too Many Requests

                wait_time = (2 ** attempt) * 5  # Exponential backoff

                print(f"Rate limited. Waiting {wait_time}s...")

                time.sleep(wait_time)

            else:

                raise

    raise Exception("Max retries exceeded")

Database Connection Issues

import mysql.connector

from mysql.connector import Error

def safe_db_connection():

    """Create database connection with error handling"""

    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,

            autocommit=False

        )

        if connection.is_connected():

            print("Database connection successful")

            return connection

    except Error as e:

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

        return None

Handling Missing Data

def clean_artifact_data(df: pd.DataFrame) -> pd.DataFrame:

    """Clean and validate artifact data"""

    # Replace empty strings with None

    df = df.replace('', None)

    # Handle missing titles

    df['title'] = df['title'].fillna('Untitled')

    # Ensure numeric fields

    df['accession_year'] = pd.to_numeric(df['accession_year'], errors='coerce')

    # Truncate long text fields

    df['title'] = df['title'].str[:500]

    df['culture'] = df['culture'].str[:255]

    return df

Running the Application

# Start Streamlit dashboard

streamlit run app.py

# Run ETL pipeline only

python etl_pipeline.py

# Run with custom configuration

HARVARD_API_KEY=your_key DB_HOST=localhost streamlit run app.py

This skill enables AI agents to help developers build complete ETL pipelines for cultural heritage data, from API extraction through SQL analytics to interactive visualization.