Recipes: mentions, polling and datasets – Docs

Docs

Recipes: mentions, polling and datasets

Ready-to-run examples for brand mentions, cashtag and contract searches, polling without duplicates, following a set of accounts through a list, and collecting a search window to JSON Lines in parallel.

Short, complete programs for the jobs people use TwittrAPI for most. They use curl or the Python SDK with the simple format, and read the key from TWITTRAPI_API_KEY. Every page of results is one billed call; calls that fail on our side are refunded and 429 is not charged (billing).

Brand mentions

Mentions come in three shapes: the handle, the name and the domain. Catch all three, drop your own account and plain retweets:

(@acme OR "acme app" OR url:acme.com) -from:acme -filter:nativeretweets

Mentions of the last day, English, without job posts and giveaways. The filters object adds operators for you:

curl -G 'https://api.twittrapi.com/search' \
  -H 'X-API-KEY: YOUR_API_KEY' \
  --data-urlencode 'searchItem=(@acme OR "acme app" OR url:acme.com) -from:acme' \
  --data-urlencode 'category=Latest' \
  --data-urlencode 'filters={"lang":"en","noneOfTheseWords":["hiring","giveaway"],"withinTime":"1d"}' \
  --data-urlencode 'format=simple'

Common words need context: "acme" alone matches everything, "acme" plus your product words or domain does not. For a campaign hashtag with traction only, use filters={"hashtags":["AcmeSummer"],"minimumLikesCount":20,"removeReplies":true}. All operators and filter keys: search reference.

Cashtags and contract addresses

Latest $BTC or $ETH tweets with some engagement, no replies:

curl -G 'https://api.twittrapi.com/search' \
  -H 'X-API-KEY: YOUR_API_KEY' \
  --data-urlencode 'searchItem=$BTC OR $ETH' \
  --data-urlencode 'category=Latest' \
  --data-urlencode 'filters={"minimumLikesCount":5,"removeReplies":true,"noneOfTheseWords":["giveaway","airdrop"]}' \
  --data-urlencode 'format=simple'

Cashtags are plain text in the query: $BTC, $SOL OR $JUP. The $ must be URL-encoded (%24); --data-urlencode and the SDKs do that for you.
Contract addresses are matched as text. Quote them, "0x6982508145454ce325ddbe47a25d4ec3d2311933", and also search the project handle and domain (url:example.xyz).
Query plus filters can be up to 512 characters. Square brackets are rejected with 422.

Poll a search without duplicates

Ask for the newest tweets every 30 seconds and pass the highest ID you have as since_id: (or sinceId in filters). Each tweet arrives once, oldest first. The SDK retries 429 and 5xx with backoff. To start from a point in time instead of a tweet you have, turn the date into an ID with the tweet ID to date converter.

import time
 
from twittrapi import TwittrAPI, TwittrAPIError
 
client = TwittrAPI(response_format="simple")  # key from TWITTRAPI_API_KEY
QUERY = "($BTC OR $ETH) -filter:replies"
newest = None  # highest tweet ID seen so far
 
while True:
    query = f"{QUERY} since_id:{newest}" if newest else QUERY
    try:
        page = client.search.tweets(query, category="Latest")
    except TwittrAPIError as err:  # 429 and 5xx are already retried by the SDK
        print("error:", err)
        time.sleep(30)
        continue
    tweets = [t for t in page["data"] if not t.get("unavailable")]
    for t in sorted(tweets, key=lambda t: int(t["id"])):  # oldest first
        author = t["author"]["username"] if t["author"] else "?"
        print(t["createdAt"], author, t["text"][:140])
    if tweets:
        newest = max((t["id"] for t in tweets), key=int)
    time.sleep(30)

Every poll is one call whether or not it finds something new, so the interval sets the cost. For new tweets from a fixed set of accounts within seconds, without running the loop yourself, see account monitoring.

Follow a set of accounts with one call

Put the accounts you follow on a public X list. List tweets returns the newest tweets of all members, so one call per poll covers the whole list, however many accounts are on it. The first run prints the current page; after that only new tweets.

import time
 
from twittrapi import TwittrAPI
 
client = TwittrAPI(response_format="simple")
LIST_ID = "1234567890"  # a public X list with the accounts you follow
seen: set[str] = set()
 
while True:
    page = client.lists.tweets(LIST_ID, count=20)
    fresh = [t for t in page["data"] if not t.get("unavailable") and t["id"] not in seen]
    for t in sorted(fresh, key=lambda t: int(t["id"])):
        author = t["author"]["username"] if t["author"] else "?"
        print(t["createdAt"], author, t["text"][:140], t["url"])
    seen.update(t["id"] for t in fresh)  # keep the newest few thousand in production
    time.sleep(15)

Collect a search window to JSON Lines

With the simple format every page carries nextCursor and hasNextPage, so the loop has a clear end. The SDK's paginate follows the cursor for you and retries 429 and 5xx with backoff.

import json
 
from twittrapi import TwittrAPI
 
client = TwittrAPI(response_format="simple", max_retries=5)  # key from TWITTRAPI_API_KEY
 
 
def collect(query: str, day: str, next_day: str, path: str) -> int:
    """Write every tweet of one day to JSON Lines. Returns the pages fetched (= calls)."""
    pages = 0
    with open(path, "w", encoding="utf-8") as out:
        for page in client.paginate(
            client.search.tweets,
            f"{query} since:{day} until:{next_day}",
            category="Latest",
            count=100,
            max_pages=None,  # follow nextCursor until hasNextPage is false
        ):
            pages += 1
            for t in page["data"]:
                if t.get("unavailable"):
                    continue
                row = {"id": t["id"], "createdAt": t["createdAt"], "lang": t["lang"],
                       "authorId": (t["author"] or {}).get("id"), "text": t["text"]}
                out.write(json.dumps(row, ensure_ascii=False) + "\n")
    return pages

count is a hint (1 to 100); X decides the real page size.
Cursors are X's and can expire after a while. If a run can be interrupted, store the last cursor with your progress and resume soon.

Going faster: split the time range

Pagination is sequential: page 2 needs the cursor from page 1. At roughly a second per response, one cursor chain gives about one page per second, however high your rate limit is. Split the range into windows (a day, an hour) and run the windows in parallel:

from concurrent.futures import ThreadPoolExecutor
from datetime import date, timedelta
 
QUERY = "(climate OR klima) lang:de"
days = [date(2026, 9, 1) + timedelta(d) for d in range(30)]
 
 
def run(day: date) -> None:
    path = f"climate_de_{day}.jsonl"
    pages = collect(QUERY, str(day), str(day + timedelta(1)), path)
    print(f"{day}  {pages} pages  → {path}")
 
 
with ThreadPoolExecutor(max_workers=25) as pool:  # stays under the included rate limit
    list(pool.map(run, days))

Keep the number of workers below your rate limit in requests per second. Higher limits are monthly add-ons you can book for the month of a crawl.