| name | harvard-art-museums-data-pipeline |
| description | Build end-to-end data engineering pipelines with the Harvard Art Museums API, ETL processes, SQL analytics, and Streamlit visualization |
| triggers | ["build a data pipeline for museum artifacts","create ETL workflow for Harvard Art Museums API","set up artifact data collection and analytics","implement museum data engineering pipeline","build streamlit dashboard for art museum data","create SQL analytics for artifact collections","extract and analyze Harvard museum artifacts","design data warehouse for art collections"] |
Harvard Art Museums Data Pipeline
Skill by ara.so — Data Skills collection.
This skill enables you to build production-ready data engineering pipelines using the Harvard Art Museums API. It covers ETL workflows, relational database design, SQL analytics, and interactive Streamlit dashboards for artifact data visualization.
What This Project Does
The Harvard-Artifacts-Collection-Data-Engineering-Analytics-App demonstrates a complete data pipeline:
- Extract: Fetches artifact data from Harvard Art Museums API with pagination and rate limiting
- Transform: Converts nested JSON into normalized relational tables
- Load: Batch inserts data into MySQL/TiDB Cloud databases
- Analyze: Executes analytical SQL queries for insights
- Visualize: Renders interactive dashboards with Plotly and Streamlit
The architecture follows: API → ETL → SQL → Analytics → Visualization
Installation
git clone https://github.com/Manali0711/Harvard-Artifacts-Collection-Data-Engineering-Analytics-App.git
cd Harvard-Artifacts-Collection-Data-Engineering-Analytics-App
pip install -r requirements.txt
Dependencies
streamlit
pandas
requests
mysql-connector-python
plotly
python-dotenv
Configuration
Environment Variables
Create a .env file in the project root:
HARVARD_API_KEY=your_api_key_here
DB_HOST=your_database_host
DB_PORT=3306
DB_USER=your_db_user
DB_PASSWORD=your_db_password
DB_NAME=harvard_artifacts
Get Harvard API Key
- Visit Harvard Art Museums API
- Register for a free API key
- Add to
.env file
Database Setup
import mysql.connector
from dotenv import load_dotenv
import os
load_dotenv()
def get_db_connection():
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 setup_database():
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS artifactmetadata (
id INT PRIMARY KEY,
title VARCHAR(500),
culture VARCHAR(255),
century VARCHAR(100),
classification VARCHAR(255),
department VARCHAR(255),
technique VARCHAR(255),
medium VARCHAR(500),
dated VARCHAR(255),
url TEXT,
totalpageviews INT,
totaluniquepageviews INT
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS artifactmedia (
id INT AUTO_INCREMENT PRIMARY KEY,
artifact_id INT,
iiifbaseuri VARCHAR(500),
baseimageurl TEXT,
primaryimageurl TEXT,
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 DECIMAL(5,2),
FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
)
""")
conn.commit()
cursor.close()
conn.close()
API Integration
Basic API Request
import requests
import os
from dotenv import load_dotenv
load_dotenv()
def fetch_artifacts(page=1, size=100):
"""Fetch artifacts from Harvard Art Museums API"""
api_key = os.getenv('HARVARD_API_KEY')
base_url = "https://api.harvardartmuseums.org/object"
params = {
'apikey': api_key,
'page': page,
'size': size,
'hasimage': 1
}
response = requests.get(base_url, params=params)
response.raise_for_status()
return response.json()
Paginated Data Collection
def collect_all_artifacts(max_records=1000):
"""Collect artifacts with pagination handling"""
all_artifacts = []
page = 1
size = 100
while len(all_artifacts) < max_records:
try:
data = fetch_artifacts(page=page, size=size)
records = data.get('records', [])
if not records:
break
all_artifacts.extend(records)
if data['info']['next'] is None:
break
page += 1
import time
time.sleep(0.5)
except Exception as e:
print(f"Error fetching page {page}: {e}")
break
return all_artifacts[:max_records]
ETL Pipeline
Extract and Transform
import pandas as pd
def transform_artifact_metadata(artifacts):
"""Transform artifacts into metadata DataFrame"""
metadata = []
for artifact in artifacts:
metadata.append({
'id': artifact.get('id'),
'title': artifact.get('title', 'Unknown')[:500],
'culture': artifact.get('culture', 'Unknown')[:255],
'century': artifact.get('century', 'Unknown')[:100],
'classification': artifact.get('classification', 'Unknown')[:255],
'department': artifact.get('department', 'Unknown')[:255],
'technique': artifact.get('technique', 'Unknown')[:255],
'medium': artifact.get('medium', 'Unknown')[:500],
'dated': artifact.get('dated', 'Unknown')[:255],
'url': artifact.get('url', ''),
'totalpageviews': artifact.get('totalpageviews', 0),
'totaluniquepageviews': artifact.get('totaluniquepageviews', 0)
})
return pd.DataFrame(metadata)
def transform_artifact_media(artifacts):
"""Transform artifacts into media DataFrame"""
media = []
for artifact in artifacts:
artifact_id = artifact.get('id')
images = artifact.get('images', [])
if images:
primary_image = images[0]
media.append({
'artifact_id': artifact_id,
'iiifbaseuri': primary_image.get('iiifbaseuri', ''),
'baseimageurl': primary_image.get('baseimageurl', ''),
'primaryimageurl': artifact.get('primaryimageurl', '')
})
return pd.DataFrame(media)
def transform_artifact_colors(artifacts):
"""Transform artifacts into colors DataFrame"""
colors = []
for artifact in artifacts:
artifact_id = artifact.get('id')
color_list = artifact.get('colors', [])
for color in color_list:
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)
Load into Database
def load_metadata(df, conn):
"""Batch insert metadata into database"""
cursor = conn.cursor()
insert_query = """
INSERT INTO artifactmetadata
(id, title, culture, century, classification, department,
technique, medium, dated, url, totalpageviews, totaluniquepageviews)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
title=VALUES(title), culture=VALUES(culture)
"""
data = [tuple(row) for row in df.values]
cursor.executemany(insert_query, data)
conn.commit()
cursor.close()
def run_etl_pipeline(max_records=1000):
"""Execute complete ETL pipeline"""
print("Extracting artifacts from API...")
artifacts = collect_all_artifacts(max_records)
print("Transforming data...")
df_metadata = transform_artifact_metadata(artifacts)
df_media = transform_artifact_media(artifacts)
df_colors = transform_artifact_colors(artifacts)
print("Loading into database...")
conn = get_db_connection()
load_metadata(df_metadata, conn)
load_metadata(df_media, conn)
load_metadata(df_colors, conn)
conn.close()
print(f"ETL complete: {len(artifacts)} artifacts processed")
SQL Analytics
Sample Analytical Queries
QUERY_BY_CULTURE = """
SELECT culture, COUNT(*) as count
FROM artifactmetadata
WHERE culture != 'Unknown'
GROUP BY culture
ORDER BY count DESC
LIMIT 20
"""
QUERY_TOP_VIEWED = """
SELECT title, culture, totalpageviews
FROM artifactmetadata
ORDER BY totalpageviews DESC
LIMIT 10
"""
QUERY_BY_CENTURY = """
SELECT century, COUNT(*) as count
FROM artifactmetadata
WHERE century != 'Unknown'
GROUP BY century
ORDER BY count DESC
"""
QUERY_COLOR_DISTRIBUTION = """
SELECT color, COUNT(*) as count, AVG(percent) as avg_percent
FROM artifactcolors
GROUP BY color
ORDER BY count DESC
LIMIT 15
"""
QUERY_DEPARTMENT_STATS = """
SELECT
department,
COUNT(*) as total_artifacts,
AVG(totalpageviews) as avg_views
FROM artifactmetadata
WHERE department != 'Unknown'
GROUP BY department
ORDER BY total_artifacts DESC
"""
def execute_query(query):
"""Execute SQL query and return DataFrame"""
conn = get_db_connection()
df = pd.read_sql(query, conn)
conn.close()
return df
Streamlit Dashboard
Main Application
import streamlit as st
import plotly.express as px
def main():
st.set_page_config(
page_title="Harvard Artifacts Analytics",
page_icon="🏛️",
layout="wide"
)
st.title("🏛️ Harvard Art Museums Analytics Dashboard")
st.sidebar.header("Navigation")
page = st.sidebar.selectbox(
"Choose a page",
["Data Collection", "SQL Analytics", "Visualizations"]
)
if page == "Data Collection":
show_data_collection()
elif page == "SQL Analytics":
show_sql_analytics()
elif page == "Visualizations":
show_visualizations()
def show_data_collection():
st.header("📥 Data Collection")
num_records = st.number_input(
"Number of records to collect",
min_value=100,
max_value=10000,
value=1000,
step=100
)
if st.button("Start ETL Pipeline"):
with st.spinner("Running ETL pipeline..."):
run_etl_pipeline(num_records)
st.success(f"Successfully collected {num_records} artifacts!")
def show_sql_analytics():
st.header("📊 SQL Analytics")
queries = {
"Artifacts by Culture": QUERY_BY_CULTURE,
"Most Viewed Artifacts": QUERY_TOP_VIEWED,
"Artifacts by Century": QUERY_BY_CENTURY,
"Color Distribution": QUERY_COLOR_DISTRIBUTION,
"Department Statistics": QUERY_DEPARTMENT_STATS
}
selected_query = st.selectbox("Select Query", list(queries.keys()))
if st.button("Execute Query"):
df = execute_query(queries[selected_query])
st.dataframe(df, use_container_width=True)
if len(df.columns) >= 2:
fig = px.bar(
df,
x=df.columns[0],
y=df.columns[1],
title=selected_query
)
st.plotly_chart(fig, use_container_width=True)
if __name__ == "__main__":
main()
Run the Dashboard
streamlit run app.py
Common Patterns
Rate Limiting API Requests
import time
from functools import wraps
def rate_limit(calls_per_second=2):
"""Decorator to rate limit API calls"""
min_interval = 1.0 / calls_per_second
last_called = [0.0]
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() - last_called[0]
left_to_wait = min_interval - elapsed
if left_to_wait > 0:
time.sleep(left_to_wait)
ret = func(*args, **kwargs)
last_called[0] = time.time()
return ret
return wrapper
return decorator
@rate_limit(calls_per_second=2)
def fetch_artifact_by_id(artifact_id):
"""Fetch single artifact with rate limiting"""
api_key = os.getenv('HARVARD_API_KEY')
url = f"https://api.harvardartmuseums.org/object/{artifact_id}"
response = requests.get(url, params={'apikey': api_key})
return response.json()
Error Handling and Retries
from requests.exceptions import RequestException
def fetch_with_retry(url, params, max_retries=3):
"""Fetch with exponential backoff retry"""
for attempt in range(max_retries):
try:
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
return response.json()
except RequestException as e:
if attempt == max_retries - 1:
raise
wait_time = 2 ** attempt
print(f"Retry {attempt + 1}/{max_retries} after {wait_time}s")
time.sleep(wait_time)
Troubleshooting
API Key Issues
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv('HARVARD_API_KEY')
if not api_key:
raise ValueError("HARVARD_API_KEY not found in environment variables")
response = requests.get(
"https://api.harvardartmuseums.org/object",
params={'apikey': api_key, 'size': 1}
)
print(f"API Status: {response.status_code}")
Database Connection Errors
def test_database_connection():
"""Test database connectivity"""
try:
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT 1")
result = cursor.fetchone()
cursor.close()
conn.close()
print("Database connection successful")
return True
except Exception as e:
print(f"Database connection failed: {e}")
return False
Memory Issues with Large Datasets
def batch_process_artifacts(total_records, batch_size=100):
"""Process artifacts in batches to manage memory"""
num_batches = (total_records + batch_size - 1) // batch_size
for batch_num in range(num_batches):
page = batch_num + 1
artifacts = fetch_artifacts(page=page, size=batch_size)
df = transform_artifact_metadata(artifacts['records'])
conn = get_db_connection()
load_metadata(df, conn)
conn.close()
print(f"Processed batch {batch_num + 1}/{num_batches}")
Streamlit Caching for Performance
@st.cache_data(ttl=3600)
def cached_query_execution(query):
"""Cache query results for 1 hour"""
return execute_query(query)
df = cached_query_execution(QUERY_BY_CULTURE)
st.dataframe(df)