SKILL.md
Harvard Art Museums Data Engineering & Analytics
Skill by ara.so — Data Skills collection.
This project demonstrates a complete data engineering workflow: extracting data from the Harvard Art Museums API, transforming it into relational tables, loading into SQL databases, and building interactive analytics dashboards with Streamlit.
What This Project Does
- API Integration: Fetches artifact data from Harvard Art Museums API with pagination and rate limiting
- ETL Pipeline: Transforms nested JSON into structured relational tables (metadata, media, colors)
- SQL Storage: Loads data into MySQL/TiDB Cloud with proper schema design
- Analytics: Executes 20+ predefined analytical queries
- Visualization: Interactive Plotly dashboards for data exploration
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"
export DB_HOST="your_db_host"
export DB_USER="your_db_user"
export DB_PASSWORD="your_db_password"
export DB_NAME="harvard_artifacts"Requirements typically include:
streamlit
pandas
requests
mysql-connector-python
plotly
python-dotenvAPI Integration
Getting API Key
- Register at Harvard Art Museums API
- Store key in environment variable or
.envfile
Fetching Artifacts
import requests
import os
API_KEY = os.getenv("HARVARD_API_KEY")
BASE_URL = "https://api.harvardartmuseums.org/object"
def fetch_artifacts(size=100, page=1):
"""Fetch artifacts with pagination"""
params = {
"apikey": API_KEY,
"size": size,
"page": page,
"hasimage": 1 # Only artifacts with images
}
response = requests.get(BASE_URL, params=params)
response.raise_for_status()
return response.json()
# Example usage
data = fetch_artifacts(size=50, page=1)
artifacts = data.get("records", [])
total_records = data.get("info", {}).get("totalrecords", 0)
print(f"Fetched {len(artifacts)} of {total_records} artifacts")Handling Pagination
def fetch_all_artifacts(max_records=500):
"""Fetch multiple pages of artifacts"""
all_artifacts = []
page = 1
size = 100
while len(all_artifacts) < max_records:
data = fetch_artifacts(size=size, page=page)
records = data.get("records", [])
if not records:
break
all_artifacts.extend(records)
page += 1
# Respect rate limits
import time
time.sleep(0.5)
return all_artifacts[:max_records]ETL Pipeline
Extract: Parse API Response
import pandas as pd
def extract_metadata(artifacts):
"""Extract core artifact metadata"""
metadata_list = []
for artifact in artifacts:
metadata = {
"object_id": artifact.get("objectid"),
"title": artifact.get("title"),
"culture": artifact.get("culture"),
"period": artifact.get("period"),
"century": artifact.get("century"),
"classification": artifact.get("classification"),
"department": artifact.get("department"),
"dated": artifact.get("dated"),
"division": artifact.get("division")
}
metadata_list.append(metadata)
return pd.DataFrame(metadata_list)
def extract_media(artifacts):
"""Extract media/image information"""
media_list = []
for artifact in artifacts:
object_id = artifact.get("objectid")
images = artifact.get("images", [])
for img in images:
media = {
"object_id": object_id,
"image_id": img.get("imageid"),
"base_url": img.get("baseimageurl"),
"width": img.get("width"),
"height": img.get("height"),
"format": img.get("format")
}
media_list.append(media)
return pd.DataFrame(media_list)
def extract_colors(artifacts):
"""Extract color palette information"""
color_list = []
for artifact in artifacts:
object_id = artifact.get("objectid")
colors = artifact.get("colors", [])
for color in colors:
color_data = {
"object_id": object_id,
"color": color.get("color"),
"spectrum": color.get("spectrum"),
"hue": color.get("hue"),
"percent": color.get("percent")
}
color_list.append(color_data)
return pd.DataFrame(color_list)Transform: Clean and Validate
def transform_metadata(df):
"""Clean and transform metadata"""
# Remove duplicates
df = df.drop_duplicates(subset=["object_id"])
# Handle nulls
df = df.fillna("")
# Truncate long text fields
df["title"] = df["title"].str[:255]
df["culture"] = df["culture"].str[:100]
return df
def transform_media(df):
"""Clean media data"""
# Remove rows without image_id
df = df.dropna(subset=["image_id"])
# Convert dimensions to integers
df["width"] = pd.to_numeric(df["width"], errors="coerce").fillna(0).astype(int)
df["height"] = pd.to_numeric(df["height"], errors="coerce").fillna(0).astype(int)
return dfLoad: Insert into SQL Database
import mysql.connector
from mysql.connector import Error
def get_db_connection():
"""Create database connection"""
return mysql.connector.connect(
host=os.getenv("DB_HOST"),
user=os.getenv("DB_USER"),
password=os.getenv("DB_PASSWORD"),
database=os.getenv("DB_NAME")
)
def create_tables(connection):
"""Create database schema"""
cursor = connection.cursor()
# Metadata table
cursor.execute("""
CREATE TABLE IF NOT EXISTS artifactmetadata (
object_id INT PRIMARY KEY,
title VARCHAR(255),
culture VARCHAR(100),
period VARCHAR(100),
century VARCHAR(50),
classification VARCHAR(100),
department VARCHAR(100),
dated VARCHAR(100),
division VARCHAR(100)
)
""")
# Media table
cursor.execute("""
CREATE TABLE IF NOT EXISTS artifactmedia (
id INT AUTO_INCREMENT PRIMARY KEY,
object_id INT,
image_id INT,
base_url TEXT,
width INT,
height INT,
format VARCHAR(50),
FOREIGN KEY (object_id) REFERENCES artifactmetadata(object_id)
)
""")
# Colors table
cursor.execute("""
CREATE TABLE IF NOT EXISTS artifactcolors (
id INT AUTO_INCREMENT PRIMARY KEY,
object_id INT,
color VARCHAR(50),
spectrum VARCHAR(50),
hue VARCHAR(50),
percent FLOAT,
FOREIGN KEY (object_id) REFERENCES artifactmetadata(object_id)
)
""")
connection.commit()
cursor.close()
def load_to_database(df, table_name, connection):
"""Batch insert dataframe into database"""
cursor = connection.cursor()
# Prepare insert statement
cols = ", ".join(df.columns)
placeholders = ", ".join(["%s"] * len(df.columns))
insert_sql = f"INSERT IGNORE INTO {table_name} ({cols}) VALUES ({placeholders})"
# Convert dataframe to list of tuples
data = [tuple(row) for row in df.values]
# Batch insert
cursor.executemany(insert_sql, data)
connection.commit()
cursor.close()
print(f"Inserted {cursor.rowcount} rows into {table_name}")Streamlit Analytics Dashboard
Main Application Structure
import streamlit as st
import plotly.express as px
def main():
st.set_page_config(page_title="Harvard Artifacts Analytics", layout="wide")
st.title("🏛️ Harvard Art Museums Analytics Dashboard")
# Sidebar for navigation
page = st.sidebar.selectbox(
"Select Feature",
["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 fetch", 10, 1000, 100)
if st.button("Fetch Data"):
with st.spinner("Fetching artifacts..."):
artifacts = fetch_all_artifacts(max_records=num_records)
# ETL process
metadata_df = transform_metadata(extract_metadata(artifacts))
media_df = transform_media(extract_media(artifacts))
colors_df = extract_colors(artifacts)
# Load to database
conn = get_db_connection()
create_tables(conn)
load_to_database(metadata_df, "artifactmetadata", conn)
load_to_database(media_df, "artifactmedia", conn)
load_to_database(colors_df, "artifactcolors", conn)
conn.close()
st.success(f"✅ Loaded {len(metadata_df)} artifacts into database")
if __name__ == "__main__":
main()SQL Analytics Queries
ANALYTICS_QUERIES = {
"Artifacts by Century": """
SELECT century, COUNT(*) as count
FROM artifactmetadata
WHERE century != ''
GROUP BY century
ORDER BY count DESC
LIMIT 10
""",
"Top Cultures": """
SELECT culture, COUNT(*) as artifact_count
FROM artifactmetadata
WHERE culture != ''
GROUP BY culture
ORDER BY artifact_count DESC
LIMIT 15
""",
"Department Distribution": """
SELECT department, COUNT(*) as count
FROM artifactmetadata
WHERE department != ''
GROUP BY department
ORDER BY count DESC
""",
"Color Palette Analysis": """
SELECT color, COUNT(*) as frequency, AVG(percent) as avg_percent
FROM artifactcolors
GROUP BY color
ORDER BY frequency DESC
LIMIT 20
""",
"Image Dimensions Analysis": """
SELECT
CASE
WHEN width < 500 THEN 'Small'
WHEN width < 1000 THEN 'Medium'
ELSE 'Large'
END as size_category,
COUNT(*) as count
FROM artifactmedia
GROUP BY size_category
"""
}
def show_sql_analytics():
st.header("📊 SQL Analytics")
query_name = st.selectbox("Select Analysis", list(ANALYTICS_QUERIES.keys()))
if st.button("Run Query"):
conn = get_db_connection()
query = ANALYTICS_QUERIES[query_name]
df = pd.read_sql(query, conn)
conn.close()
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_name)
st.plotly_chart(fig, use_container_width=True)Common Patterns
Environment Configuration
from dotenv import load_dotenv
import os
load_dotenv()
CONFIG = {
"api_key": os.getenv("HARVARD_API_KEY"),
"db_host": os.getenv("DB_HOST"),
"db_user": os.getenv("DB_USER"),
"db_password": os.getenv("DB_PASSWORD"),
"db_name": os.getenv("DB_NAME", "harvard_artifacts")
}Error Handling
def safe_api_call(url, params, max_retries=3):
"""API call 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:
st.error(f"API call failed: {e}")
return None
time.sleep(2 ** attempt) # Exponential backoffTroubleshooting
API Rate Limiting: Add delays between requests
import time
time.sleep(0.5) # 500ms between requestsDatabase Connection Issues: Check credentials and network
try:
conn = get_db_connection()
conn.ping(reconnect=True)
except Error as e:
st.error(f"Database connection failed: {e}")Missing Data in API Response: Always use .get() with defaults
title = artifact.get("title", "Untitled")
images = artifact.get("images", [])Memory Issues with Large Datasets: Use chunking
chunk_size = 100
for i in range(0, len(data), chunk_size):
chunk = data[i:i+chunk_size]
load_to_database(pd.DataFrame(chunk), table_name, conn)