How to Implement Twitter Premium Search API for Historical Tweets in Python

Install Required Libraries

• Ensure you have Python 3.x installed in your environment. Use pip to install the required packages.

 

• Install the requests library for making HTTP requests to the Twitter API.

 

• Install the tweepy library, which is commonly used for interfacing with Twitter. However, for Premium API, we will use requests primarily.

pip install requests tweepy

Configure Environment Variables

• Store your Twitter API credentials securely. Use environment variables or a configuration file to manage them.

 

• Set up the following variables: TWITTER_CONSUMER_KEY, TWITTER_CONSUMER_SECRET, TWITTER_ACCESS_TOKEN, and TWITTER_ACCESS_TOKEN\_SECRET.

Build API Request

• Focus on constructing a proper HTTP request for querying historical tweets using the Twitter Premium Search API.

 

• Use the appropriate endpoint for the API tier you are using (e.g., Full Archive or 30-Day). Example endpoint: https://api.twitter.com/1.1/tweets/search/fullarchive/[environment].json.

import os
import requests
from requests.auth import HTTPBasicAuth

def get_headers():
    return {
        'Authorization': f'Bearer {os.environ["TWITTER_BEARER_TOKEN"]}',
        'Content-Type': 'application/json',
    }

Define Search Parameters

• Specify the query parameters according to the type of data you need (keywords, from date, to date, etc.).

 

• Twitter Premium API uses a JSON body with fields like 'query', 'maxResults', and 'fromDate'.

def build_search_request():
    search_request = {
        'query': 'from:jack',
        'maxResults': '10',
        'fromDate': '202201010000',
        'toDate': '202201312359'
    }
    return search_request

Make the API Call

• Use the requests library to send a POST request with the JSON body and headers configured earlier.

 

• Ensure proper error-handling to capture issues like exceeding rate limits or authentication errors.

def fetch_tweets():
    url = "https://api.twitter.com/1.1/tweets/search/fullarchive/your_environment_name.json"
    try:
        response = requests.post(url, headers=get_headers(), json=build_search_request())
        response.raise_for_status()  # Raises exception for HTTP error codes
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"An error occurred: {e}")
        return None

Process the Retrieved Data

• Once the tweets are fetched, parse the JSON response. You’ll typically look for the key 'results' which contains the retrieved tweets.

 

• Iterate through the results and extract the required information such as tweet content, creation date, user name, etc.

def process_tweets(data):
    if not data:
        return

    for tweet in data.get('results', []):
        print(f"Date: {tweet['created_at']}")
        print(f"User: {tweet['user']['name']}")
        print(f"Tweet: {tweet['text']}")
        print("-" * 40)

Execute the Code

• Bring together the components in a main execution block or script that prepares the search request, makes the API call, and processes the results.

 

• Use the script to conduct your required historical data searches on Twitter using the Premium API.

def main():
    tweets_data = fetch_tweets()
    process_tweets(tweets_data)

if __name__ == "__main__":
    main()