harvard-artifacts-data-engineering-app

name	harvard-artifacts-data-engineering-app
description	Build ETL pipelines and analytics dashboards using the Harvard Art Museums API with SQL storage and Streamlit visualization
triggers	["build a data pipeline for museum artifacts","create an ETL workflow with Harvard Art Museums API","analyze art collection data with SQL","visualize museum artifacts data in Streamlit","extract and transform API data into relational database","build analytics dashboard for art museum collections","create data engineering pipeline for cultural artifacts","query and visualize Harvard museum data"]

Harvard Artifacts Data Engineering App

Skill by ara.so — Data Skills collection.

This project demonstrates end-to-end data engineering using the Harvard Art Museums API. It extracts artifact data, transforms it into relational tables, loads it into SQL databases (MySQL/TiDB), and provides interactive analytics through Streamlit dashboards with Plotly visualizations.

What It Does

• API Integration: Fetches paginated artifact data from Harvard Art Museums API
• ETL Pipeline: Transforms nested JSON into normalized relational tables
• SQL Storage: Creates and populates artifactmetadata, artifactmedia, and artifactcolors tables
• Analytics Queries: 20+ predefined SQL queries for artifact insights
• Interactive Visualization: Streamlit dashboard with Plotly charts

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

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

Requirements

streamlit
pandas
requests
pymysql
plotly
python-dotenv

Configuration

Database Setup

The application expects a MySQL/TiDB database with three main tables:

import pymysql
import os

# Database connection
def get_db_connection():
    return pymysql.connect(
        host=os.getenv('DB_HOST'),
        user=os.getenv('DB_USER'),
        password=os.getenv('DB_PASSWORD'),
        database=os.getenv('DB_NAME'),
        cursorclass=pymysql.cursors.DictCursor
    )

# Create tables
def create_tables(connection):
    with connection.cursor() as cursor:
        # Artifact Metadata table
        cursor.execute("""
        CREATE TABLE IF NOT EXISTS artifactmetadata (
            id INT PRIMARY KEY,
            title VARCHAR(500),
            culture VARCHAR(255),
            period VARCHAR(255),
            century VARCHAR(100),
            classification VARCHAR(255),
            department VARCHAR(255),
            technique VARCHAR(500),
            dated VARCHAR(255)
        )
        """)
        
        # Artifact Media table
        cursor.execute("""
        CREATE TABLE IF NOT EXISTS artifactmedia (
            media_id INT AUTO_INCREMENT PRIMARY KEY,
            artifact_id INT,
            baseimageurl VARCHAR(1000),
            primaryimageurl VARCHAR(1000),
            FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
        )
        """)
        
        # Artifact Colors table
        cursor.execute("""
        CREATE TABLE IF NOT EXISTS 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)
        )
        """)
        
    connection.commit()

API Configuration

import requests
import os

API_KEY = os.getenv('HARVARD_API_KEY')
BASE_URL = "https://api.harvardartmuseums.org/object"

def fetch_artifacts(page=1, size=100):
    """Fetch artifacts from Harvard Art Museums API"""
    params = {
        'apikey': API_KEY,
        'page': page,
        'size': size
    }
    
    response = requests.get(BASE_URL, params=params)
    response.raise_for_status()
    return response.json()

Key Components

1. ETL Pipeline

import pandas as pd

def extract_artifacts(num_pages=5):
    """Extract artifacts from API"""
    all_records = []
    
    for page in range(1, num_pages + 1):
        data = fetch_artifacts(page=page, size=100)
        records = data.get('records', [])
        all_records.extend(records)
        
        # Respect rate limits
        if not data.get('info', {}).get('next'):
            break
    
    return all_records

