Harvard Artifacts Collection Analytics App

Skill by ara.so — Data Skills collection

What This Project Does

The Harvard-Artifacts-Collection-Data-Engineering-Analytics-App is an end-to-end data engineering application that demonstrates real-world ETL pipelines, SQL analytics, and interactive visualization. It extracts artifact data from the Harvard Art Museums API, transforms it into relational database structures, and provides interactive analytics dashboards using Streamlit.

Key capabilities:

• Extract artifact metadata, media, and color data from Harvard Art Museums API

• Transform nested JSON into normalized relational tables

• Load data into MySQL/TiDB Cloud databases

• Execute analytical SQL queries

• Visualize results with interactive Plotly charts

• Handle API pagination and rate limiting

Installation

Prerequisites

# Python 3.7+

python --version

# Install required packages

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

Project Setup

# 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

# Create .env file for configuration

touch .env

Environment Configuration

Create a .env file with the following variables:

# Harvard Art Museums API

HARVARD_API_KEY=your_api_key_here

# Database Configuration (MySQL/TiDB Cloud)

DB_HOST=your_database_host

DB_PORT=3306

DB_USER=your_database_user

DB_PASSWORD=your_database_password

DB_NAME=harvard_artifacts

To get a Harvard Art Museums API key:

• Visit https://www.harvardartmuseums.org/collections/api

• Request an API key (free)

• Add it to your .env file

Running the Application

# Start the Streamlit app

streamlit run app.py

# The app will open in your browser at http://localhost:8501

Database Schema

The application creates three main tables:

-- Artifact metadata table

CREATE TABLE artifactmetadata (

    id INT PRIMARY KEY,

    title VARCHAR(500),

    culture VARCHAR(255),

    century VARCHAR(255),

    dated VARCHAR(255),

    classification VARCHAR(255),

    department VARCHAR(255),

    technique VARCHAR(500),

    medium VARCHAR(500),

    dimensions VARCHAR(500),

    creditline TEXT,

    accession_number VARCHAR(255),

    url TEXT

);

-- Artifact media table

CREATE TABLE artifactmedia (

    media_id INT AUTO_INCREMENT PRIMARY KEY,

    artifact_id INT,

    image_url TEXT,

    media_type VARCHAR(100),

    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)

);

-- Artifact colors table

CREATE TABLE artifactcolors (

    color_id INT AUTO_INCREMENT PRIMARY KEY,

    artifact_id INT,

    color VARCHAR(50),

    percentage FLOAT,

    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)

);

Code Examples

1. API Data Extraction

import requests

import os

from dotenv import load_dotenv

load_dotenv()

def fetch_artifacts(api_key, page=1, size=100):

    """

    Fetch artifacts from Harvard Art Museums API

    Args:

        api_key: Harvard API key

        page: Page number for pagination

        size: Number of records per page (max 100)

    Returns:

        JSON response with artifact data

    """

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

    params = {

        'apikey': api_key,

        'page': page,

        'size': size,

        'hasimage': 1  # Only artifacts with images

    }

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

    response.raise_for_status()

    return response.json()

# Usage

api_key = os.getenv('HARVARD_API_KEY')

data = fetch_artifacts(api_key, page=1, size=50)

print(f"Total records: {data['info']['totalrecords']}")

print(f"Fetched: {len(data['records'])} artifacts")

2. ETL Pipeline - Transform

import pandas as pd

