Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
Build end-to-end ETL pipelines and analytics dashboards using the Harvard Art Museums API with Python, SQL, and Streamlit
triggers
["how do I build a data pipeline with Harvard Art Museums API","create an ETL pipeline for museum artifacts data","set up analytics dashboard for Harvard art collection","extract and transform Harvard museum data into SQL","build Streamlit app for art artifacts analytics","visualize Harvard Art Museums data with SQL queries","implement artifact collection data engineering pipeline","analyze museum artifact metadata with Python and SQL"]
This project provides an end-to-end data engineering and analytics application for the Harvard Art Museums API. It demonstrates real-world ETL pipelines, SQL database design, analytical queries, and interactive visualization using Streamlit.
What This Project Does
The application implements a complete data pipeline:
Extract: Fetches artifact data from Harvard Art Museums API with pagination and rate limiting
import streamlit as st
import pandas as pd
import plotly.express as px
import mysql.connector
import os
st.set_page_config(page_title="Harvard Artifacts Analytics", layout="wide")
# Sidebar configuration
st.sidebar.title("Harvard Art Museums Analytics")
st.sidebar.markdown("### Configuration")
# Database connection@st.cache_resourcedefget_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')
)
defexecute_query(query):
"""Execute SQL query and return results as DataFrame"""
conn = get_database_connection()
df = pd.read_sql(query, conn)
return df
# Main content
st.title("🎨 Harvard Art Museums Collection Analytics")
# ETL Pipeline Sectionif st.sidebar.button("Run ETL Pipeline"):
with st.spinner("Fetching data from API..."):
api_key = os.getenv('HARVARD_API_KEY')
artifacts = extract_all_artifacts(api_key, max_pages=5)
st.success(f"Extracted {len(artifacts)} artifacts")
with st.spinner("Transforming data..."):
metadata_df, media_df, colors_df = transform_artifacts(artifacts)
st.success("Data transformation complete")
with st.spinner("Loading to database..."):
db_config = {
'host': os.getenv('DB_HOST'),
'user': os.getenv('DB_USER'),
'password': os.getenv('DB_PASSWORD'),
'database': os.getenv('DB_NAME')
}
load_to_database(metadata_df, media_df, colors_df, db_config)
st.success("Data loaded successfully!")
Analytics Queries
# Predefined analytical queries
QUERIES = {
"Artifacts by Culture": """
SELECT culture, COUNT(*) as artifact_count
FROM artifactmetadata
WHERE culture IS NOT NULL
GROUP BY culture
ORDER BY artifact_count DESC
LIMIT 15
""",
"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, COUNT(*) as frequency, AVG(percentage) as avg_percentage
FROM artifactcolors
GROUP BY color
ORDER BY frequency DESC
LIMIT 10
""",
"Media Availability": """
SELECT
has_image,
COUNT(*) as count,
ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 2) as percentage
FROM artifactmedia
GROUP BY has_image
""",
"Artifacts by Department": """
SELECT department, COUNT(*) as total_artifacts
FROM artifactmetadata
WHERE department IS NOT NULL
GROUP BY department
ORDER BY total_artifacts DESC
"""
}
# Query selector
query_name = st.selectbox("Select Analysis", list(QUERIES.keys()))
if st.button("Run Query"):
df = execute_query(QUERIES[query_name])
# Display results
st.dataframe(df)
# Auto-generate visualizationiflen(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
Pattern 1: Incremental Data Loading
defget_latest_artifact_id(cursor):
"""Get the most recent artifact ID in database"""
cursor.execute("SELECT MAX(id) FROM artifactmetadata")
result = cursor.fetchone()
return result[0] if result[0] else0defincremental_etl(api_key, db_config):
"""Load only new artifacts since last ETL run"""
conn = mysql.connector.connect(**db_config)
cursor = conn.cursor()
latest_id = get_latest_artifact_id(cursor)
# Fetch only newer artifacts
artifacts = fetch_artifacts(api_key, size=100)
new_artifacts = [a for a in artifacts if a.get('id', 0) > latest_id]
if new_artifacts:
metadata_df, media_df, colors_df = transform_artifacts(new_artifacts)
load_to_database(metadata_df, media_df, colors_df, db_config)
cursor.close()
conn.close()
Pattern 2: Error Handling and Logging
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
defsafe_fetch_artifacts(api_key, page=1, max_retries=3):
"""Fetch with retry logic"""for attempt inrange(max_retries):
try:
data = fetch_artifacts(api_key, page)
logger.info(f"Successfully fetched page {page}")
return data
except requests.RequestException as e:
logger.error(f"Attempt {attempt + 1} failed: {e}")
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # Exponential backoff
Running the Application
# Start the Streamlit app
streamlit run app.py
# Access the dashboard at http://localhost:8501
Troubleshooting
API Rate Limiting: If you encounter 429 errors, increase the sleep time between requests:
time.sleep(1) # Increase from 0.5 to 1 second
Database Connection Issues: Verify environment variables are set:
import os
print(f"DB Host: {os.getenv('DB_HOST')}")
print(f"DB User: {os.getenv('DB_USER')}")
Missing Data Fields: Handle None values in transformations:
