harvard-artifacts-data-engineering-pipeline

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

Harvard Artifacts Data Engineering 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, relational database design, SQL analytics, and interactive Streamlit dashboards.

What This Project Does

• API Integration: Fetches artifact data from Harvard Art Museums API with pagination and rate limiting
• ETL Pipeline: Extracts nested JSON, transforms into relational schema, loads into SQL database
• Database Design: Implements normalized tables (artifactmetadata, artifactmedia, artifactcolors)
• SQL Analytics: Executes 20+ predefined analytical queries
• Visualization: Interactive Plotly charts rendered through 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

Required Dependencies

streamlit
pandas
requests
mysql-connector-python
plotly
python-dotenv

Configuration

API Key Setup

Get your API key from Harvard Art Museums API.

Create a .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=your_database_name

Database Setup

The project uses MySQL or TiDB Cloud. Create the database schema:

CREATE DATABASE harvard_artifacts;
USE harvard_artifacts;

CREATE TABLE artifactmetadata (
    id INT PRIMARY KEY,
    title VARCHAR(500),
    culture VARCHAR(200),
    period VARCHAR(200),
    century VARCHAR(100),
    division VARCHAR(200),
    department VARCHAR(200),
    classification VARCHAR(200),
    technique VARCHAR(500),
    medium VARCHAR(500),
    url VARCHAR(500),
    dated VARCHAR(200)
);

CREATE TABLE artifactmedia (
    id INT AUTO_INCREMENT PRIMARY KEY,
    artifact_id INT,
    media_url VARCHAR(1000),
    media_type VARCHAR(100),
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);

CREATE TABLE artifactcolors (
    id INT AUTO_INCREMENT PRIMARY KEY,
    artifact_id INT,
    color_hex VARCHAR(10),
    color_percent FLOAT,
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);

Core Components

ETL Pipeline

Extract: Fetch Data from API

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')
    url = f"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)
    
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"API request failed: {response.status_code}")

# Fetch multiple pages
def fetch_all_artifacts(max_pages=10):
    """Fetch artifacts with pagination"""
    all_records = []
    
    for page in range(1, max_pages + 1):
        print(f"Fetching page {page}...")
        data = fetch_artifacts(page=page)
        all_records.extend(data.get('records', []))
        
        # Check if more pages available
        if not data.get('info', {}).get('next'):
            
    
     all_records

Transform: Process JSON to Relational Data

import pandas as pd

def transform_artifacts(raw_data):
    """Transform nested JSON to relational dataframes"""
    
    # Metadata table
    metadata_records = []
    media_records = []
    color_records = []
    
    for artifact in raw_data:
        # Extract metadata
        metadata_records.append({
            'id': artifact.get('id'),
            'title': artifact.get('title'),
            'culture': artifact.get('culture'),
            'period': artifact.get('period'),
            'century': artifact.get('century'),
            'division': artifact.get('division'),
            'department': artifact.get('department'),
            'classification': artifact.get('classification'),
            'technique': artifact.get('technique'),
            'medium': artifact.get('medium'),
            'url': artifact.get('url'),
            'dated': artifact.get('dated')
        })
        
        # Extract media
        for image in artifact.get('images', []):
            media_records.append({
                'artifact_id': artifact.get('id'),
                'media_url': image.get('baseimageurl'),
                'media_type': 
            })
        
        
         color  artifact.get(, []):
            color_records.append({
                : artifact.get(),
                : color.get(),
                : color.get()
            })
    
     {
        : pd.DataFrame(metadata_records),
        : pd.DataFrame(media_records),
        : pd.DataFrame(color_records)
    }

Load: Insert into SQL Database

import mysql.connector
from mysql.connector import Error

def create_connection():
    """Create database connection"""
    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')
        )
        return connection
    except Error as e:
        print(f"Error connecting to database: {e}")
        return None

def load_to_database(dataframes):
    """Load transformed data to SQL database"""
    connection = create_connection()
    if not connection:
        return False
    
    cursor = connection.cursor()
    
    try:
        # Load metadata
        for _, row in dataframes['metadata'].iterrows():
            query = """
                INSERT INTO artifactmetadata 
                (id, title, culture, period, century, division, department, 
                 classification, technique, medium, url, dated)
                VALUES (%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 dataframes['media'].iterrows():
            query = """
                INSERT INTO artifactmedia (artifact_id, media_url, media_type)
                VALUES (%s, %s, %s)
            """
            cursor.execute(query, (row))
        
        
         _, row  dataframes[].iterrows():
            query = 
            cursor.execute(query, (row))
        
        connection.commit()
         
        
     Error  e:
        ()
        connection.rollback()
         
    :
        cursor.close()
        connection.close()

Complete ETL Workflow

def run_etl_pipeline(max_pages=5):
    """Execute complete ETL pipeline"""
    print("Starting ETL pipeline...")
    
    # Extract
    print("Extracting data from API...")
    raw_data = fetch_all_artifacts(max_pages=max_pages)
    print(f"Extracted {len(raw_data)} artifacts")
    
    # Transform
    print("Transforming data...")
    dataframes = transform_artifacts(raw_data)
    print(f"Transformed into {len(dataframes['metadata'])} metadata records")
    
    # Load
    print("Loading data to database...")
    success = load_to_database(dataframes)
    
    if success:
        print("ETL pipeline completed successfully!")
    else:
        print("ETL pipeline failed!")
    
    return success

SQL Analytics Queries

Sample Analytical Queries

# Query 1: Top 10 cultures by artifact count
query_cultures = """
SELECT culture, COUNT(*) as artifact_count
FROM artifactmetadata
WHERE culture IS NOT NULL
GROUP BY culture
ORDER BY artifact_count DESC
LIMIT 10
"""

# Query 2: Artifacts by century
query_century = """
SELECT century, COUNT(*) as count
FROM artifactmetadata
WHERE century IS NOT NULL
GROUP BY century
ORDER BY count DESC
"""

# Query 3: Department distribution
query_departments = """
SELECT department, COUNT(*) as total_artifacts
FROM artifactmetadata
WHERE department IS NOT NULL
GROUP BY department
ORDER BY total_artifacts DESC
"""

# Query 4: Media availability
query_media = """
SELECT 
    CASE 
        WHEN media_count > 0 THEN 'Has Media'
        ELSE 'No Media'
    END as media_status,
    COUNT(*) as artifact_count
