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)