Commit 05599e4a authored by jackie / Andrea Ida Malkah Klaura's avatar jackie / Andrea Ida Malkah Klaura
Browse files

use tank package; add api call demos

parent 89ca6d55
......@@ -3,11 +3,17 @@
import argparse
from config import config as cfg
from steering import implicit
import tank.session
import requests
parser = argparse.ArgumentParser(
description="Demo for the OIDC client functions/stubs for AURA.")
parser.add_argument("-v", "--verbosity", action="count",
help="Activate verbose output. Use several times to increase")
parser.add_argument("-s", "--steering", action="store_true",
help="Make an authorised test call against the steering API")
parser.add_argument("-t", "--tank", action="store_true",
help="Make an authorised test call against the tank API")
args = parser.parse_args()
if args.verbosity:
......@@ -15,10 +21,72 @@ if args.verbosity:
else:
cfg["verbosity"] = 0
# we have to set the response_type according to our flow type
parameters = {
"response_type": "id_token token",
}
# now we can retrieve an access token from steering
oidc = implicit.get_token(cfg, parameters)
if cfg["verbosity"] == 0:
print(oidc)
print("access token:", oidc["access_token"])
# with the steering access token we can also retrieve a session token from tank
tank_session = tank.session.get_session_token(cfg, oidc["access_token"])
if cfg["verbosity"] == 0:
print("tank session token:", tank_session["token"])
## this is to demonstrate a steering API call with the retrieved access token
if args.steering:
print()
print("Accessing steering userinfo endpoint, to retrieve user ID.")
url = cfg["base_url"] + cfg["userinfo_endpoint"]
headers = {
"User-Agent": cfg["user_agent"],
"Authorization": "Bearer " + oidc["access_token"]
}
try:
response = requests.get(url, headers=headers)
except:
e = sys.exc_info()
print(e[0].__name__, ':', e[1])
sys.exit(1)
resp = response.json()
uid = resp["sub"]
privileged = resp["privileged"]
print("Accessing steering shows endpoint.")
url = cfg["base_url"] + "/api/v1/shows"
payload = { "owner": uid }
try:
response = requests.get(url, headers=headers, params=payload)
except:
e = sys.exc_info()
print(e[0].__name__, ':', e[1])
sys.exit(1)
shows = response.json()
print("Shows in steering you own:")
for show in shows:
print(" [" + str(show["id"]) + "]", show["name"])
if privileged:
print("As you are a privileged user you have access to all other shows too.")
## this is to demonstrate a tank API call with the retrieved access token
if args.tank:
print()
print("Accessing tank shows endpoint.")
url = cfg["tank_base"] + "/api/v1/shows"
headers = {
"User-Agent": cfg["user_agent"],
"Authorization": "Bearer " + tank_session["token"]
}
try:
response = requests.get(url, headers=headers)
except:
e = sys.exc_info()
print(e[0].__name__, ':', e[1])
sys.exit(1)
shows = response.json()
print("Shows you have access to on tank, that already have some content:")
for show in shows["results"]:
print(" -", show["name"])
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment