Using the Jam API for Data Export

What we do here

• Start from the sessions object

• Extract the score (0-100) for the session and whether it was successfully completed

• Add user names from the users object

• Add mission titles from the missions object

• Export the result as a csv that can be shared with stakeholders to do computations on

Python Code

import pandas as pd
import requests
from google.colab import userdata
from typing import Any

# --- Configuration ---
API_KEY: str = userdata.get('JAM-API')
BASE_URL: str = "https://api.wejam.ai/api/v1/data-exports"
HEADERS: dict[str, str] = {"accept": "application/json", "X-API-KEY": API_KEY}

def fetch_paginated(endpoint: str) -> pd.DataFrame:
    """Fetches all pages for a given endpoint and returns a DataFrame."""
    page: int = 1
    results: list[dict[str, Any]] = []

    while True:
        url: str = f"{BASE_URL}/{endpoint}?page={page}&limit=100"
        response: requests.Response = requests.get(url, headers=HEADERS, timeout=10)
        response.raise_for_status()
        data: dict[str, Any] = response.json()

        results.extend(data["data"])
        if not data["meta"].get("hasNext"):
            break
        page += 1

    return pd.DataFrame(results)

def process_data(df_sessions: pd.DataFrame, df_users: pd.DataFrame, df_missions: pd.DataFrame) -> pd.DataFrame:
    """Handles all data cleaning, merging, and restructuring logic (DRY)."""
    
    # 1. Clean Sessions: Extract score early to keep logic localized
    # Using .get safely and ensuring integer type for the 'Score' column
    df_sessions["score"] = df_sessions["analysis"].apply(
        lambda x: x.get("score") if isinstance(x, dict) else 0
    ).fillna(0).astype(int)

    # 2. Enrich Sessions with User Data
    # We merge only the columns we need (id, names) to keep the DataFrame lean
    df_enriched = df_sessions.merge(
        df_users[["id", "firstName", "lastName"]],
        left_on="learnerUserId",
        right_on="id",
        how="left"
    )

    # 3. Enrich with Mission Titles
    # We use suffixes to avoid name collisions if 'id' or other meta-cols exist
    df_enriched = df_enriched.merge(
        df_missions[["id", "title"]],
        left_on="missionId",
        right_on="id",
        how="left",
        suffixes=('', '_mission')
    )

    # 4. Final Transformation: Create derived columns and select output
    # Concatenating name strings into a single column as requested
    df_enriched["UserName"] = df_enriched["firstName"].fillna("") + " " + df_enriched["lastName"].fillna("")
    
    # Selecting and renaming in one step to finalize the schema
    output = df_enriched[[
        "UserName",
        "title",
        "createdAt",
        "score",
        "completed"
    ]].rename(columns={
        "title": "MissionTitle",
        "createdAt": "SessionDate",
        "score": "Score",
        "completed": "Completed"
    })

    return output.sort_values("SessionDate", ascending=False)

def main() -> None:
    # Step 1: I/O - Fetch raw data
    print("Fetching data from API...")
    df_u = fetch_paginated("users")
    df_s = fetch_paginated("sessions")
    df_m = fetch_paginated("missions")

    # Step 2: Logic - Process and transform
    df_final = process_data(df_s, df_u, df_m)

    # Step 3: I/O - Export results to csv
    df_final.to_csv("sessions_with_users.csv", index=False)

    print(f"\nSuccess. Processed {len(df_final)} sessions.")
    display(df_final.head(20))

if __name__ == "__main__":
    main()

Sample output