SKILL.md
Harvard Artifacts ETL Pipeline
Skill by ara.so — Data Skills collection.
This skill enables AI coding agents to help developers build end-to-end data engineering pipelines using the Harvard Art Museums API. The project demonstrates real-world ETL patterns, SQL database design, analytical queries, and interactive visualization using Streamlit.
What This Project Does
The Harvard Artifacts Collection application:
- Fetches artifact data from Harvard Art Museums API with pagination
- Transforms nested JSON into relational database tables
- Loads data into MySQL/TiDB Cloud databases
- Executes analytical SQL queries on artifact metadata, media, and colors
- Visualizes results through interactive Plotly charts in Streamlit
Architecture: 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.txt
# Required packages
pip install streamlit pandas requests mysql-connector-python plotly python-dotenvConfiguration
Environment Variables
Create a .env file in the project root:
# Harvard API
HARVARD_API_KEY=your_api_key_here
# Database credentials
DB_HOST=gateway01.your-region.prod.aws.tidbcloud.com
DB_PORT=4000
DB_USER=your_username
DB_PASSWORD=your_password
DB_NAME=artifacts_dbDatabase Setup
import mysql.connector
from dotenv import load_dotenv
import os
load_dotenv()
# Create database connection
conn = 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')
)
cursor = conn.cursor()
# Create tables
cursor.execute("""
CREATE TABLE IF NOT EXISTS artifactmetadata (
id INT PRIMARY KEY,
title VARCHAR(500),
culture VARCHAR(200),
century VARCHAR(100),
department VARCHAR(200),
classification VARCHAR(200),
dated VARCHAR(200),
url TEXT,
totalpageviews INT,
totaluniquepageviews INT
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS artifactmedia (
id INT AUTO_INCREMENT PRIMARY KEY,
artifact_id INT,
baseimageurl VARCHAR(500),
format VARCHAR(50),
height INT,
width INT,
FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS 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)
)
""")
conn.commit()Key API Patterns
Fetching Artifacts with Pagination
import requests
import os
from dotenv import load_dotenv
load_dotenv()
def fetch_artifacts(page=1, size=100):
"""Fetch artifacts from Harvard API with pagination"""
api_key = os.getenv('HARVARD_API_KEY')
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)
if response.status_code == 200:
data = response.json()
return data['records'], data['info']
else:
raise Exception(f"API Error: {response.status_code}")
# Fetch multiple pages
def fetch_all_artifacts(max_pages=10):
all_artifacts = []
for page in range(1, max_pages + 1):
artifacts, info = fetch_artifacts(page=page)
all_artifacts.extend(artifacts)
# Check if more pages exist
if page >= info['pages']:
break
return all_artifactsETL Pipeline Implementation
import pandas as pd
def extract_metadata(artifacts):
"""Extract artifact metadata"""
metadata = []
for artifact in artifacts:
metadata.append({
'id': artifact.get('id'),
'title': artifact.get('title', '')[:500],
'culture': artifact.get('culture', '')[:200],
'century': artifact.get('century', '')[:100],
'department': artifact.get('department', '')[:200],
'classification': artifact.get('classification', '')[:200],
'dated': artifact.get('dated', '')[:200],
'url': artifact.get('url', ''),
'totalpageviews': artifact.get('totalpageviews', 0),
'totaluniquepageviews': artifact.get('totaluniquepageviews', 0)
})
return pd.DataFrame(metadata)
def extract_media(artifacts):
"""Extract media information"""
media = []
for artifact in artifacts:
artifact_id = artifact.get('id')
primary_image = artifact.get('primaryimageurl')
if primary_image:
media.append({
'artifact_id': artifact_id,
'baseimageurl': primary_image,
'format': 'image',
'height': None,
'width': None
})
# Extract images from images array
for img in artifact.get('images', []):
media.append({
'artifact_id': artifact_id,
'baseimageurl': img.get('baseimageurl', ''),
'format': img.get('format', ''),
'height': img.get('height'),
'width': img.get('width')
})
return pd.DataFrame(media)
def extract_colors(artifacts):
"""Extract color information"""
colors = []
for artifact in artifacts:
artifact_id = artifact.get('id')
for color in artifact.get('colors', []):
colors.append({
'artifact_id': artifact_id,
'color': color.get('color', ''),
'spectrum': color.get('spectrum', ''),
'hue': color.get('hue', ''),
'percent': color.get('percent', 0.0)
})
return pd.DataFrame(colors)Loading Data to SQL
def load_to_sql(df, table_name, conn):
"""Batch insert DataFrame into SQL table"""
cursor = conn.cursor()
if df.empty:
return
# Build INSERT statement
columns = ', '.join(df.columns)
placeholders = ', '.join(['%s'] * len(df.columns))
insert_query = f"""
INSERT INTO {table_name} ({columns})
VALUES ({placeholders})
ON DUPLICATE KEY UPDATE
{', '.join([f"{col}=VALUES({col})" for col in df.columns if col != 'id'])}
"""
# Convert DataFrame to list of tuples
data = [tuple(row) for row in df.values]
# Batch insert
cursor.executemany(insert_query, data)
conn.commit()
print(f"Loaded {len(data)} records into {table_name}")
# Complete ETL process
def run_etl_pipeline(max_pages=5):
artifacts = fetch_all_artifacts(max_pages=max_pages)
# Extract
metadata_df = extract_metadata(artifacts)
media_df = extract_media(artifacts)
colors_df = extract_colors(artifacts)
# Load
load_to_sql(metadata_df, 'artifactmetadata', conn)
load_to_sql(media_df, 'artifactmedia', conn)
load_to_sql(colors_df, 'artifactcolors', conn)Streamlit Dashboard Implementation
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 Analytics")
# Sidebar for ETL operations
st.sidebar.header("Data Collection")
pages_to_fetch = st.sidebar.slider("Pages to fetch", 1, 50, 5)
if st.sidebar.button("Run ETL Pipeline"):
with st.spinner("Fetching and processing data..."):
run_etl_pipeline(max_pages=pages_to_fetch)
st.success("ETL completed successfully!")
# Analytics queries
st.header("📊 Analytics Dashboard")
queries = {
"Artifacts by Century": """
SELECT century, COUNT(*) as count
FROM artifactmetadata
WHERE century IS NOT NULL AND century != ''
GROUP BY century
ORDER BY count DESC
LIMIT 20
""",
"Artifacts by Culture": """
SELECT culture, COUNT(*) as count
FROM artifactmetadata
WHERE culture IS NOT NULL AND culture != ''
GROUP BY culture
ORDER BY count DESC
LIMIT 15
""",
"Color Distribution": """
SELECT color, COUNT(*) as count, AVG(percent) as avg_percent
FROM artifactcolors
GROUP BY color
ORDER BY count DESC
LIMIT 10
""",
"Media Availability": """
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
""",
"Top Viewed Artifacts": """
SELECT title, totalpageviews, culture, century
FROM artifactmetadata
WHERE totalpageviews > 0
ORDER BY totalpageviews DESC
LIMIT 10
"""
}
query_choice = st.selectbox("Select Analysis", list(queries.keys()))
if st.button("Run Query"):
cursor = conn.cursor()
cursor.execute(queries[query_choice])
results = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
df = pd.DataFrame(results, columns=columns)
st.subheader("Results")
st.dataframe(df)
# Auto-generate visualization
if len(df.columns) >= 2:
fig = px.bar(df, x=df.columns[0], y=df.columns[1],
title=query_choice)
st.plotly_chart(fig, use_container_width=True)Common Analytical Queries
# Query: Artifacts with most colors
"""
SELECT a.title, COUNT(c.id) as color_count
FROM artifactmetadata a
JOIN artifactcolors c ON a.id = c.artifact_id
GROUP BY a.id, a.title
ORDER BY color_count DESC
LIMIT 10
"""
# Query: Department breakdown
"""
SELECT department, COUNT(*) as count,
AVG(totalpageviews) as avg_views
FROM artifactmetadata
WHERE department IS NOT NULL
GROUP BY department
ORDER BY count DESC
"""
# Query: Dominant color per culture
"""
SELECT culture, color, SUM(percent) as total_percent
FROM artifactmetadata a
JOIN artifactcolors c ON a.id = c.artifact_id
WHERE culture IS NOT NULL
GROUP BY culture, color
ORDER BY culture, total_percent DESC
"""Troubleshooting
API Rate Limiting
import time
def fetch_with_retry(page, max_retries=3):
for attempt in range(max_retries):
try:
return fetch_artifacts(page=page)
except Exception as e:
if "429" in str(e): # Rate limit
wait_time = 2 ** attempt
time.sleep(wait_time)
else:
raise
raise Exception("Max retries exceeded")Database Connection Issues
def get_connection():
"""Create database connection with error handling"""
try:
conn = 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'),
connect_timeout=10
)
return conn
except mysql.connector.Error as err:
st.error(f"Database connection failed: {err}")
return NoneHandling Missing Data
def safe_extract(artifact, key, default=''):
"""Safely extract nested data"""
value = artifact.get(key, default)
return value if value is not None else default
# Use in extraction
metadata.append({
'id': artifact.get('id'),
'title': safe_extract(artifact, 'title', 'Unknown'),
'culture': safe_extract(artifact, 'culture', 'Unknown'),
# ...
})Running the Application
# Start the Streamlit app
streamlit run app.py
# The app will open at http://localhost:8501This skill provides comprehensive guidance for building ETL pipelines with museum APIs, implementing SQL analytics, and creating interactive dashboards with Streamlit.