TwitterAPIs language clients

import os
import requests

BASE = "https://api.twitterapis.com"
KEY = os.environ["TWITTERAPIS_KEY"]

session = requests.Session()
session.headers["Authorization"] = f"Bearer {KEY}"

def get(path, **params):
    """Call any read endpoint. path is the docs path, e.g. 'user/info'."""
    res = session.get(f"{BASE}/twitter/{path}", params=params, timeout=30)
    res.raise_for_status()
    return res.json()

# Resolve a handle to the permanent numeric user ID
profile = get("user/info", username="naval")
print(profile["user"]["id"], profile["user"]["followers_count"])

# Search posts, then page with the returned cursor
page = get("tweet/advanced_search", query="from:naval min_faves:500")
for tweet in page["tweets"]:
    print(tweet["id"], tweet["text"][:80])

const BASE = "https://api.twitterapis.com";
const KEY = process.env.TWITTERAPIS_KEY;

async function get(path, params = {}) {
  const url = new URL(`/twitter/${path}`, BASE);
  for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);

  const res = await fetch(url, {
    headers: { Authorization: `Bearer ${KEY}` },
  });
  if (!res.ok) {
    throw new Error(`${res.status} ${await res.text()}`);
  }
  return res.json();
}

// Resolve a handle to the permanent numeric user ID
const { user } = await get("user/info", { username: "naval" });
console.log(user.id, user.followers_count);

// Search posts
const page = await get("tweet/advanced_search", {
  query: "from:naval min_faves:500",
});
for (const tweet of page.tweets) console.log(tweet.id, tweet.text);

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"os"
)

const base = "https://api.twitterapis.com"

func get(path string, params url.Values, out any) error {
	req, err := http.NewRequest("GET", base+"/twitter/"+path+"?"+params.Encode(), nil)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+os.Getenv("TWITTERAPIS_KEY"))

	res, err := http.DefaultClient.Do(req)
	if err != nil {
		return err
	}
	defer res.Body.Close()
	if res.StatusCode != http.StatusOK {
		return fmt.Errorf("twitterapis: status %d", res.StatusCode)
	}
	return json.NewDecoder(res.Body).Decode(out)
}

type profile struct {
	User struct {
		ID             string `json:"id"`
		FollowersCount int    `json:"followers_count"`
	} `json:"user"`
}

func main() {
	var p profile
	if err := get("user/info", url.Values{"username": {"naval"}}, &p); err != nil {
		panic(err)
	}
	fmt.Println(p.User.ID, p.User.FollowersCount)
}

<?php
const BASE = "https://api.twitterapis.com";

function twitterapis_get(string $path, array $params = []): array {
    $url = BASE . "/twitter/" . $path . "?" . http_build_query($params);
    $ch = curl_init($url);
    curl_setopt_array($ch, [
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_HTTPHEADER => ["Authorization: Bearer " . getenv("TWITTERAPIS_KEY")],
        CURLOPT_TIMEOUT => 30,
    ]);
    $body = curl_exec($ch);
    $status = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
    curl_close($ch);
    if ($status !== 200) {
        throw new RuntimeException("twitterapis: status $status");
    }
    return json_decode($body, true);
}

$profile = twitterapis_get("user/info", ["username" => "naval"]);
echo $profile["user"]["id"], " ", $profile["user"]["followers_count"], PHP_EOL;

require "net/http"
require "json"
require "uri"

BASE = "https://api.twitterapis.com"

def get(path, params = {})
  uri = URI("#{BASE}/twitter/#{path}")
  uri.query = URI.encode_www_form(params)

  req = Net::HTTP::Get.new(uri)
  req["Authorization"] = "Bearer #{ENV.fetch('TWITTERAPIS_KEY')}"

  res = Net::HTTP.start(uri.hostname, uri.port, use_ssl: true) do |http|
    http.request(req)
  end
  raise "twitterapis: status #{res.code}" unless res.code == "200"

  JSON.parse(res.body)
end

profile = get("user/info", username: "naval")
puts profile["user"]["id"], profile["user"]["followers_count"]

curl "https://api.twitterapis.com/twitter/user/info?username=naval" \
  -H "Authorization: Bearer $TWITTERAPIS_KEY"

curl -G "https://api.twitterapis.com/twitter/tweet/advanced_search" \
  --data-urlencode "query=from:naval min_faves:500" \
  -H "Authorization: Bearer $TWITTERAPIS_KEY"

import os
import requests

BASE = "https://api.twitterapis.com"
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['TWITTERAPIS_KEY']}"

def get(path, **params):
    res = session.get(f"{BASE}/twitter/{path}", params=params, timeout=30)
    res.raise_for_status()
    return res.json()

profile = get("user/info", username="naval")
print(profile["user"]["id"], profile["user"]["followers_count"])

# BOTH terminators, because the two endpoint families fail in
# opposite directions: search nulls the cursor on a page that is still FULL,
# and follower-graph endpoints never null it at all. Either guard alone loops.
MAX_CALLS = 200
cursor, calls, seen = None, 0, set()

while calls < MAX_CALLS:
    page = get("tweet/advanced_search",
               query="from:naval min_faves:500",
               **({"cursor": cursor} if cursor else {}))
    calls += 1

    tweets = page.get("tweets") or []
    if not tweets:
        break                       # empty page: nothing left

    for tweet in tweets:
        print(tweet["id"], tweet["text"][:80])

    cursor = page.get("next_cursor")
    if not cursor:
        break                       # cursor gone: this WAS the last page
    if cursor in seen:              # catches a stuck cursor AND any cycle
        raise RuntimeError(f"cursor stopped advancing at {cursor}")
    seen.add(cursor)
else:
    print(f"stopped at the {MAX_CALLS}-call cap, resume from {cursor}")

import json

# BOTH terminators, plus a repeat check. Follower-graph endpoints return a
# NON-NULL cursor even on the last page, so the empty collection is what ends
# THIS call; but user/followers_v2 nulls the cursor on a page that is still
# full, and omitting the cursor restarts at page one, so the cursor guard has
# to be here too. A cursor that never changes is the third way this loops.
MAX_CALLS = 500
cursor, calls, seen = None, 0, set()
with open("followers.ndjson", "w") as out:
    while calls < MAX_CALLS:
        page = get("user/followers",
                   username="naval",
                   **({"cursor": cursor} if cursor else {}))
        calls += 1

        users = page.get("users") or []
        if not users:              # <- the real terminator
            break

        for user in users:
            out.write(json.dumps(user) + "\n")
        out.flush()                # partial file stays valid and resumable

        cursor = page.get("next_cursor")
        if not cursor:
            break                  # cursor spent: that WAS the last page
        if cursor in seen:         # catches a stuck cursor AND any cycle
            raise RuntimeError(f"cursor stopped advancing at {cursor}")
        seen.add(cursor)
    else:
        print(f"stopped at the {MAX_CALLS}-call cap, resume from {cursor}")

# Measured records per call (biller usage_logs, 396,817 calls, Aug 2026)
YIELD = {"user/tweets": 18.78, "bulk": 12.96, "tweet/advanced_search": 7.62}
PRICE_PER_CALL = 0.0008

def estimate(records_needed, endpoint="tweet/advanced_search"):
    calls = records_needed / YIELD[endpoint]
    return calls, calls * PRICE_PER_CALL

calls, usd = estimate(50_000, "tweet/advanced_search")
print(f"{calls:,.0f} calls, about ${usd:,.2f}")   # 6,562 calls, about $5.25

# Cap by CALLS, not by records: a query that returns nothing still bills.
MAX_CALLS = 7000