Making an API call in Python with an API that requires a bearer token

Question

Looking for some help with integrating a JSON API call into a Python program.

I am looking to integrate the following API into a Python .py program to allow it to be called and the response to be printed.

The API guidance states that a bearer token must be generated to allow calls to the API, which I have done successfully. However I am unsure of the syntax to include this token as bearer token authentication in Python API request.

I can successfully complete the above request using cURL with a token included. I have tried "urllib" and "requests" routes but to no avail.

Full API details: IBM X-Force Exchange API Documentation - IP Reputation

Talithatalk

Answer 1

It just means it expects that as a key in your header data

import requests
endpoint = ".../api/ip"
data = {"ip": "1.1.2.3"}
headers = {"Authorization": "Bearer MYREALLYLONGTOKENIGOT"}

print(requests.post(endpoint, data=data, headers=headers).json())

Hetaera

The above throws the following syntax error: Traceback (most recent call last): File "bearerreturn.py", line 6, in <module> print requests.post(endpoint,data=data,headers=headers).json() TypeError: 'dict' object is not callable Code Below: import requests endpoint = "https://xforce-api.mybluemix.net:443/api/ip" data = {"ip":"1.1.2.3"} headers = {"Bearer token":"TOKEN WAS INSERTED HERE"} print requests.post(endpoint,data=data,headers=headers).json() Any Ideas? – Talithatalk

you have an old version of requests ... json is a dict in your version and not a function requests.post(...).json ... dont call it – Hetaera

Thanks Joran Beasley. Updated Requests library via pip and this allowed me to keep original syntax. However now when I run the above it outputs this .json response: {u'error': u'Not authorized. Access is only allowed via https://exchange.xforce.ibmcloud.com/#/'} This is same as if I hit the URL directly in a browser. Am I missing something with token or the way endpoint is configured? Code: import requests endpoint = "https://xforce-api.mybluemix.net:443/ipr/" data = {"ip":"1.1.2.3"} headers = {"Bearer token":"TOKEN_HERE"} print requests.post(endpoint,data=data,headers=headers).json() – Talithatalk

unfortunately I cant really help with that... it is either a bad endpoint or your credentials are invalid (are you using their example token, that is only configured for their url?) or perhaps you need to put your app url in their deleoper panel for your code ... chances are thats your first token ... you need to exchange token for a refresh token that you can then use to get a more permanent token (at least thats how oauth2 usually works..) – Hetaera

whoops looks like i had the header wrong try the updated code – Hetaera

I am using a uniquely generated token that was generated via their guide: xforce-api.mybluemix.net:443/auth/anonymousToken – Talithatalk

well if i find myself with lots of free time ill give it a shot and see if i can get back to you ... can you give me a link to their guide? – Hetaera

Tried updated code: More success. Still throws following error though. {u'error': u'Not found.'} Code: import requests endpoint = "https://xforce-api.mybluemix.net:443/ipr/" data = {"ip":"8.8.8.8"} headers = {"Authorization":"Bearer REALLY_LONG_TOKEN_HERE"} print requests.post(endpoint,data=data,headers=headers).json() API Documentation Here: xforce-api.mybluemix.net/doc/#!/Authentication/… Thanks for your help so far! It really is much appreciated Joran. – Talithatalk

Answer 2

If you are using requests module, an alternative option is to write an auth class, as discussed in "New Forms of Authentication":

import requests

class BearerAuth(requests.auth.AuthBase):
    def __init__(self, token):
        self.token = token
    def __call__(self, r):
        r.headers["authorization"] = "Bearer " + self.token
        return r

and then can you send requests like this

response = requests.get('https://www.example.com/', auth=BearerAuth('3pVzwec1Gs1m'))

which allows you to use the same auth argument just like basic auth, and may help you in certain situations.

Danita

This might be useful with zeep too. It uses requests.auth type of authorizations (for http headers auth, not soap headers) – Meitner

Answer 3

The token has to be placed in an Authorization header according to the following format:

Authorization: Bearer [Token_Value]

Code below:

import urllib2
import json

def get_auth_token():
    """
    get an auth token
    """
    req=urllib2.Request("https://xforce-api.mybluemix.net/auth/anonymousToken")
    response=urllib2.urlopen(req)
    html=response.read()
    json_obj=json.loads(html)
    token_string=json_obj["token"].encode("ascii","ignore")
    return token_string

def get_response_json_object(url, auth_token):
    """
    returns json object with info
    """
    auth_token=get_auth_token()
    req=urllib2.Request(url, None, {"Authorization": "Bearer %s" %auth_token})
    response=urllib2.urlopen(req)
    html=response.read()
    json_obj=json.loads(html)
    return json_obj

Ladyfinger

For Python3 : req = urllib.request.Request(urlstr, None, {"Authorization": "Bearer %s" % enc_authstr}) response = urllib.request.urlopen(req) – Coprophagous

Answer 4

Here is full example of implementation in cURL and in Python - for authorization and for making API calls

cURL

1. Authorization

You have received access data like this:

Username: johndoe

Password: zznAQOoWyj8uuAgq

Consumer Key: ggczWttBWlTjXCEtk3Yie_WJGEIa

Consumer Secret: uuzPjjJykiuuLfHkfgSdXLV98Ciga

