In [1]:
from requests_oauthlib import OAuth2Session
import requests
import base64 # To conver password to base64 encoded string
import json
In [535]:
# This guide show how to generate consumer key and consumer secret or an application in WSO2 API manager and use those keys to ultimatly access APIs
# Consumer key and consumer secret can be generated via WSO2 API manager Store UI and i'm using WSO2 APIM store REST API(https://docs.wso2.com/display/AM200/apidocs) for generating those.
# Will mention/point to DOCs how to do the same thing using UI in nessary places.
In [574]:
host = "localhost"
web_http_port = 9763 # Handle through web application (Tomcat)
web_https_port = 9443
synapse_https_port = 8243 # Handle through APIM Gateway server (ESB)
In [537]:
# This contains the default hostname and port used for APIM , change them according to your enviroment
In [551]:
dcr_endpoint = "http://{host}:{port}/client-registration/v0.10/register".format(host=host, port=web_http_port) # dynamic client registration endpoint
authorization_basic = base64.b64encode(b'admin@knnect.com:admin')
payload = {
"callbackUrl": "http://localhost:5000/callback",
"clientName": "testing_3",
"tokenScope": "Production",
"owner": "admin@knnect.com",
"grantType": "password refresh_token",
"saasApp": True
}
headers = {
"Content-Type" : "application/json",
"Authorization" : "{type} {value}".format(type="Basic", value=authorization_basic.decode())
}
In [552]:
response = requests.post(dcr_endpoint, data=json.dumps(payload), headers=headers, json=True)
if not response.status_code == 201:
print("ERROR Code: {}\nSomething went wrong, check the console logs\n{}".format(response.status_code, json.loads(response.content.decode())))
In [553]:
data = json.loads(response.content.decode())
print("clientSecret = {}\nclientId = {}".format(data['clientSecret'], data['clientId']))
In [554]:
json.loads(data['jsonString'])
Out[554]:
In [555]:
# Now we have clientSecret and clientId which can be used to
In [556]:
auth_api = "https://{host}:{port}/authorize".format(host=host, port=synapse_https_port) # Authorization API URL
scope = "all"
payload = {
"response_type": "code",
"client_id": data['clientId'],
"scope": scope,
"redirect_uri": "http://localhost:5000/callback"
}
headers = {
"Content-Type" : "application/x-www-form-urlencoded"
}
print(auth_api)
In [557]:
response = requests.get(auth_api, params=payload, verify=False, headers=headers)
In [558]:
# At this point we user has been redirected to authentication login page to autharise the client in this case the python agent to grant access to give authorization code, Since this is a browser intermidate processe at this point we need a browser to give username and password
In [559]:
# So We try the another method of getting access token which is Generating Access Tokens with User Credentials - Password Grant Type
In [560]:
token_api = "https://{host}:{port}/token".format(host=host, port=synapse_https_port)
clientSecret_clientId = '{}:{}'.format(data['clientId'], data['clientSecret'])
authorization_basic = base64.b64encode(clientSecret_clientId.encode())
scope = "apim:subscribe"
payload = {
"grant_type": "password",
"username": "admin",
"password": "admin",
"scope" : scope
}
headers = {
"Content-Type" : "application/x-www-form-urlencoded",
"Authorization" : "Basic {}".format(authorization_basic.decode())
}
print(headers)
print(payload)
In [561]:
response = requests.post(token_api, data=payload, verify=False, headers=headers)
In [568]:
print(response.status_code)
access_token = response.json()['access_token']
In [575]:
# Invoke service API using the newly generated access token
service_url = "https://{}:{}/api/am/store/v0.10/applications".format(host, web_https_port)
headers = {
"Authorization": "{type} {value}".format(type="Bearer", value=access_token)
}
print(service_url)
print(headers)
In [576]:
service_response = requests.get(service_url, headers=headers, verify=False)
In [577]:
service_response.json()
Out[577]:
In [1]:
# ****************** Testing class not related to above work ***************************
In [2]:
class Testing(object):
static_var = 5
def method_1(self):
self.static_var += 1
def method_2(self):
Testing.static_var +=1
In [3]:
t = Testing()
In [9]:
t.static_var, Testing.static_var
Out[9]:
In [21]:
t.method_1()
In [26]:
temp = """
<?xml version='1.0' encoding='UTF-8'?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:wsa="http://www.w3.org/2005/08/addressing">
<soapenv:Body>
<m0:getQuote xmlns:m0="http://services.samples">
<m0:request>
<m0:symbol>{}</m0:symbol>
</m0:request>
</m0:getQuote>
</soapenv:Body>
</soapenv:Envelope>
"""
baila = ""
for i in range(1000000):
baila += "IBM{}".format(i)
In [27]:
with open(r'sample.xml', 'w') as f:
f.write(temp.format(baila))
In [2]:
graph_api = "https://graph.facebook.com"
context = '/me'
# context = '/me/picture'
access_token = "EAAE7mlmuKdsBAO6MZBYuteJ1u6IXXiHohF0LxqUpwTg7MWVKFORK0ZBA8mxyjVZAb5iVN8Svph2EGYptosLn01VRoreMl53JMMNhLcnt2FrQdciN5Xg43bdasGZCH8LBVOQNivfYGu9xaOKySjEUwFpt68bXOIanQqvjDLcBCwZDZD"
header = {
'Authorization': 'Bearer {}'.format(access_token),
'Accept': 'application/json'
}
request_url = wso2_api + context
response = requests.get(request_url, headers=data)
In [14]:
response.json()
Out[14]:
In [4]:
wso2_api = "https://localhost:8243/t/bookstore.com/bookstore/1.0.0"
context = '/book'
# context = '/me/picture'
access_token = "9ea3f8a1-60c1-3698-a2a6-8ad5f2b8b0aa"
headers = {
'Authorization': 'Bearer {}'.format(access_token),
'Accept': 'application/json'
}
request_url = wso2_api + context
response = requests.get(request_url, headers=headers, verify=False)
In [6]:
response.json()
Out[6]:
In [ ]: