The Insights API

{  "name": "My Report",

  "show_in_ui": false,

  "form": {

    "deployment_id": 5716,

    "report_type": "compute_insights_overview",

    "start_date": "2024-01-01",

    "end_date": "2024-03-31"

  }

}

# Recast Insights API — R Usage Example
#
# Workflow: Fetch latest deployment → Create report → Poll → Download CSVs
#
# Prerequisites:
#   install.packages(c("httr2", "jsonlite"))
#   Add to .Renviron: API_PAT=gr_your_token_here
#     Generate your token: log into the Recast app → click your email (top right) → Generate API token
#     Store it in ~/.Renviron as: API_PAT=gr_your_token_here
#     Never hard-code or share this token.

library(httr2)
library(jsonlite)

# ── Edit these ────────────────────────────────────────────────────────────────
CLIENT_SLUG <- "myslug"                    # Your client slug (visible in the app URL)
REPORT_TYPE <- "compute_insights_overview"
START_DATE  <- "2024-01-01"
END_DATE    <- "2024-03-31"
SHOW_IN_UI  <- FALSE                           # TRUE to see this report in the app

# ── Rarely needs changing ─────────────────────────────────────────────────────
BASE_URL <- "https://api.getrecast.com"
PAT      <- Sys.getenv("API_PAT")             # Set in ~/.Renviron, never hard-coded

# ── Helpers ───────────────────────────────────────────────────────────────────
parse <- \(resp) resp |> resp_body_string() |> fromJSON(simplifyVector = FALSE)

check <- function(resp, expected = 200) {
  if (resp_status(resp) != expected) {
    body <- tryCatch(resp_body_string(resp), error = \(e) "(no body)")
    stop(sprintf("HTTP %d: %s", resp_status(resp), body))
  }
  resp
}

api_request <- function(...) {
  request(BASE_URL) |>
    req_url_path_append("v1", "clients", CLIENT_SLUG, ...) |>
    req_auth_bearer_token(PAT) |>
    req_headers(Accept = "application/json") |>
    req_error(is_error = \(resp) FALSE)
}

# ── Step 1: Fetch the latest deployment ──────────────────────────────────────
deployment_id <- api_request("deployments") |>
  req_perform() |> check() |> parse() |>
  (\(r) r$data[[1]]$id)()
cat(sprintf("Using deployment ID: %d\n", deployment_id))

# ── Step 2: Create a report ───────────────────────────────────────────────────
resp <- api_request("reports") |>
  req_method("POST") |>
  req_headers(`Content-Type` = "application/json") |>
  req_body_json(list(
    name       = paste("Channel Attribution -", Sys.Date()),
    show_in_ui = SHOW_IN_UI,
    form       = list(
      deployment_id = deployment_id,
      report_type   = REPORT_TYPE,
      start_date    = START_DATE,
      end_date      = END_DATE
    )
  ), auto_unbox = TRUE) |>
  req_perform() |>
  check(expected = 201)

report_id <- parse(resp)$id
cat(sprintf("Created report %d\n", report_id))

# ── Step 3: Poll until complete (typically 1–5 minutes) ──────────────────────
repeat {
  result <- api_request("reports", report_id) |> req_perform() |> check() |> parse()
  cat(sprintf("Status: %s\n", result$status))
  if (!result$status %in% c("ready", "processing")) break
  Sys.sleep(30)
}

if (result$status != "success") {
  stop(sprintf("Report ended with status: %s", result$status))
}

# ── Step 4: Download all available CSVs ──────────────────────────────────────
downloads <- result$results$downloads
cat(sprintf("Downloading %d files...\n", length(downloads)))

for (dl in downloads) {
  csv_resp <- api_request("reports", report_id, "downloads", dl$key) |>
    req_perform() |>
    check()
  df <- read.csv(text = resp_body_string(csv_resp), check.names = FALSE)
  write.csv(df, file = paste0(dl$key, ".csv"), row.names = FALSE)
  cat(sprintf("  Saved %s.csv (%d rows)\n", dl$key, nrow(df)))
}

# Recast Insights API — Python Usage Example
#
# Workflow: Fetch latest deployment → Create report → Poll → Download CSVs
#
# Prerequisites:
#   pip install requests pandas
#   Generate your token: log into the Recast app → click your email (top right) → Generate API token
#   Set as an environment variable: export API_PAT=gr_your_token_here
#   Never hard-code or share this token.

import os
import time
import io
import requests
import pandas as pd
from datetime import date

# ── Edit these ────────────────────────────────────────────────────────────────
CLIENT_SLUG = "myslug"                     # Your client slug (visible in the app URL)
REPORT_TYPE = "compute_insights_overview"
START_DATE  = "2024-01-01"
END_DATE    = "2024-03-31"
SHOW_IN_UI  = False                            # True to see this report in the app

# ── Rarely needs changing ─────────────────────────────────────────────────────
BASE_URL = "https://api.getrecast.com"
PAT      = os.environ["API_PAT"]              # Set via: export API_PAT=gr_your_token_here

BASE_PATH = f"{BASE_URL}/v1/clients/{CLIENT_SLUG}"
HEADERS   = {"Authorization": f"Bearer {PAT}", "Accept": "application/json"}

# ── Helpers ───────────────────────────────────────────────────────────────────
def check(resp, expected=200):
    if resp.status_code != expected:
        body = resp.text or "(no body)"
        raise Exception(f"HTTP {resp.status_code}: {body}")
    return resp

# ── Step 1: Fetch the latest deployment ──────────────────────────────────────
deployments = check(requests.get(f"{BASE_PATH}/deployments", headers=HEADERS)).json()
deployment_id = deployments["data"][0]["id"]
print(f"Using deployment ID: {deployment_id}")

# ── Step 2: Create a report ───────────────────────────────────────────────────
resp = check(requests.post(
    f"{BASE_PATH}/reports",
    headers={**HEADERS, "Content-Type": "application/json"},
    json={
        "name": f"Channel Attribution - {date.today()}",
        "show_in_ui": SHOW_IN_UI,
        "form": {
            "deployment_id": deployment_id,
            "report_type": REPORT_TYPE,
            "start_date": START_DATE,
            "end_date": END_DATE,
        },
    },
), expected=201)

report_id = resp.json()["id"]
print(f"Created report {report_id}")

# ── Step 3: Poll until complete (typically 1–5 minutes) ──────────────────────
while True:
    result = check(requests.get(f"{BASE_PATH}/reports/{report_id}", headers=HEADERS)).json()
    print(f"Status: {result['status']}")
    if result["status"] not in ("ready", "processing"):
        break
    time.sleep(30)

if result["status"] != "success":
    raise Exception(f"Report ended with status: {result['status']}")

# ── Step 4: Download all available CSVs ──────────────────────────────────────
downloads = result["results"]["downloads"]
print(f"Downloading {len(downloads)} files...")

for dl in downloads:
    csv_resp = check(requests.get(
        f"{BASE_PATH}/reports/{report_id}/downloads/{dl['key']}",
        headers=HEADERS,
    ))
    df = pd.read_csv(io.StringIO(csv_resp.text))
    df.to_csv(f"{dl['key']}.csv", index=False)
    print(f"  Saved {dl['key']}.csv ({len(df)} rows)")