Which you can call in cURL like this:

curl -k -d "grant_type=password&username=Username&password=Password" \

                    -H "Authorization: Basic Base64(consumer-key:consumer-secret)" \

                       https://somedomain.test.com/token

or for this case it would be:

curl -k -d "grant_type=password&username=johndoe&password=zznAQOoWyj8uuAgq" \

                    -H "Authorization: Basic zzRjettzNUJXbFRqWENuuGszWWllX1iiR0VJYTpRelBLZkp5a2l2V0xmSGtmZ1NkWExWzzhDaWdh" \

                      https://somedomain.test.com/token

Answer would be something like:

{
    "access_token": "zz8d62zz-56zz-34zz-9zzf-azze1b8057f8",
    "refresh_token": "zzazz4c3-zz2e-zz25-zz97-ezz6e219cbf6",
    "scope": "default",
    "token_type": "Bearer",
    "expires_in": 3600
}

2. Calling API

Here is how you call some API that uses authentication from above. Limit and offset are just examples of 2 parameters that API could implement. You need access_token from above inserted after "Bearer ".So here is how you call some API with authentication data from above:

curl -k -X GET "https://somedomain.test.com/api/Users/Year/2020/Workers?offset=1&limit=100" -H "accept: application/json" -H "Authorization: Bearer zz8d62zz-56zz-34zz-9zzf-azze1b8057f8"

Python

Same thing from above implemented in Python. I've put text in comments so code could be copy-pasted.

# Authorization data

import base64
import requests

username = 'johndoe'
password= 'zznAQOoWyj8uuAgq'
consumer_key = 'ggczWttBWlTjXCEtk3Yie_WJGEIa'
consumer_secret = 'uuzPjjJykiuuLfHkfgSdXLV98Ciga'
consumer_key_secret = consumer_key+":"+consumer_secret
consumer_key_secret_enc = base64.b64encode(consumer_key_secret.encode()).decode()

# Your decoded key will be something like:
#zzRjettzNUJXbFRqWENuuGszWWllX1iiR0VJYTpRelBLZkp5a2l2V0xmSGtmZ1NkWExWzzhDaWdh


headersAuth = {
    'Authorization': 'Basic '+ str(consumer_key_secret_enc),
}

data = {
  'grant_type': 'password',
  'username': username,
  'password': password
}

## Authentication request

response = requests.post('https://somedomain.test.com/token', headers=headersAuth, data=data, verify=True)
j = response.json()

# When you print that response you will get dictionary like this:

    {
        "access_token": "zz8d62zz-56zz-34zz-9zzf-azze1b8057f8",
        "refresh_token": "zzazz4c3-zz2e-zz25-zz97-ezz6e219cbf6",
        "scope": "default",
        "token_type": "Bearer",
        "expires_in": 3600
    }

# You have to use `access_token` in API calls explained bellow.
# You can get `access_token` with j['access_token'].


# Using authentication to make API calls   

## Define header for making API calls that will hold authentication data

headersAPI = {
    'accept': 'application/json',
    'Authorization': 'Bearer '+j['access_token'],
}

### Usage of parameters defined in your API
params = (
    ('offset', '0'),
    ('limit', '20'),
)

# Making sample API call with authentication and API parameters data

response = requests.get('https://somedomain.test.com/api/Users/Year/2020/Workers', headers=headersAPI, params=params, verify=True)
api_response = response.json()

Fotina

This works. However, I am facing a weird issue. First requests call always gives 401 and second call (with same data) works! Is there a way to avoid the 2nd call? For now I am checking the status_code of first call and if it's 401 I am making a second call. – Retrieval

regarding the first line of this answer: you don't have to scream to draw attention for upvotes (-1) – Expedite

Answer 5

import json
import os
import requests

def lambda_handler(event, context):
    print(event)
    item = list(map(lambda x: x['detail']['item'], event['inputData']))
    print("item List :", item)
    consumer_key = os.getenv('consumer_key')
    consumer_secret = os.getenv('consumer_secret')
    entitlement_url=os.getenv('entitlement_url')
    storage_url=os.getenv('storage_url')
    access_token = get_jwt_token(consumer_key,consumer_secret,entitlement_url)
    print("Response from entitlement: ", access_token)
    for listID in list:
        print("listID: ", listID)
        response = get_storage_service(access_token,storage_url,listID)
        print("Response from storage: ", response.text)

    return "Success"

def get_jwt_token(consumer_key, consumer_secret, url):
    data = 'grant_type=client_credentials&client_id=' + consumer_key + '&client_secret=' + consumer_secret
    header = {"Content-type": "application/x-www-form-urlencoded"}
    try:
        response = requests.post(url, data=data, headers=header)
        access_token = json.loads(response.text)
        final_response=access_token['access_token']

    except requests.exceptions as err:
        print(err)
        final_response = 'error'
    return final_response


def get_storage_service(jwt_token, url, list_id):
    final_url = url + list_id + "/data"
    print("Final url is :", final_url)
    headers_api = {
        'Authorization': 'Bearer ' + jwt_token

    }
    try:
        response = requests.get(url=final_url, headers=headers_api)
    except requests.exceptions as err:
        print(err)
        response = 'error'
    return response

using enviornment variable

Postbellum