harvard-artifacts-etl-streamlit-analytics

# Harvard Artifacts ETL & Analytics Skill > Skill by [ara.so](https://ara.so) — Data Skills collection. This skill enables AI coding agents to help developers build end-to-end data engineering and analytics applications using the Harvard Art Museums API. The project demonstrates ETL pipelines, SQL database design, analytical queries, and interactive Streamlit visualizations for museum artifact data. ## What This Project Does The Harvard Artifacts Collection Data Engineering & Analytics App provides: - **API Integration**: Fetch artifact data from Harvard Art Museums API with pagination and rate limiting - **ETL Pipeline**: Extract, transform, and load artifact metadata, media, and color data into relational SQL tables - **SQL Analytics**: Pre-built analytical queries for insights on culture, century, media availability, and color patterns - **Interactive Dashboard**: Streamlit-based UI with Plotly visualizations for query results ## Installation ### Prerequisites - Python 3.8+ - MySQL or TiDB Cloud database - Harvard Art Museums API key (get from https://docs.harvardartmuseums.org/api-docs/) ### Setup Steps ```bash # 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_here" export DB_HOST="your_db_host" export DB_USER="your_db_user" export DB_PASSWORD="your_db_password" export DB_NAME="harvard_artifacts" # Run the Streamlit app streamlit run app.py ``` ### Dependencies (requirements.txt) ```txt streamlit pandas requests mysql-connector-python plotly python-dotenv ``` ## Configuration ### Database Connection ```python import mysql.connector import os def get_db_connection(): """Create database connection using environment variables""" return mysql.connector.connect( host=os.getenv('DB_HOST', 'localhost'), user=os.getenv('DB_USER', 'root'), password=os.getenv('DB_PASSWORD'), database=os.getenv('DB_NAME', 'harvard_artifacts'), port=int(os.getenv('DB_PORT', 3306)) ) ``` ### API Configuration ```python import os API_KEY = os.getenv('HARVARD_API_KEY') BASE_URL = "https://api.harvardartmuseums.org/object" ``` ## Database Schema ### Create Tables ```sql -- Artifact Metadata Table CREATE TABLE artifactmetadata ( id INT PRIMARY KEY, title VARCHAR(500), culture VARCHAR(255), century VARCHAR(100), classification VARCHAR(255), division VARCHAR(255), department VARCHAR(255), dated VARCHAR(255), accessionyear INT, url VARCHAR(500) ); -- Artifact Media Table CREATE TABLE artifactmedia ( media_id INT AUTO_INCREMENT PRIMARY KEY, artifact_id INT, baseimageurl VARCHAR(500), iiifbaseuri VARCHAR(500), primaryimageurl VARCHAR(500), FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id) ); -- Artifact Colors Table CREATE TABLE artifactcolors ( color_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) ); ``` ## ETL Pipeline Implementation ### Extract: Fetch Data from API ```python import requests import time def fetch_artifacts(api_key, num_pages=5, page_size=100): """ Fetch artifact data from Harvard Art Museums API with pagination Args: api_key: Harvard API key num_pages: Number of pages to fetch page_size: Records per page (max 100) Returns: List of artifact records """ artifacts = [] for page in range(1, num_pages + 1): params = { 'apikey': api_key, 'size': page_size, 'page': page } response = requests.get(BASE_URL, params=params) if response.status_code == 200: data = response.json() artifacts.extend(data.get('records', [])) print(f"Fetched page {page}/{num_pages}") # Rate limiting: API allows 2500 requests/day time.sleep(0.5) else: print(f"Error fetching page {page}: {response.status_code}") break return artifacts ``` ### Transform: Clean and Structure Data ```python import pandas as pd def transform_artifacts(raw_data): """ Transform raw API data into structured dataframes Returns: Tuple of (metadata_df, media_df, colors_df) """ metadata_list = [] media_list = [] colors_list = [] for artifact in raw_data: # Extract metadata metadata = { 'id': artifact.get('id'), 'title': artifact.get('title', '')[:500], 'culture': artifact.get('culture', '')[:255], 'century': artifact.get('century', '')[:100], 'classification': artifact.get('classification', '')[:255], 'division': artifact.get('division', '')[:255], 'department': artifact.get('department', '')[:255], 'dated': artifact.get('dated', '')[:255], 'accessionyear': artifact.get('accessionyear'), 'url': artifact.get('url', '')[:500] } metadata_list.append(metadata) # Extract media information if artifact.get('primaryimageurl') or artifact.get('baseimageurl'): media = { 'artifact_id': artifact.get('id'), 'baseimageurl': artifact.get('baseimageurl', '')[:500], 'iiifbaseuri': artifact.get('iiifbaseuri', '')[:500], 'primaryimageurl': artifact.get('primaryimageurl', '')[:500] } media_list.append(media) # Extract color data for color_obj in artifact.get('colors', []): color = { 'artifact_id': artifact.get('id'), 'color': color_obj.get('color', '')[:50], 'spectrum': color_obj.get('spectrum', '')[:50], 'hue': color_obj.get('hue', '')[:50], 'percent': color_obj.get('percent') } colors_list.append(color) return ( pd.DataFrame(metadata_list), pd.DataFrame(media_list), pd.DataFrame(colors_list) ) ``` ### Load: Insert into SQL Database ```python def load_to_database(metadata_df, media_df, colors_df, connection): """ Load transformed data into SQL database using batch inserts """ cursor = connection.cursor() # Insert metadata metadata_query = """ INSERT INTO artifactmetadata (id, title, culture, century, classification, division, department, dated, accessionyear, url) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE title=VALUES(title) """ cursor.executemany(metadata_query, metadata_df.values.tolist()) # Insert media media_query = """ INSERT INTO artifactmedia (artifact_id, baseimageurl, iiifbaseuri, primaryimageurl) VALUES (%s, %s, %s, %s) """ cursor.executemany(media_query, media_df.values.tolist()) # Insert colors colors_query = """ INSERT INTO artifactcolors (artifact_id, color, spectrum, hue, percent) VALUES (%s, %s, %s, %s, %s) """ cursor.executemany(colors_query, colors_df.values.tolist()) connection.commit() cursor.close() print(f"Loaded {len(metadata_df)} artifacts, {len(media_df)} media records, {len(colors_df)} color records") ``` ## Analytics Queries ### Sample Analytical Queries ```python ANALYTICS_QUERIES = { "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 20 """, "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 15 """, "Media Availability": """ SELECT COUNT(DISTINCT m.artifact_id) as with_media, (SELECT COUNT(*) FROM artifactmetadata) as total_artifacts, ROUND(COUNT(DISTINCT m.artifact_id) * 100.0 / (SELECT COUNT(*) FROM artifactmetadata), 2) as percentage FROM artifactmedia m """, "Top Colors Used": """ SELECT color, COUNT(*) as count, AVG(percent) as avg_percent FROM artifactcolors WHERE color IS NOT NULL GROUP BY color ORDER BY count DESC LIMIT 20 """, "Department Distribution": """ SELECT department, COUNT(*) as count FROM artifactmetadata WHERE department IS NOT NULL AND department != '' GROUP BY department ORDER BY count DESC """, "Artifacts with Images by Century": """ SELECT am.century, COUNT(DISTINCT am.id) as artifact_count FROM artifactmetadata am JOIN artifactmedia media ON am.id = media.artifact_id WHERE am.century IS NOT NULL AND media.primaryimageurl IS NOT NULL GROUP BY am.century ORDER BY artifact_count DESC LIMIT 15 """ } def execute_query(query, connection): """Execute SQL query and return DataFrame""" return pd.read_sql(query, connection) ``` ## Streamlit Dashboard Implementation ### Basic App Structure ```python import streamlit as st import plotly.express as px def main(): st.title("🏛️ Harvard Art Museums Analytics Dashboard") # Sidebar for navigation page = st.sidebar.selectbox( "Select Page", ["Data Collection", "Analytics", "Visualizations"] ) if page == "Data Collection": show_data_collection_page() elif page == "Analytics": show_analytics_page() else: show_visualizations_page() def show_data_collection_page(): """Page for ETL operations""" st.header("Data Collection & ETL") api_key = st.text_input("Harvard API Key", type="password", value=os.getenv('HARVARD_API_KEY', '')) num_pages = st.slider("Number of pages to fetch", 1, 10, 5) if st.button("Run ETL Pipeline"): with st.spinner("Fetching data from API..."): raw_data = fetch_artifacts(api_key, num_pages) st.success(f"Fetched {len(raw_data)} artifacts") with st.spinner("Transforming data..."): metadata_df, media_df, colors_df = transform_artifacts(raw_data) st.success("Data transformed") with st.spinner("Loading to database..."): conn = get_db_connection() load_to_database(metadata_df, media_df, colors_df, conn) conn.close() st.success("Data loaded to database") def show_analytics_page(): """Page for running SQL queries""" st.header("SQL Analytics") query_name = st.selectbox("Select Query", list(ANALYTICS_QUERIES.keys())) if st.button("Run Query"): conn = get_db_connection() df = execute_query(ANALYTICS_QUERIES[query_name], conn) conn.close() st.dataframe(df) # Auto-generate visualization if len(df.columns) == 2 and 'count' in df.columns.str.lower(): fig = px.bar(df, x=df.columns[0], y='count', title=query_name) st.plotly_chart(fig) if __name__ == "__main__": main() ``` ## Common Patterns ### Pattern: Incremental Data Loading ```python def get_last_artifact_id(connection): """Get the highest artifact ID already in database""" cursor = connection.cursor() cursor.execute("SELECT MAX(id) FROM artifactmetadata") result = cursor.fetchone() cursor.close() return result[0] if result[0] else 0 def fetch_new_artifacts_only(api_key, last_id): """Fetch only artifacts newer than last_id""" params = { 'apikey': api_key, 'size': 100, 'sort': 'id', 'sortorder': 'asc',