Using the Jam API for Data Export

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

# --- Configuration ---
API_KEY: str | None = userdata.get('JAM-API')
if not API_KEY:
    raise RuntimeError("Set your Jam API key as a Colab secret named 'JAM-API' before running.")
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
    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
    df_enriched = df_sessions.merge(
        df_users[["id", "firstName", "lastName"]],
        left_on="learnerUserId",
        right_on="id",
        how="left"
    )

    # 3. Enrich with Mission Titles
    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
    df_enriched["UserName"] = df_enriched["firstName"].fillna("") + " " + df_enriched["lastName"].fillna("")

    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: 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: Process and transform
    df_final = process_data(df_s, df_u, df_m)

    # Step 3: 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()

# /// script
# dependencies = [
#   "requests<3",
#   "pandas",
# ]
# ///

import os
from typing import Any

import pandas as pd
import requests

# Load the key from the environment - never hard-code it into the script.
API_KEY: str | None = os.environ.get("JAM_API_KEY")
if not API_KEY:
    raise SystemExit("Set the JAM_API_KEY environment variable before running.")
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) -> list[dict[str, Any]]:
    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 results


def main() -> None:
    users_data: list[dict[str, Any]] = fetch_paginated("users")
    sessions_data: list[dict[str, Any]] = fetch_paginated("sessions")

    df_sessions: pd.DataFrame = pd.DataFrame(sessions_data)
    df_users: pd.DataFrame = pd.DataFrame(users_data)

    # Count sessions per user
    session_counts: pd.DataFrame = (
        df_sessions["learnerUserId"].value_counts().rename_axis("userId").reset_index(name="sessionCount")
    )

    # Join with users
    df_users["userId"] = df_users["id"]
    df_merged: pd.DataFrame = session_counts.merge(df_users, on="userId", how="left")

    # Get most active user
    most_active_user: pd.Series = df_merged.sort_values(by="sessionCount", ascending=False).iloc[0]

    print("Most Active User:")
    print(f"Name:     {most_active_user['firstName']} {most_active_user['lastName']}")
    print(f"Email:    {most_active_user['email']}")
    print(f"Sessions: {most_active_user['sessionCount']}")


if __name__ == "__main__":
    main()

JAM_API_KEY="your-key" uv run most_active_user.py