def transform_artifacts(raw_data):

    """

    Transform raw API data into structured DataFrames

    Args:

        raw_data: JSON response from API

    Returns:

        Dictionary of DataFrames (metadata, media, colors)

    """

    artifacts = []

    media_list = []

    colors_list = []

    for record in raw_data['records']:

        # Extract metadata

        artifact = {

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

            'title': record.get('title', 'Unknown'),

            'culture': record.get('culture'),

            'century': record.get('century'),

            'dated': record.get('dated'),

            'classification': record.get('classification'),

            'department': record.get('department'),

            'technique': record.get('technique'),

            'medium': record.get('medium'),

            'dimensions': record.get('dimensions'),

            'creditline': record.get('creditline'),

            'accession_number': record.get('accessionnumber'),

            'url': record.get('url')

        }

        artifacts.append(artifact)

        # Extract media

        if 'images' in record and record['images']:

            for img in record['images']:

                media_list.append({

                    'artifact_id': record['id'],

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

                    'media_type': 'image'

                })

        # Extract colors

        if 'colors' in record and record['colors']:

            for color in record['colors']:

                colors_list.append({

                    'artifact_id': record['id'],

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

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

                })

    return {

        'metadata': pd.DataFrame(artifacts),

        'media': pd.DataFrame(media_list),

        'colors': pd.DataFrame(colors_list)

    }

3. Database Loading

import mysql.connector

from mysql.connector import Error

