Rate limits

from twikit import Client
from twikit.errors import TooManyRequests
import asyncio
import time

client = Client('en-US')

async def search_with_retry(query):
    try:
        tweets = await client.search_tweet(query, 'Latest')
        return tweets
    except TooManyRequests as e:
        # Get the reset time from the exception
        reset_time = e.rate_limit_reset
        if reset_time:
            wait_time = reset_time - int(time.time())
            print(f'Rate limited. Waiting {wait_time} seconds...')
            await asyncio.sleep(wait_time)
            # Retry after waiting
            return await client.search_tweet(query, 'Latest')
        else:
            print('Rate limited. Waiting 15 minutes...')
            await asyncio.sleep(900)  # Wait 15 minutes
            return await client.search_tweet(query, 'Latest')

# Good: Request smaller batches
tweets = await client.search_tweet('python', 'Latest', count=20)

# Process the first batch
for tweet in tweets:
    process_tweet(tweet)

# Get more only if needed
if need_more:
    more_tweets = await tweets.next()

import asyncio
from twikit.errors import TooManyRequests

async def fetch_with_backoff(fetch_func, max_retries=3):
    """Fetch data with exponential backoff on rate limits."""
    for attempt in range(max_retries):
        try:
            return await fetch_func()
        except TooManyRequests as e:
            if attempt == max_retries - 1:
                raise
            
            wait_time = 2 ** attempt * 60  # 1m, 2m, 4m
            if e.rate_limit_reset:
                wait_time = min(wait_time, e.rate_limit_reset - int(time.time()))
            
            print(f'Rate limited. Retrying in {wait_time}s (attempt {attempt + 1}/{max_retries})')
            await asyncio.sleep(wait_time)

# Usage
tweets = await fetch_with_backoff(
    lambda: client.search_tweet('query', 'Latest')
)

import asyncio
from datetime import datetime, timedelta

class CachedClient:
    def __init__(self, client):
        self.client = client
        self.cache = {}
    
    async def get_user_cached(self, screen_name, cache_duration=300):
        """Get user with 5-minute cache."""
        now = datetime.now()
        
        if screen_name in self.cache:
            user, cached_at = self.cache[screen_name]
            if now - cached_at < timedelta(seconds=cache_duration):
                return user
        
        # Fetch from API
        user = await self.client.get_user_by_screen_name(screen_name)
        self.cache[screen_name] = (user, now)
        return user

# Bad: Multiple separate requests
for user_id in user_ids:
    user = await client.get_user_by_id(user_id)
    await asyncio.sleep(1)  # Still wasteful

# Good: Use pagination features
# Many Twikit methods return Result objects that efficiently handle batching
followers = await user.get_followers(count=100)  # Get 100 at once
for follower in followers:
    process_follower(follower)

class RateLimitTracker:
    def __init__(self, client):
        self.client = client
        self.request_counts = {}
        self.window_start = time.time()
    
    async def search_tweet(self, query, product, **kwargs):
        """Track search requests."""
        endpoint = 'SearchTimeline'
        
        # Reset counter every 15 minutes
        if time.time() - self.window_start > 900:
            self.request_counts = {}
            self.window_start = time.time()
        
        # Increment counter
        self.request_counts[endpoint] = self.request_counts.get(endpoint, 0) + 1
        
        print(f'{endpoint}: {self.request_counts[endpoint]}/50 requests used')
        
        return await self.client.search_tweet(query, product, **kwargs)

# Initial request
tweets = await client.search_tweet('python', 'Latest', count=20)

# Process first batch
for tweet in tweets:
    print(tweet.text)

# Get more results (uses cursor internally)
more_tweets = await tweets.next()
for tweet in more_tweets:
    print(tweet.text)

# Continue paginating
even_more = await more_tweets.next()

# This method yields tweets as they're fetched
async for tweet in client.get_user_tweets_by_id(user_id, 'Tweets'):
    print(tweet.text)
    
    # Each iteration may trigger an API request when fetching the next batch
    # Be mindful of rate limits (50 requests/15min for UserTweets)

try:
    tweets = await client.search_tweet('query', 'Latest')
except TooManyRequests as e:
    print(f'Rate limit hit!')
    print(f'Reset at: {e.rate_limit_reset}')
    print(f'Message: {e}')
    print(f'Headers: {e.headers}')