def transform_artifacts(records):
    """Transform nested JSON to relational format"""
    metadata = []
    media = []
    colors = []
    
    for record in records:
        artifact_id = record.get('id')
        
        # Extract metadata
        metadata.append({
            'id': artifact_id,
            'title': record.get('title'),
            'culture': record.get('culture'),
            'period': record.get('period'),
            'century': record.get('century'),
            'classification': record.get('classification'),
            'department': record.get('department'),
            'technique': record.get('technique'),
            'dated': record.get('dated')
        })
        
        # Extract media
        media.append({
            'artifact_id': artifact_id,
            'baseimageurl': record.get('baseimageurl'),
            'primaryimageurl': record.get('primaryimageurl')
        })
        
        # Extract colors
        for color_data in record.get('colors', []):
            colors.append({
                'artifact_id': artifact_id,
                'color': color_data.get('color'),
                'spectrum': color_data.get('spectrum'),
                'percentage': color_data.get('percent')
            })
    
    return (
        pd.DataFrame(metadata),
        pd.DataFrame(media),
        pd.DataFrame(colors)
    )

def load_to_database(df_metadata, df_media, df_colors):
    """Load dataframes to SQL database"""
    connection = get_db_connection()
    
    try:
        # Load metadata
        for _, row in df_metadata.iterrows():
            with connection.cursor() as cursor:
                cursor.execute("""
                INSERT INTO artifactmetadata 
                (id, title, culture, period, century, classification, department, technique, dated)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE title=VALUES(title)
                """, tuple(row))
        
        # Load media
        for _, row in df_media.iterrows():
            with connection.cursor() as cursor:
                cursor.execute("""
                INSERT INTO artifactmedia (artifact_id, baseimageurl, primaryimageurl)
                VALUES (%s, %s, %s)
                """, tuple(row))
        
        # Load colors
        for _, row in df_colors.iterrows():
            with connection.cursor() as cursor:
                cursor.execute("""
                INSERT INTO artifactcolors (artifact_id, color, spectrum, percentage)
                VALUES (%s, %s, %s, %s)
                """, tuple(row))
        
        connection.commit()
    finally:
        connection.close()

2. SQL Analytics Queries

# Sample analytical queries
ANALYTICS_QUERIES = {
    "Artifacts by Culture": """
        SELECT culture, COUNT(*) as artifact_count
        FROM artifactmetadata
        WHERE culture IS NOT NULL
        GROUP BY culture
        ORDER BY artifact_count DESC
        LIMIT 10
    """,
    
    "Century Distribution": """
        SELECT century, COUNT(*) as count
        FROM artifactmetadata
        WHERE century IS NOT NULL
        GROUP BY century
        ORDER BY count DESC
    """,
    
    "Department Statistics": """
        SELECT department, 
               COUNT(*) as total_artifacts,
               COUNT(DISTINCT culture) as cultures_represented
        FROM artifactmetadata
        WHERE department IS NOT NULL
        GROUP BY department
        ORDER BY total_artifacts DESC
    """,
    
    "Media Availability": """
        SELECT 
            SUM(CASE WHEN primaryimageurl IS NOT NULL THEN 1 ELSE 0 END) as with_image,
            SUM(CASE WHEN primaryimageurl IS NULL THEN 1 ELSE 0 END) as without_image
        FROM artifactmedia
    """,
    
    "Top Colors": """
        SELECT color, spectrum, 
               COUNT(*) as occurrences,
               AVG(percentage) as avg_percentage
        FROM artifactcolors
        GROUP BY color, spectrum
        ORDER BY occurrences DESC
        LIMIT 15
    """,
    
    "Classification by Century": """
        SELECT century, classification, COUNT(*) as count
        FROM artifactmetadata
        WHERE century IS NOT NULL AND classification IS NOT NULL
        GROUP BY century, classification
        ORDER BY century, count DESC
    """
}

def execute_query(query_name):
    """Execute analytical query and return results"""
    connection = get_db_connection()
    
    try:
        df = pd.read_sql(ANALYTICS_QUERIES[query_name], connection)
        return df
    finally:
        connection.close()

3. Streamlit Dashboard

import streamlit as st
import plotly.express as px

