YouTube Search Hashtag API

# GET /v1/youtube/search/hashtag: Metered: holds 1 credit per page up front and refunds what the call does not use. Cache hits are free and failed calls are refunded.
# Check the price first for 0 credits: add dry_run=1 to the query. Nothing is fetched or charged.
# --max-time 10: this endpoint's measured p99 latency x 1.5.
# --retry 3: retries timeouts, 429 and 5xx, honoring Retry-After.
# -o response.json keeps only the final attempt's body.
# One Idempotency-Key per request, reused on every retry, so a retried call is never charged twice.
# Next page: send pagination.next_cursor back as cursor until pagination.has_more is false.
IDEMPOTENCY_KEY=$(od -An -N16 -tx1 /dev/urandom | tr -d ' \n')
curl -sS --max-time 10 --retry 3 -o response.json \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Idempotency-Key: $IDEMPOTENCY_KEY" \
  "https://www.socialcrawl.dev/v1/youtube/search/hashtag?hashtag=shorts" && cat response.json

# GET /v1/youtube/search/hashtag: Metered: holds 1 credit per page up front and refunds what the call does not use. Cache hits are free and failed calls are refunded.
# Check the price first for 0 credits: add dry_run=1 to the query. Nothing is fetched or charged.
import os
import random
import time
import uuid

import requests

API_KEY = os.environ.get("SOCIALCRAWL_API_KEY", "YOUR_API_KEY")
TIMEOUT_S = 10  # this endpoint's measured p99 latency x 1.5.
MAX_RETRIES = 3
URL = "https://www.socialcrawl.dev/v1/youtube/search/hashtag"


def backoff(attempt):
    # Exponential backoff with jitter: about 1 s, 2 s, then 4 s.
    return 2 ** attempt * (0.5 + random.random() / 2)


def socialcrawl(params):
    # One Idempotency-Key per request, reused on every retry, so a retried call is never charged twice.
    idempotency_key = str(uuid.uuid4())
    for attempt in range(MAX_RETRIES + 1):
        try:
            res = requests.get(
                URL,
                params=params,
                headers={"x-api-key": API_KEY, "Idempotency-Key": idempotency_key},
                timeout=TIMEOUT_S,
            )
            body = res.json()
        except (requests.RequestException, ValueError):
            # A timeout, a dropped connection or a non-JSON gateway page: retry with the same key.
            if attempt == MAX_RETRIES:
                raise
            time.sleep(backoff(attempt))
            continue
        if body["success"]:
            return body
        error = body["error"]
        # Branch on error["type"], never on the message text.
        if error["type"] == "RESOURCE_NOT_FOUND":
            return None  # refunded: nothing to read
        retry_after = res.headers.get("Retry-After")
        # SERVICE_UNAVAILABLE without Retry-After is a withdrawn endpoint: permanent.
        permanent = error["type"] == "SERVICE_UNAVAILABLE" and retry_after is None
        if not error.get("retryable") or permanent or attempt == MAX_RETRIES:
            raise RuntimeError(f"{error['type']}: {error['message']} (request_id {body['request_id']})")
        # Retry-After (seconds) wins; otherwise back off.
        time.sleep(float(retry_after) if retry_after else backoff(attempt))


params = {
    "hashtag": "shorts",
}

# Walk the list: send pagination.next_cursor back as cursor until has_more is false.
MAX_PAGES = 10  # each page is one call
items = []
for _ in range(MAX_PAGES):
    body = socialcrawl(params)
    if body is None:
        break
    items.extend(body["data"]["items"])
    page = body.get("pagination") or {}
    if not page.get("has_more") or not page.get("next_cursor"):
        break
    params["cursor"] = page["next_cursor"]
print(f"{len(items)} items")

// GET /v1/youtube/search/hashtag: Metered: holds 1 credit per page up front and refunds what the call does not use. Cache hits are free and failed calls are refunded.
// Check the price first for 0 credits: add dry_run=1 to the query. Nothing is fetched or charged.
const API_KEY = process.env.SOCIALCRAWL_API_KEY ?? "YOUR_API_KEY";
const TIMEOUT_MS = 10_000; // this endpoint's measured p99 latency x 1.5.
const MAX_RETRIES = 3;

type Page = { next_cursor: string | null; has_more: boolean };
type SocialCrawlResponse =
  | { success: true; data: any; pagination?: Page; credits_used: number; request_id: string }
  | { success: false; error: { type: string; message: string; retryable: boolean }; request_id: string };

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
// Exponential backoff with jitter: about 1 s, 2 s, then 4 s.
const backoffMs = (attempt: number) => 2 ** attempt * 1000 * (0.5 + Math.random() / 2);

async function socialcrawl(url: URL) {
  // One Idempotency-Key per request, reused on every retry, so a retried call is never charged twice.
  const idempotencyKey = crypto.randomUUID();
  for (let attempt = 0; ; attempt++) {
    let res: Response;
    let body: SocialCrawlResponse;
    try {
      res = await fetch(url, {
        headers: {
          "x-api-key": API_KEY,
          "Idempotency-Key": idempotencyKey,
        },
        signal: AbortSignal.timeout(TIMEOUT_MS),
      });
      body = (await res.json()) as SocialCrawlResponse;
    } catch (error) {
      // A timeout, a dropped connection or a non-JSON gateway page: retry with the same key.
      if (attempt >= MAX_RETRIES) throw error;
      await sleep(backoffMs(attempt));
      continue;
    }
    if (body.success) return body;
    // Branch on error.type, never on the message text.
    if (body.error.type === "RESOURCE_NOT_FOUND") return null; // refunded: nothing to read
    const retryAfter = res.headers.get("Retry-After");
    // SERVICE_UNAVAILABLE without Retry-After is a withdrawn endpoint: permanent.
    const permanent = body.error.type === "SERVICE_UNAVAILABLE" && retryAfter === null;
    if (!body.error.retryable || permanent || attempt >= MAX_RETRIES) {
      throw new Error(`${body.error.type}: ${body.error.message} (request_id ${body.request_id})`);
    }
    // Retry-After (seconds) wins; otherwise back off.
    await sleep(retryAfter ? Number(retryAfter) * 1000 : backoffMs(attempt));
  }
}

const url = new URL("https://www.socialcrawl.dev/v1/youtube/search/hashtag");
url.searchParams.set("hashtag", "shorts");

// Walk the list: send pagination.next_cursor back as cursor until has_more is false.
const MAX_PAGES = 10; // each page is one call
const items: unknown[] = [];
for (let page = 0; page < MAX_PAGES; page++) {
  const body = await socialcrawl(url);
  if (!body) break;
  items.push(...body.data.items);
  const next = body.pagination?.next_cursor;
  if (!body.pagination?.has_more || !next) break;
  url.searchParams.set("cursor", next);
}
console.log(`${items.length} items`);