SKILL.md
Harvard Artifacts ETL Analytics Skill
Skill by ara.so — Data Skills collection.
This skill enables AI coding agents to help developers build end-to-end ETL pipelines and analytics applications using the Harvard Art Museums API. The project demonstrates real-world data engineering patterns including API integration, data transformation, SQL database design, and interactive visualization with Streamlit.
What This Project Does
The Harvard Artifacts Collection application:
- Extracts artifact data from Harvard Art Museums API with pagination and rate limiting
- Transforms nested JSON into normalized relational tables
- Loads data into MySQL/TiDB Cloud databases
- Provides 20+ predefined analytical SQL queries
- Visualizes results with interactive Plotly charts in Streamlit
Architecture Flow: API → ETL → SQL → Analytics → Visualization
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.txtRequired packages:
streamlit
pandas
requests
mysql-connector-python
plotly
python-dotenvConfiguration
Environment Variables
Create a .env file in the project root:
# Harvard Art Museums API
HARVARD_API_KEY=your_api_key_here
# Database Connection
DB_HOST=your_database_host
DB_PORT=3306
DB_USER=your_db_user
DB_PASSWORD=your_db_password
DB_NAME=harvard_artifactsGet Harvard API Key
Register at: https://www.harvardartmuseums.org/collections/api
Database Setup
-- Create database
CREATE DATABASE harvard_artifacts;
USE harvard_artifacts;
-- Artifact metadata table
CREATE TABLE artifactmetadata (
id INT PRIMARY KEY,
title VARCHAR(500),
culture VARCHAR(200),
century VARCHAR(100),
classification VARCHAR(200),
department VARCHAR(200),
dated VARCHAR(200),
division VARCHAR(200),
medium VARCHAR(500),
technique VARCHAR(500),
period VARCHAR(200),
accessionyear INT,
totalpageviews INT,
totaluniquepageviews INT
);
-- Artifact media table
CREATE TABLE artifactmedia (
id INT AUTO_INCREMENT PRIMARY KEY,
artifact_id INT,
iiifbaseuri VARCHAR(500),
baseimageurl VARCHAR(500),
primaryimageurl VARCHAR(500),
imagecopyright TEXT,
FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);
-- Artifact colors table
CREATE TABLE artifactcolors (
id INT AUTO_INCREMENT PRIMARY KEY,
artifact_id INT,
color VARCHAR(50),
spectrum VARCHAR(50),
hue VARCHAR(50),
percent FLOAT,
FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);Running the Application
streamlit run app.pyThe Streamlit dashboard will open at http://localhost:8501
Key Code Patterns
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 with pagination
"""
url = "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)
response.raise_for_status()
return response.json()
# Usage
api_key = os.getenv("HARVARD_API_KEY")
data = fetch_artifacts(api_key, page=1, size=100)
print(f"Total records: {data['info']['totalrecords']}")
print(f"Total pages: {data['info']['pages']}")ETL Pipeline with Pagination
import pandas as pd
import time
def extract_all_artifacts(api_key, max_pages=10):
"""
Extract artifacts across multiple pages with rate limiting
"""
all_artifacts = []
for page in range(1, max_pages + 1):
try:
data = fetch_artifacts(api_key, page=page)
artifacts = data.get('records', [])
all_artifacts.extend(artifacts)
print(f"Extracted page {page}/{max_pages}: {len(artifacts)} artifacts")
time.sleep(0.5) # Rate limiting
except Exception as e:
print(f"Error on page {page}: {e}")
break
return all_artifacts
def transform_artifacts(raw_artifacts):
"""
Transform nested JSON into structured DataFrames
"""
metadata_list = []
media_list = []
colors_list = []
for artifact in raw_artifacts:
# Metadata
metadata_list.append({
'id': artifact.get('id'),
'title': artifact.get('title'),
'culture': artifact.get('culture'),
'century': artifact.get('century'),
'classification': artifact.get('classification'),
'department': artifact.get('department'),
'dated': artifact.get('dated'),
'division': artifact.get('division'),
'medium': artifact.get('medium'),
'technique': artifact.get('technique'),
'period': artifact.get('period'),
'accessionyear': artifact.get('accessionyear'),
'totalpageviews': artifact.get('totalpageviews', 0),
'totaluniquepageviews': artifact.get('totaluniquepageviews', 0)
})
# Media
if artifact.get('primaryimageurl'):
media_list.append({
'artifact_id': artifact.get('id'),
'iiifbaseuri': artifact.get('iiifbaseuri'),
'baseimageurl': artifact.get('baseimageurl'),
'primaryimageurl': artifact.get('primaryimageurl'),
'imagecopyright': artifact.get('imagecopyright')
})
# Colors
for color in artifact.get('colors', []):
colors_list.append({
'artifact_id': artifact.get('id'),
'color': color.get('color'),
'spectrum': color.get('spectrum'),
'hue': color.get('hue'),
'percent': color.get('percent')
})
return (
pd.DataFrame(metadata_list),
pd.DataFrame(media_list),
pd.DataFrame(colors_list)
)Database Loading
import mysql.connector
from mysql.connector import Error
def get_db_connection():
"""
Create MySQL database connection from environment variables
"""
return mysql.connector.connect(
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')
)
def load_metadata(df_metadata):
"""
Batch insert artifact metadata into database
"""
conn = get_db_connection()
cursor = conn.cursor()
insert_query = """
INSERT INTO artifactmetadata
(id, title, culture, century, classification, department, dated,
division, medium, technique, period, accessionyear,
totalpageviews, totaluniquepageviews)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
title=VALUES(title), culture=VALUES(culture)
"""
records = df_metadata.to_records(index=False).tolist()
cursor.executemany(insert_query, records)
conn.commit()
print(f"Inserted {cursor.rowcount} metadata records")
cursor.close()
conn.close()
def load_all_data(df_metadata, df_media, df_colors):
"""
Load all transformed data into respective tables
"""
load_metadata(df_metadata)
# Similar functions for media and colors
print("ETL pipeline completed successfully")Streamlit Analytics Dashboard
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 Collection Analytics")
# Sidebar for query selection
query_options = {
"Top 10 Cultures by Artifact Count": """
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
""",
"Most Common Colors": """
SELECT color, COUNT(*) as usage_count, AVG(percent) as avg_percent
FROM artifactcolors
GROUP BY color
ORDER BY usage_count DESC
LIMIT 10
""",
"Department Distribution": """
SELECT department, COUNT(*) as artifact_count
FROM artifactmetadata
WHERE department IS NOT NULL
GROUP BY department
ORDER BY artifact_count DESC
"""
}
selected_query = st.sidebar.selectbox("Select Analysis", list(query_options.keys()))
if st.button("Run Analysis"):
conn = get_db_connection()
df_result = pd.read_sql(query_options[selected_query], conn)
conn.close()
st.subheader(f"Results: {selected_query}")
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=selected_query)
st.plotly_chart(fig, use_container_width=True)Complete ETL Workflow
def run_full_etl_pipeline():
"""
Complete ETL pipeline from API to database
"""
# Extract
print("Starting extraction...")
api_key = os.getenv("HARVARD_API_KEY")
raw_artifacts = extract_all_artifacts(api_key, max_pages=5)
# Transform
print("Transforming data...")
df_metadata, df_media, df_colors = transform_artifacts(raw_artifacts)
# Load
print("Loading to database...")
load_all_data(df_metadata, df_media, df_colors)
print(f"Pipeline complete! Processed {len(raw_artifacts)} artifacts")
return df_metadata, df_media, df_colors
# Execute pipeline
if __name__ == "__main__":
run_full_etl_pipeline()Common Analytical Queries
-- Top viewed artifacts
SELECT title, culture, totalpageviews
FROM artifactmetadata
ORDER BY totalpageviews DESC
LIMIT 20;
-- Artifacts with color data
SELECT a.title, a.culture, c.color, c.percent
FROM artifactmetadata a
JOIN artifactcolors c ON a.id = c.artifact_id
WHERE c.percent > 50
ORDER BY c.percent DESC;
-- Media availability by department
SELECT
a.department,
COUNT(*) as total_artifacts,
COUNT(m.id) as with_media,
ROUND(COUNT(m.id) * 100.0 / COUNT(*), 2) as media_percentage
FROM artifactmetadata a
LEFT JOIN artifactmedia m ON a.id = m.artifact_id
GROUP BY a.department
ORDER BY total_artifacts DESC;Troubleshooting
API Rate Limits:
- Add
time.sleep(0.5)between requests
- Implement exponential backoff for 429 errors
Database Connection Issues:
try:
conn = get_db_connection()
conn.ping(reconnect=True, attempts=3, delay=5)
except Error as e:
print(f"Database error: {e}")Missing Data Fields:
# Safe field access
culture = artifact.get('culture', 'Unknown')
colors = artifact.get('colors', [])Streamlit Caching:
@st.cache_data(ttl=3600)
def load_cached_data(query):
conn = get_db_connection()
df = pd.read_sql(query, conn)
conn.close()
return dfThis skill provides everything needed to build production-ready ETL pipelines and analytics dashboards for museum collection data using modern Python data engineering tools.