Wednesday 15 February 2017

Twitter Application-Only Authentication in Python

Twitter provides a number of REST API endpoints. All endpoints require requests to be authenticated. Authentication can be done with any of the two:

Application-only authentication is useful in cases where the user context is not available or is simply not required. Not all endpoints support it, since many of them would need user grant before a request is made on a user's behalf. Such endpoints could be for creating tweets for a user, deleting them, et cetera. A small subset of endpoints does not require user context. Search Tweets is one such endpoint. In the rest of the post I'll create a client that implements application-only authentication flow in Python.

You're going to need the Requests package. You must also register a Twitter app at https://apps.twitter.com/ to obtain Consumer Key and Consumer Secret for the application.

Encoding Consumer Key and Secret

Twitter requires the consumer key and secret to be encoded before requesting the Bearer token. Bearer token credentials can be obtained easily:

  • URL encode both Consumer Key and Secret
  • Concatenate them with a semicolon in between
  • Encode the result in Base64

from urllib.parse import quote_plus
from base64 import b64encode


def get_bearer_credentials(consumer_key, consumer_secret):
    encoded_ck = quote_plus(consumer_key)
    encoded_cs = quote_plus(consumer_secret)

    credentials = '{}:{}'.format(encoded_ck, encoded_cs)

    b64credentials = b64encode(credentials.encode('ascii'))
    return b64credentials.decode('ascii')

Requesting the Bearer token

Bearer token can be obtained by making a POST request to https://api.twitter.com/oauth2/token with the Authorization header containing the Bearer credentials. Twitter implements Client Credentials Grant of OAuth2 as defined in RFC-6749. As per the RFC, the server must respond with an Access token for a valid client request (having valid credentials). The Access token can be of certain types, Bearer is one of them and is what Twitter uses. A valid request for Bearer token must have:

  • Authorization header set to Basic {{ credentials }}
  • Content-Type header set to application/x-www-form-urlencoded;charset=UTF-8
  • grant_type=client_credentials as request body

import requests

OAUTH2_TOKEN_URL = 'https://api.twitter.com/oauth2/token'


def get_access_token(bearer_credentials):
    headers = {
        'Authorization': 'Basic {}'.format(bearer_credentials),
        'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'
    }
    data = 'grant_type=client_credentials'

    response = requests.post(OAUTH2_TOKEN_URL, data=data, headers=headers)

    if response.status_code != requests.codes.ok:
        raise Exception('Invalid request or invalid credentials')

    response_body = response.json()
    if response_body.get('token_type', '') == 'bearer':
        access_token = response_body.get('access_token')
        return access_token
    else:
        raise Exception(('Invalid token type of returned access token. Token type')
                       ('is not bearer.'))

Making an authenticated request

An authenticated request can be constructed by setting the Authorization header to Bearer {{ access_token }}. The following is an example request to fetch popular tweets containing the hashtag twitter: 

import requests

SEARCH_TWEETS_URL = 'https://api.twitter.com/1.1/search/tweets.json'

# `access_token` obtained in the above step
headers = {'Authorization': 'Bearer {}'.format(access_token)}
payload = {'q': '#twitter', 'count': 10, 'result_type': 'popular'}


response = requests.get(SEARCH_TWEETS_URL, params=payload, headers=headers)

if response.status_code != requests.codes.ok:
    raise Exception('Invalid request')

# Twitter Statuses (tweets)
status_list = response.json().get('statuses', [])

Invalidating Bearer token

A Bearer token can be invalidated by making a POST request to https://api.twitter.com/oauth2/invalidate_token with the Authorization header and body correctly set. 

import requests

INVALIDATE_URL = 'https://api.twitter.com/oauth2/invalidate_token'


def invalidate_token(bearer_credentials, access_token):
    headers = {
        'Authorization': 'Basic {}'.format(bearer_credentials)}
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    data = 'access_token={}'.format(access_token)

    response = requests.post(INVALIDATE_URL, data=data, headers=headers)

    if response.status_code != requests.codes.ok:
        raise Exception('An error occurred. Access token was not invalidated.')

    print('Access token was invalidated.')

Handling common errors

Twitter's documentation describes some common errors that might occur while getting or invalidating a Bearer token, or while performing some action through an authenticated request. The code below is a simple check for these errors: 

# a `response` after a request
if response.status_code != requests.codes.ok:
    errors = response.json()
    for error in errors:
        print('Code: {}, Message: {}'.format(error.code, error.message))

---

I've created a simple application using the above APIs: shivamMg/kayak


3 comments: