harvard-artifacts-collection-data-engineering

name	harvard-artifacts-collection-data-engineering
description	End-to-end data engineering and analytics application for Harvard Art Museums API with ETL pipelines, SQL analytics, and Streamlit visualization
triggers	["build an ETL pipeline for museum artifact data","connect to Harvard Art Museums API","create a data engineering pipeline with Streamlit","analyze Harvard museum collection with SQL","set up artifact collection data warehouse","visualize museum data with interactive dashboards","implement batch data loading from Harvard API","query and analyze art museum metadata"]

Harvard Artifacts Collection Data Engineering

Skill by ara.so — Data Skills collection.

This project provides an end-to-end data engineering and analytics application built on the Harvard Art Museums API. It demonstrates real-world ETL pipelines, SQL database design, analytical queries, and interactive visualization using Streamlit. The architecture follows: API → ETL → SQL → Analytics → Visualization.

What This Project Does

• API Integration: Fetches artifact data from Harvard Art Museums API with pagination and rate limiting
• ETL Pipeline: Extracts, transforms, and loads nested JSON into relational database tables
• SQL Database: Stores structured data across artifactmetadata, artifactmedia, and artifactcolors tables
• Analytics: Executes 20+ predefined SQL queries for insights
• Visualization: Interactive dashboards using Plotly and 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 packages:

streamlit
pandas
requests
mysql-connector-python
plotly

Configuration

Environment Variables

Set up your configuration before running:

# Harvard Art Museums API Key (get from https://www.harvardartmuseums.org/collections/api)
export HARVARD_API_KEY="your_api_key_here"

# Database credentials
export DB_HOST="your_database_host"
export DB_PORT="3306"
export DB_USER="your_database_user"
export DB_PASSWORD="your_database_password"
export DB_NAME="harvard_artifacts"

Database Setup

Create the required tables:

CREATE DATABASE harvard_artifacts;
USE harvard_artifacts;

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

CREATE TABLE artifactmedia (
    media_id INT AUTO_INCREMENT PRIMARY KEY,
    artifact_id INT,
    baseimageurl TEXT,
    iiifbaseuri TEXT,
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);

CREATE TABLE artifactcolors (
    color_id INT AUTO_INCREMENT PRIMARY KEY,
    artifact_id INT,
    color VARCHAR(50),
    spectrum VARCHAR(50),
    percentage FLOAT,
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);

Key Components and Usage

1. API Data Extraction

import requests
import os

class HarvardAPIClient:
    def __init__(self):
        self.api_key = os.getenv('HARVARD_API_KEY')
        self.base_url = "https://api.harvardartmuseums.org/object"
    
    def fetch_artifacts(self, page=1, size=100):
        """Fetch artifacts with pagination"""
        params = {
            'apikey': self.api_key,
            'page': page,
            'size': size,
            'hasimage': 1  # Only artifacts with images
        }
        response = requests.get(self.base_url, params=params)
        response.raise_for_status()
        return response.json()
    
    def fetch_multiple_pages(self, num_pages=10):
        """Fetch multiple pages of artifacts"""
        all_artifacts = []
        for page in range(1, num_pages + 1):
            data = self.fetch_artifacts(page=page)
            all_artifacts.extend(data.get('records', []))
        return all_artifacts

2. ETL Pipeline Implementation

import pandas as pd
import mysql.connector
from typing import List, Dict

class ArtifactETL:
    def __init__(self, db_config):
        self.db_config = db_config
        self.conn = None
        
    def connect_db(self):
        """Establish database connection"""
        self.conn = mysql.connector.connect(
            host=self.db_config['host'],
            port=self.db_config['port'],
            user=self.db_config['user'],
            password=self.db_config['password'],
            database=self.db_config['database']
        )
        return self.conn
    
    def transform_artifacts(self, raw_data: List[Dict]) -> pd.DataFrame:
        """Transform raw JSON to structured DataFrame"""
        artifacts = []
        for item in raw_data:
            artifact = {
                'id': item.get('id'),
                'title': item.get('title', '')[:500],
                : item.get(, ),
                : item.get(, ),
                : item.get(, ),
                : item.get(, ),
                : item.get(, ),
                : item.get(, ),
                : item.get(, ),
                : item.get(, ),
                : item.get(, ),
                : item.get(, )
            }
            artifacts.append(artifact)
         pd.DataFrame(artifacts)
    
     () -> pd.DataFrame:
        
        media_records = []
         item  raw_data:
            artifact_id = item.get()
               item:
                media_records.append({
                    : artifact_id,
                    : item.get(, ),
                    : item.get(, )
                })
         pd.DataFrame(media_records)
    
     () -> pd.DataFrame:
        
        color_records = []
         item  raw_data:
            artifact_id = item.get()
            colors = item.get(, [])
             color  colors:
                color_records.append({
                    : artifact_id,
                    : color.get(, ),
                    : color.get(, ),
                    : color.get(, )
                })
         pd.DataFrame(color_records)
    
     ():
        
        cursor = .conn.cursor()
        
        
        cols = .join(df.columns)
        placeholders = .join([] * (df.columns))
        
        sql = 
        
        
        data = [(row)  row  df.values]
        
        
        cursor.executemany(sql, data)
        .conn.commit()
        cursor.close()
        
         (data)

3. Complete ETL Workflow

