Live data examples

import asyncio
import os
from typing import Any

import httpx


async def fetch_live_posts(handle: str, max_pages: int = 3) -> list[dict[str, Any]]:
    posts: list[dict[str, Any]] = []
    cursor: str | None = None
    headers = {"Authorization": f"Bearer {os.environ['CREATORAUDIT_API_KEY']}"}
    async with httpx.AsyncClient(
        base_url="https://api.creatoraudit.com/v2/",
        headers=headers,
        timeout=60.0,
    ) as client:
        for _ in range(max_pages):
            params = {"handle": handle}
            if cursor is not None:
                params["cursor"] = cursor
            response = await client.get("live/tiktok/posts", params=params)
            response.raise_for_status()
            body = response.json()
            posts.extend(body["data"])
            page = body["pagination"]
            if not page["has_next"]:
                break
            cursor = page["next_cursor"]
    return posts


if __name__ == "__main__":
    print(asyncio.run(fetch_live_posts("nasa")))

type LivePost = {
  type: "live_post";
  platform: "instagram" | "tiktok";
  external_id: string;
  url: string;
  like_count: number | null;
};

type LiveFeed = {
  data: LivePost[];
  pagination: { next_cursor: string | null; has_next: boolean };
  meta: { fetched_at: string; cached: false; warnings: string[] };
};

async function fetchLivePosts(
  apiKey: string,
  handle: string,
  maxPages = 3,
): Promise<LivePost[]> {
  const posts: LivePost[] = [];
  let cursor: string | null = null;
  for (let pageNumber = 0; pageNumber < maxPages; pageNumber++) {
    const url = new URL("https://api.creatoraudit.com/v2/live/tiktok/posts");
    url.searchParams.set("handle", handle);
    if (cursor !== null) url.searchParams.set("cursor", cursor);
    const response = await fetch(url, {
      headers: { Authorization: `Bearer ${apiKey}` },
      cache: "no-store",
      signal: AbortSignal.timeout(60_000),
    });
    if (!response.ok) {
      throw new Error(`HTTP ${response.status}: ${await response.text()}`);
    }
    const body: LiveFeed = await response.json();
    posts.push(...body.data);
    if (!body.pagination.has_next) break;
    cursor = body.pagination.next_cursor;
  }
  return posts;
}