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 ETL pipelines and analytics dashboards for Harvard Art Museums API data using Python, SQL, and Streamlit
triggers
["how do I fetch Harvard Art Museums data","build an ETL pipeline for museum artifacts","create analytics dashboard with Streamlit and museum API","extract and transform Harvard artifacts data","set up SQL database for art collection data","query Harvard Art Museums API with pagination","visualize museum artifact data with Plotly","build data engineering pipeline for cultural heritage data"]
This project provides a complete data engineering and analytics solution for the Harvard Art Museums API. It demonstrates ETL pipeline construction, relational database design, SQL analytics, and interactive visualization using Streamlit. The application extracts artifact metadata, transforms nested JSON into structured tables, loads data into SQL databases, and provides 20+ analytical queries with auto-generated visualizations.
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 variablesexport HARVARD_API_KEY="your_api_key_here"export DB_HOST=
DB_USER=
DB_PASSWORD=
DB_NAME=
Fetch artifacts with pagination and rate limiting:
import requests
import time
import os
deffetch_artifacts(api_key, page=1, size=100):
"""Fetch artifacts from Harvard Art Museums API with pagination"""
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:
return response.json()
else:
raise Exception(f"API Error: {response.status_code}")
# Collect multiple pages with rate limitingdefcollect_artifacts(num_pages=10):
api_key = os.getenv('HARVARD_API_KEY')
all_artifacts = []
for page inrange(1, num_pages + 1):
print(f"Fetching page {page}...")
data = fetch_artifacts(api_key, page=page)
all_artifacts.extend(data.get('records', []))
time.sleep(1) # Rate limitingreturn all_artifacts
2. ETL Pipeline
Transform nested JSON into relational tables:
import pandas as pd
import json
defextract_artifact_metadata(artifacts):
"""Extract main artifact metadata"""
metadata = []
for artifact in artifacts:
metadata.append({
'objectid': 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'),
'division': artifact.get('division'),
'dated': artifact.get('dated'),
'description': artifact.get('description'),
'technique': artifact.get('technique'),
'medium': artifact.get('medium'),
'dimensions': artifact.get('dimensions'),
'creditline': artifact.get('creditline'),
'accessionyear': artifact.get('accessionyear')
})
return pd.DataFrame(metadata)
defextract_artifact_media(artifacts):
"""Extract media/image data"""
media = []
for artifact in artifacts:
objectid = artifact.get('objectid')
images = artifact.get('images', [])
for img in images:
media.append({
'objectid': objectid,
'imageid': img.get('imageid'),
'baseimageurl': img.get('baseimageurl'),
'iiifbaseuri': img.get('iiifbaseuri'),
'height': img.get('height'),
'width': img.get('width'),
'format': img.get('format')
})
return pd.DataFrame(media)
defextract_artifact_colors(artifacts):
"""Extract color palette data"""
colors = []
for artifact in artifacts:
objectid = artifact.get('objectid')
color_data = artifact.get('colors', [])
for color in color_data:
colors.append({
'objectid': objectid,
'color': color.get('color'),
'spectrum': color.get('spectrum'),
'hue': color.get('hue'),
'percent': color.get('percent')
})
return pd.DataFrame(colors)
# Complete ETL processdefrun_etl():
print("Starting ETL pipeline...")
# Extract
artifacts = collect_artifacts(num_pages=5)
# Transform
df_metadata = extract_artifact_metadata(artifacts)
df_media = extract_artifact_media(artifacts)
df_colors = extract_artifact_colors(artifacts)
print(f"Extracted {len(df_metadata)} artifacts")
print(f"Extracted {len(df_media)} media records")
print(f"Extracted {len(df_colors)} color records")
return df_metadata, df_media, df_colors
3. Database Schema and Loading
import mysql.connector
from mysql.connector import Error
import os
defget_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')
)
defcreate_tables(connection):
"""Create database schema"""
cursor = connection.cursor()
# Artifact metadata table
cursor.execute("""
CREATE TABLE IF NOT EXISTS artifactmetadata (
objectid INT PRIMARY KEY,
title TEXT,
culture VARCHAR(255),
period VARCHAR(255),
century VARCHAR(100),
classification VARCHAR(255),
department VARCHAR(255),
division VARCHAR(255),
dated VARCHAR(255),
description TEXT,
technique TEXT,
medium TEXT,
dimensions TEXT,
creditline TEXT,
accessionyear INT
)
""")
# Artifact media table
cursor.execute("""
CREATE TABLE IF NOT EXISTS artifactmedia (
id INT AUTO_INCREMENT PRIMARY KEY,
objectid INT,
imageid INT,
baseimageurl TEXT,
iiifbaseuri TEXT,
height INT,
width INT,
format VARCHAR(50),
FOREIGN KEY (objectid) REFERENCES artifactmetadata(objectid)
)
""")
# Artifact colors table
cursor.execute("""
CREATE TABLE IF NOT EXISTS artifactcolors (
id INT AUTO_INCREMENT PRIMARY KEY,
objectid INT,
color VARCHAR(100),
spectrum VARCHAR(100),
hue VARCHAR(100),
percent FLOAT,
FOREIGN KEY (objectid) REFERENCES artifactmetadata(objectid)
)
""")
connection.commit()
print("Tables created successfully")
defload_dataframe_to_sql(df, table_name, connection):
"""Batch insert dataframe into SQL table"""
cursor = connection.cursor()
# Prepare column names and placeholders
columns = ', '.join(df.columns)
placeholders = ', '.join(['%s'] * len(df.columns))
insert_query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"# Convert dataframe to list of tuples
data = [tuple(x) for x in df.to_numpy()]
# Batch insert
cursor.executemany(insert_query, data)
connection.commit()
print(f"Loaded {len(df)} records into {table_name}")
# Complete load processdefload_to_database(df_metadata, df_media, df_colors):
connection = get_db_connection()
create_tables(connection)
load_dataframe_to_sql(df_metadata, 'artifactmetadata', connection)
load_dataframe_to_sql(df_media, 'artifactmedia', connection)
load_dataframe_to_sql(df_colors, 'artifactcolors', connection)
connection.close()
print("Data loading complete!")
4. SQL Analytics Queries
defexecute_query(query, connection):
"""Execute SQL query and return results as DataFrame"""return pd.read_sql(query, connection)
# Sample analytical queries
ANALYTICS_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 10
""",
"Artifacts by Century": """
SELECT century, COUNT(*) as count
FROM artifactmetadata
WHERE century IS NOT NULL
GROUP BY century
ORDER BY count DESC
""",
"Top Departments": """
SELECT department, COUNT(*) as total_artifacts
FROM artifactmetadata
WHERE department IS NOT NULL
GROUP BY department
ORDER BY total_artifacts DESC
LIMIT 10
""",
"Media Availability": """
SELECT
CASE WHEN m.objectid 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.objectid = m.objectid
GROUP BY media_status
""",
"Color Distribution": """
SELECT color, COUNT(*) as frequency, AVG(percent) as avg_percent
FROM artifactcolors
GROUP BY color
ORDER BY frequency DESC
LIMIT 15
""",
"Artifacts by Classification": """
SELECT classification, COUNT(*) as count
FROM artifactmetadata
WHERE classification IS NOT NULL
GROUP BY classification
ORDER BY count DESC
LIMIT 10
""",
"Average Image Dimensions": """
SELECT
AVG(height) as avg_height,
AVG(width) as avg_width,
COUNT(*) as total_images
FROM artifactmedia
"""
}
# Start Streamlit dashboard
streamlit run app.py
# Access at http://localhost:8501
Common Patterns
Incremental Data Loading
defget_max_objectid(connection):
"""Get the latest objectid to avoid duplicates"""
cursor = connection.cursor()
cursor.execute("SELECT MAX(objectid) FROM artifactmetadata")
result = cursor.fetchone()[0]
return result if result else0defincremental_etl():
"""Only fetch new artifacts"""
connection = get_db_connection()
max_id = get_max_objectid(connection)
# Fetch artifacts with objectid > max_id# ... implement filtering logic
connection.close()