FROM (
    SELECT a.id, COUNT(m.id) as media_count
    FROM artifactmetadata a
    LEFT JOIN artifactmedia m ON a.id = m.artifact_id
    GROUP BY a.id
) as media_stats
GROUP BY media_status
"""

# Query 5: Top colors used
query_colors = """
SELECT color_hex, COUNT(*) as usage_count, AVG(color_percent) as avg_percent
FROM artifactcolors
GROUP BY color_hex
ORDER BY usage_count DESC
LIMIT 10
"""

# Query 6: Classification distribution
query_classification = """
SELECT classification, COUNT(*) as count
FROM artifactmetadata
WHERE classification IS NOT NULL
GROUP BY classification
ORDER BY count DESC
LIMIT 15
"""

Execute Queries

def execute_query(query):
    """Execute SQL query and return results as DataFrame"""
    connection = create_connection()
    if not connection:
        return None
    
    try:
        df = pd.read_sql(query, connection)
        return df
    except Error as e:
        print(f"Query execution error: {e}")
        return None
    finally:
        connection.close()

Streamlit Dashboard

Main Application

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 - Data Analytics Dashboard")
st.markdown("---")

# Sidebar navigation
page = st.sidebar.selectbox(
    "Select Page",
    ["ETL Pipeline", "SQL Analytics", "Visualizations"]
)

if page == "ETL Pipeline":
    st.header("📥 ETL Pipeline")
    
    max_pages = st.slider("Number of pages to fetch", 1, 20, 5)
    
    if st.button("Run ETL Pipeline"):
        with st.spinner("Running ETL pipeline..."):
            success = run_etl_pipeline(max_pages=max_pages)
            
        if success:
            st.success("ETL pipeline completed successfully!")
        else:
            st.error("ETL pipeline failed. Check logs.")

elif page == "SQL Analytics":
    st.header("📊 SQL Analytics")
    
    # Query selector
    queries = {
        "Top Cultures": query_cultures,
        "Century Distribution": query_century,
        "Department Stats": query_departments,
        "Media Availability": query_media,
        "Color Usage": query_colors,
        "Classifications": query_classification
    }
    
    selected_query = st.selectbox("Select Query", list(queries.keys()))
    
    if st.button():
         st.spinner():
            df = execute_query(queries[selected_query])
        
         df      df.empty:
            st.dataframe(df, use_container_width=)
            
            
             (df.columns) >= :
                fig = px.bar(
                    df, 
                    x=df.columns[], 
                    y=df.columns[],
                    title=
                )
                st.plotly_chart(fig, use_container_width=)

 page == :
    st.header()
    
    
    df_culture = execute_query(query_cultures)
     df_culture   :
        fig1 = px.bar(df_culture, x=, y=,
                     title=)
        st.plotly_chart(fig1, use_container_width=)
    
    
    df_century = execute_query(query_century)
     df_century   :
        fig2 = px.line(df_century, x=, y=,
                      title=)
        st.plotly_chart(fig2, use_container_width=)

Common Patterns

Batch Processing with Rate Limiting

import time

def fetch_with_rate_limit(pages, delay=1):
    """Fetch data with rate limiting"""
    results = []
    
    for page in range(1, pages + 1):
        data = fetch_artifacts(page=page)
        results.extend(data.get('records', []))
        
        # Rate limiting
        if page < pages:
            time.sleep(delay)
    
    return results

Error Handling in ETL

def safe_etl_pipeline(max_pages=5):
    """ETL pipeline with comprehensive error handling"""
    try:
        raw_data = fetch_all_artifacts(max_pages=max_pages)
    except Exception as e:
        print(f"Extraction failed: {e}")
        return False
    
    try:
        dataframes = transform_artifacts(raw_data)
    except Exception as e:
        print(f"Transformation failed: {e}")
        return False
    
    try:
        success = load_to_database(dataframes)
        return success
    except Exception as e:
        print(f"Loading failed: {e}")
        return False

Running the Application

# Start Streamlit app
streamlit run app.py

# The app will be available at http://localhost:8501

Troubleshooting

API Connection Issues

# Test API connection
def test_api_connection():
    try:
        data = fetch_artifacts(page=1, size=1)
        print("API connection successful!")
        return True
    except Exception as e:
        print(f"API connection failed: {e}")
        print("Check your API key in .env file")
        return False

Database Connection Issues

# Test database connection
def test_db_connection():
    connection = create_connection()
    if connection:
        print("Database connection successful!")
        connection.close()
        return True
    else:
        print("Database connection failed!")
        print("Check DB credentials in .env file")
        return False

Common Error: Empty Results

# Verify data in database
def check_data_exists():
    query = "SELECT COUNT(*) as total FROM artifactmetadata"
    df = execute_query(query)
    
    if df is not None:
        total = df['total'].iloc[0]
        print(f"Total artifacts in database: {total}")
        return total > 0
    return False

This skill enables AI agents to guide developers through building production-ready data engineering pipelines using the Harvard Art Museums API with proper ETL practices, SQL analytics, and interactive visualizations.