【问题标题】:Making an API call in Python with an API that requires a bearer token使用需要不记名令牌的 API 在 Python 中进行 API 调用
【发布时间】:2023-03-24 00:54:01
【问题描述】:

寻求将 JSON API 调用集成到 Python 程序中的帮助。

我希望将以下 API 集成到 Python .py 程序中,以允许调用它并打印响应。

API 指南指出必须生成不记名令牌以允许调用 API,我已经成功完成了。但是我不确定在 Python API 请求中包含此令牌作为不记名令牌身份验证的语法。

我可以使用包含令牌的 cURL 成功完成上述请求。我尝试过“urllib”和“requests”路由,但无济于事。

完整的 API 详细信息:IBM X-Force Exchange API Documentation - IP Reputation

【问题讨论】:

    标签: python python-2.7 curl ibm-cloud pycurl


    【解决方案1】:

    这只是意味着它希望将其作为标题数据中的键

    import requests
    endpoint = ".../api/ip"
    data = {"ip": "1.1.2.3"}
    headers = {"Authorization": "Bearer MYREALLYLONGTOKENIGOT"}
    
    print(requests.post(endpoint, data=data, headers=headers).json())
    

    【讨论】:

    • 上面抛出以下语法错误: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 下面的代码: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?
    • 你有一个旧版本的 requests ... json 是你版本中的一个字典,而不是一个函数 requests.post(...).json ... 不要调用它
    • 感谢乔兰·比斯利。通过 pip 更新了 Requests 库,这使我能够保留原始语法。但是现在当我运行上面的代码时,它会输出这个 .json 响应:{u'error': u'Not authorized. Access is only allowed via https://exchange.xforce.ibmcloud.com/#/'} 这就像我直接在浏览器中点击 URL 一样。我是否缺少令牌或端点的配置方式?代码: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()
    • 不幸的是,我对此无能为力...它要么是一个错误的端点,要么您的凭据无效(您是否使用他们的示例令牌,只为他们的 url 配置?)或者您可能需要将您的应用程序 url 放在您的代码的 deleoper 面板中...这可能是您的第一个令牌...您需要将令牌交换为刷新令牌,然后您可以使用它来获取更永久的令牌(至少 oauth2 就是这样通常有效..)
    • 哎呀看起来我的标题错误尝试更新的代码
    【解决方案2】:

    如果您使用requests 模块,另一种选择是编写身份验证类,如“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
    

    然后你可以发送这样的请求

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

    它允许您使用与基本身份验证相同的 auth 参数,并且可能在某些情况下为您提供帮助。

    【讨论】:

    • 这可能对 zeep 也有用。它使用 requests.auth 类型的授权(用于 http 标头身份验证,而不是 soap 标头)
    【解决方案3】:

    令牌必须按照以下格式放置在授权标头中:

    授权:承载 [Token_Value]

    代码如下:

    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
    

    【讨论】:

    • 对于 Python3:req = urllib.request.Request(urlstr, None, {"Authorization": "Bearer %s" % enc_authstr}) response = urllib.request.urlopen(req)
    【解决方案4】:

    这里是在 cURL 和 Python 中实现的完整示例 - 用于授权和进行 API 调用

    卷曲

    1.授权

    您收到了这样的访问数据:

    Username: johndoe
    
    Password: zznAQOoWyj8uuAgq
    
    Consumer Key: ggczWttBWlTjXCEtk3Yie_WJGEIa
    
    Consumer Secret: uuzPjjJykiuuLfHkfgSdXLV98Ciga
    

    你可以像这样在 cURL 中调用它:

    curl -k -d "grant_type=password&username=Username&password=Password" \
    
                        -H "Authorization: Basic Base64(consumer-key:consumer-secret)" \
    
                           https://somedomain.test.com/token
    

    或者对于这种情况,它会是:

    curl -k -d "grant_type=password&username=johndoe&password=zznAQOoWyj8uuAgq" \
    
                        -H "Authorization: Basic zzRjettzNUJXbFRqWENuuGszWWllX1iiR0VJYTpRelBLZkp5a2l2V0xmSGtmZ1NkWExWzzhDaWdh" \
    
                          https://somedomain.test.com/token
    

    答案是这样的:

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

    2。调用 API

    这是您如何调用一些使用上述身份验证的 API。 Limitoffset 只是 API 可以实现的 2 个参数的示例。 您需要在"Bearer " 之后插入上面的access_token。因此,您可以使用上面的身份验证数据调用一些API:

    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

    上面的同样的东西用 Python 实现。我已将文本放入 cmets,以便可以复制粘贴代码。

    # 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()
    

    【讨论】:

    • 这行得通。但是,我面临一个奇怪的问题。第一个requests 电话总是给401,第二个电话(具有相同的数据)有效!有没有办法避免第二次通话?现在我正在检查第一次通话的status_code,如果是401,我正在拨打第二次电话。
    【解决方案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
    

    使用环境变量

    【讨论】:

      相关资源