def main():
    st.title("Harvard Art Museums Analytics Dashboard")
    
    # Sidebar for ETL controls
    st.sidebar.header("Data Collection")
    num_pages = st.sidebar.slider("Number of pages to fetch", 1, 10, 5)
    
    if st.sidebar.button("Run ETL Pipeline"):
        with st.spinner("Extracting data from API..."):
            records = extract_artifacts(num_pages)
            st.success(f"Extracted {len(records)} artifacts")
        
        with st.spinner("Transforming data..."):
            df_metadata, df_media, df_colors = transform_artifacts(records)
            st.success("Data transformed")
        
        with st.spinner("Loading to database..."):
            load_to_database(df_metadata, df_media, df_colors)
            st.success("Data loaded successfully")
    
    # Analytics section
    st.header("Analytics Queries")
    
    query_name = st.selectbox("Select Query", list(ANALYTICS_QUERIES.keys()))
    
    if st.button("Execute Query"):
        df_result = execute_query(query_name)
        
        # Display table
        st.dataframe(df_result)
        
        # Auto-generate visualization
        if len(df_result.columns) >= 2:
            fig = px.bar(
                df_result,
                x=df_result.columns[0],
                y=df_result.columns[1],
                title=query_name
            )
            st.plotly_chart(fig)
        
        # Download option
        csv = df_result.to_csv(index=False)
        st.download_button(
            label="Download CSV",
            data=csv,
            file_name=f"{query_name.replace(' ', '_')}.csv",
            mime="text/csv"
        )

if __name__ == "__main__":
    main()

Running the Application

# Start the Streamlit dashboard
streamlit run app.py

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

Common Patterns

Batch Processing

def batch_insert(connection, table, data, batch_size=1000):
    """Insert data in batches for better performance"""
    for i in range(0, len(data), batch_size):
        batch = data[i:i + batch_size]
        # Insert batch logic here
        connection.commit()

Error Handling for API Calls

import time

def fetch_with_retry(url, params, max_retries=3):
    """Fetch data with retry logic"""
    for attempt in range(max_retries):
        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff

Data Quality Checks

def validate_artifacts(df):
    """Validate artifact data quality"""
    checks = {
        'missing_ids': df['id'].isna().sum(),
        'duplicate_ids': df['id'].duplicated().sum(),
        'missing_titles': df['title'].isna().sum(),
        'total_records': len(df)
    }
    return checks

Troubleshooting

API Rate Limits

If you encounter rate limit errors:

import time

# Add delay between requests
time.sleep(0.5)  # 500ms delay

# Or implement exponential backoff

Database Connection Issues

# Test connection
try:
    connection = get_db_connection()
    connection.ping(reconnect=True)
    print("Database connection successful")
except Exception as e:
    print(f"Connection failed: {e}")

Memory Issues with Large Datasets

# Process in chunks
chunk_size = 100
for i in range(0, total_records, chunk_size):
    chunk = records[i:i + chunk_size]
    process_chunk(chunk)

Streamlit Caching

@st.cache_data(ttl=3600)  # Cache for 1 hour
def cached_query(query_name):
    return execute_query(query_name)

Advanced Usage

Custom Query Builder

def build_custom_query(filters):
    """Build dynamic SQL query from user filters"""
    base_query = "SELECT * FROM artifactmetadata WHERE 1=1"
    
    if filters.get('culture'):
        base_query += f" AND culture = '{filters['culture']}'"
    
    if filters.get('century'):
        base_query += f" AND century = '{filters['century']}'"
    
    if filters.get('department'):
        base_query += f" AND department = '{filters['department']}'"
    
    return base_query

Export to Multiple Formats

def export_results(df, format='csv'):
    """Export query results to various formats"""
    if format == 'csv':
        return df.to_csv(index=False)
    elif format == 'json':
        return df.to_json(orient='records')
    elif format == 'excel':
        return df.to_excel('results.xlsx', index=False)