def get_db_connection():

    """Create database connection"""

    try:

        connection = mysql.connector.connect(

            host=os.getenv('DB_HOST'),

            port=os.getenv('DB_PORT', 3306),

            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_artifacts_to_db(dataframes):

    """

    Load transformed data into SQL database

    Args:

        dataframes: Dictionary of DataFrames from transform_artifacts()

    """

    connection = get_db_connection()

    if not connection:

        return

    cursor = connection.cursor()

    try:

        # Insert metadata

        for _, row in dataframes['metadata'].iterrows():

            query = """

            INSERT INTO artifactmetadata

            (id, title, culture, century, dated, classification,

             department, technique, medium, dimensions, creditline,

             accession_number, url)

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

            ON DUPLICATE KEY UPDATE title=VALUES(title)

            """

            cursor.execute(query, tuple(row))

        # Insert media

        for _, row in dataframes['media'].iterrows():

            query = """

            INSERT INTO artifactmedia (artifact_id, image_url, media_type)

            VALUES (%s, %s, %s)

            """

            cursor.execute(query, tuple(row))

        # Insert colors

        for _, row in dataframes['colors'].iterrows():

            query = """

            INSERT INTO artifactcolors (artifact_id, color, percentage)

            VALUES (%s, %s, %s)

            """

            cursor.execute(query, tuple(row))

        connection.commit()

        print("Data loaded successfully!")

    except Error as e:

        print(f"Error loading data: {e}")

        connection.rollback()

    finally:

        cursor.close()

        connection.close()

4. Streamlit Dashboard

import streamlit as st

import plotly.express as px

def main():

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

    # Sidebar configuration

    st.sidebar.header("Configuration")

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

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

    num_records = st.sidebar.slider("Records to fetch", 10, 500, 100)

    # ETL Pipeline Section

    st.header("🔄 ETL Pipeline")

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

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

            raw_data = fetch_artifacts(api_key, page=1, size=num_records)

            st.success(f"Fetched {len(raw_data['records'])} artifacts")

        with st.spinner("Transforming data..."):

            transformed = transform_artifacts(raw_data)

            st.success("Data transformed successfully")

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

            load_artifacts_to_db(transformed)

            st.success("Data loaded to database")

    # Analytics Section

    st.header("📊 Analytics")

    query_options = {

        "Artifacts by Culture": """

            SELECT culture, COUNT(*) as count

            FROM artifactmetadata

            WHERE culture IS NOT NULL

            GROUP BY culture

            ORDER BY count DESC

            LIMIT 10

        """,

        "Artifacts by Century": """

            SELECT century, COUNT(*) as count

            FROM artifactmetadata

            WHERE century IS NOT NULL

            GROUP BY century

            ORDER BY count DESC

        """,

        "Top Colors Used": """

            SELECT color, SUM(percentage) as total_percentage

            FROM artifactcolors

            GROUP BY color

            ORDER BY total_percentage DESC

            LIMIT 10

        """

    }

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

    if st.button("Run Query"):

        connection = get_db_connection()

        if connection:

            df = pd.read_sql(query_options[selected_query], connection)

            st.dataframe(df)

            # Visualization

            if len(df) > 0:

                fig = px.bar(df, x=df.columns[0], y=df.columns[1],

                            title=selected_query)

                st.plotly_chart(fig)

            connection.close()

if __name__ == "__main__":

    main()

Common Analytical Queries

-- 1. Distribution of artifacts by department

SELECT department, COUNT(*) as artifact_count

FROM artifactmetadata

GROUP BY department

ORDER BY artifact_count DESC;

-- 2. Artifacts with media vs without

SELECT

    CASE WHEN m.artifact_id IS NOT NULL THEN 'Has Media' ELSE 'No Media' END as media_status,

    COUNT(*) as count

FROM artifactmetadata a

LEFT JOIN artifactmedia m ON a.id = m.artifact_id

GROUP BY media_status;

-- 3. Most common artifact classifications

SELECT classification, COUNT(*) as count

FROM artifactmetadata

WHERE classification IS NOT NULL

GROUP BY classification

ORDER BY count DESC

LIMIT 15;

-- 4. Color distribution across all artifacts

SELECT color, ROUND(AVG(percentage), 2) as avg_percentage, COUNT(*) as occurrences

FROM artifactcolors

GROUP BY color

ORDER BY occurrences DESC

LIMIT 20;

-- 5. Artifacts by culture and century

SELECT culture, century, COUNT(*) as count

FROM artifactmetadata

WHERE culture IS NOT NULL AND century IS NOT NULL

GROUP BY culture, century

ORDER BY count DESC

LIMIT 20;

Troubleshooting

API Rate Limiting

import time

def fetch_with_rate_limit(api_key, total_records=500, batch_size=100):

    """Fetch data with rate limiting"""

    all_data = []

    pages = (total_records // batch_size) + 1

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

        data = fetch_artifacts(api_key, page=page, size=batch_size)

        all_data.extend(data['records'])

        # Rate limit: 1 request per second

        time.sleep(1)

    return all_data

Database Connection Issues

def verify_db_connection():

    """Test database connectivity"""

    try:

        conn = get_db_connection()

        if conn and conn.is_connected():

            print("✓ Database connection successful")

            cursor = conn.cursor()

            cursor.execute("SELECT DATABASE()")

            db_name = cursor.fetchone()[0]

            print(f"✓ Connected to database: {db_name}")

            cursor.close()

            conn.close()

            return True

    except Error as e:

        print(f"✗ Connection failed: {e}")

        return False

Handling Missing Data

def safe_extract(record, key, default='Unknown'):

    """Safely extract nested data from API response"""

    try:

        value = record.get(key, default)

        return value if value else default

    except:

        return default

# Usage in transform function

artifact = {

    'title': safe_extract(record, 'title', 'Untitled'),

    'culture': safe_extract(record, 'culture', None),

    'century': safe_extract(record, 'century', None)

}

Best Practices

• Batch Processing: Process API data in batches to avoid memory issues

• Error Handling: Wrap API calls and DB operations in try-except blocks

• Environment Variables: Never hardcode credentials

• Data Validation: Validate data before inserting into database

• Logging: Add logging for ETL pipeline monitoring

• Incremental Loading: Track processed artifacts to avoid duplicates

import logging

logging.basicConfig(level=logging.INFO)

logger = logging.getLogger(__name__)

def etl_pipeline(api_key, num_records=100):

    """Complete ETL pipeline with logging"""

    try:

        logger.info(f"Starting ETL for {num_records} records")

        raw_data = fetch_artifacts(api_key, size=num_records)

        logger.info(f"Extracted {len(raw_data['records'])} records")

        transformed = transform_artifacts(raw_data)

        logger.info(f"Transformed into {len(transformed['metadata'])} artifacts")

        load_artifacts_to_db(transformed)

        logger.info("Data loaded successfully")

        return True

    except Exception as e:

        logger.error(f"ETL pipeline failed: {e}")

        return False