def run_etl_pipeline():
    """Execute complete ETL pipeline"""
    # Configuration
    db_config = {
        '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')
    }
    
    # Initialize clients
    api_client = HarvardAPIClient()
    etl = ArtifactETL(db_config)
    etl.connect_db()
    
    # Extract
    print("Extracting data from API...")
    raw_artifacts = api_client.fetch_multiple_pages(num_pages=5)
    print(f"Extracted {len(raw_artifacts)} artifacts")
    
    # Transform
    print("Transforming data...")
    df_artifacts = etl.transform_artifacts(raw_artifacts)
    df_media = etl.transform_media(raw_artifacts)
    df_colors = etl.transform_colors(raw_artifacts)
    
    # Load
    print("Loading data to database...")
    etl.load_data(df_artifacts, 'artifactmetadata')
    etl.load_data(df_media, 'artifactmedia')
    etl.load_data(df_colors, 'artifactcolors')
    
    print("ETL pipeline completed successfully!")
    etl.conn.close()

4. SQL Analytics Queries

class ArtifactAnalytics:
    def __init__(self, db_config):
        self.db_config = db_config
    
    def execute_query(self, query: str) -> pd.DataFrame:
        """Execute SQL query and return DataFrame"""
        conn = mysql.connector.connect(**self.db_config)
        df = pd.read_sql(query, conn)
        conn.close()
        return df
    
    def get_artifacts_by_culture(self):
        """Get artifact distribution by culture"""
        query = """
        SELECT culture, COUNT(*) as artifact_count
        FROM artifactmetadata
        WHERE culture IS NOT NULL AND culture != ''
        GROUP BY culture
        ORDER BY artifact_count DESC
        LIMIT 20;
        """
        return self.execute_query(query)
    
    def get_artifacts_by_century(self):
        """Get artifact distribution by century"""
        query = """
        SELECT century, COUNT(*) as count
        FROM artifactmetadata
        WHERE century IS NOT NULL AND century != ''
        GROUP BY century
        ORDER BY count DESC;
        """
        return self.execute_query(query)
    
    def get_color_distribution(self):
        """Get color usage across artifacts"""
        query = """
        SELECT color, COUNT(*) as frequency, AVG(percentage) as avg_percentage
        FROM artifactcolors
        GROUP BY color
        ORDER BY frequency DESC
        LIMIT 15;
        """
        return self.execute_query(query)
    
    def ():
        
        query = 
         .execute_query(query)

5. Streamlit Dashboard

import streamlit as st
import plotly.express as px

def create_dashboard():
    st.title("Harvard Art Museums Analytics Dashboard")
    
    # Sidebar for navigation
    st.sidebar.header("Navigation")
    option = st.sidebar.selectbox(
        "Choose Analysis",
        ["Overview", "Culture Analysis", "Century Distribution", "Color Patterns", "Media Analysis"]
    )
    
    # Initialize analytics
    db_config = {
        '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')
    }
    analytics = ArtifactAnalytics(db_config)
    
    if option == "Culture Analysis":
        st.header("Artifacts by Culture")
        df = analytics.get_artifacts_by_culture()
        
        # Display data table
        st.dataframe(df)
        
        # Create visualization
        fig = px.bar(df, x='culture', y='artifact_count', 
                     title='Top 20 Cultures by Artifact Count')
        st.plotly_chart(fig)
    
    elif option == "Century Distribution":
        st.header("Artifacts by Century")
        df = analytics.get_artifacts_by_century()
        
        st.dataframe(df)
        
        fig = px.bar(df, x='century', y=,
                     title=)
        st.plotly_chart(fig)
    
     option == :
        st.header()
        df = analytics.get_color_distribution()
        
        st.dataframe(df)
        
        fig = px.bar(df, x=, y=,
                     title=)
        st.plotly_chart(fig)

 __name__ == :
    create_dashboard()

Running the Application

# Start the Streamlit dashboard
streamlit run app.py

# Run ETL pipeline separately (if needed)
python etl_pipeline.py

Common Patterns

Incremental Data Loading

def incremental_load(last_update_date):
    """Load only new or updated artifacts"""
    query = f"""
    SELECT id FROM artifactmetadata 
    WHERE lastupdate > '{last_update_date}'
    """
    # Fetch only updated records from API
    # Update existing records in database

Error Handling in ETL

def safe_etl_run():
    try:
        run_etl_pipeline()
    except requests.HTTPError as e:
        print(f"API Error: {e}")
    except mysql.connector.Error as e:
        print(f"Database Error: {e}")
    except Exception as e:
        print(f"Unexpected Error: {e}")

Troubleshooting

API Rate Limiting: Add delays between requests

import time
time.sleep(0.5)  # 500ms delay between API calls

Database Connection Issues: Verify credentials and network access

# Test connection
try:
    conn = mysql.connector.connect(**db_config)
    print("Database connection successful")
    conn.close()
except Exception as e:
    print(f"Connection failed: {e}")

Missing Data Fields: Handle null values during transformation

artifact = {
    'title': item.get('title', 'Unknown')[:500],
    'culture': item.get('culture') or 'Not Specified'
}

Memory Issues with Large Datasets: Use chunking

chunk_size = 1000
for i in range(0, len(data), chunk_size):
    chunk = data[i:i+chunk_size]
    etl.load_data(pd.DataFrame(chunk), 'artifactmetadata')