d4science-python-notebooks-examples/refresh-token-and-call.ipynb

How to issue a request to a D4Science service¶

In order to issue a request to an HTTP service hosted on D4Science infrastructure, you need to have a valid UMA OAuth ACCESS_TOKEN.

You can get one from the portlet avaiable on the Science Gateway where your VRE is deployed.

However, this token has a limited duration and it needs to be refreshed through a longer lived REFRESH_TOKEN, that you can find from the same porlet.

In the following colab, you can find an example on how to deal with the expiration of an Access Token from Python.

Let's start by importing some required libraries:

Note: if you try to execute this code outside of Google Colab, you probably need to install requests and pyjwt with

pip install requests pyjwt==2.8.0

In [ ]:

import requests
import jwt
from datetime import datetime

and configure some variabiles with the values copied from the Personal Web Token Portlet and the service_url you want to issue a request to.

In [ ]:

ACCESS_TOKEN = "<YOUR_PERSONAL_OAUTH_ACCESS_TOKEN>"    # Get your Personal Access (UMA) Token from the portlet
REFRESH_TOKEN = "<YOUR_PERSONAL_OAUTH2_REFRESH_TOKEN>" # Get your OAuth2 Refresh Token from the portlet
CLIENT_ID = "sobigdata.d4science.org"                  # Get from the Refresh parameters section of the porlet
SERVICE_URL = "https://netme-sobigdata.d4science.org"  # Base URL of the service you want to get access

refresh_token_url = "https://accounts.d4science.org/auth/realms/d4science/protocol/openid-connect/token" # Get from the Refresh token URL section of the porlet

The refresh_token function is used to get a new access token by means of a refresh token

In [ ]:

def refreshToken(rtoken):
    if tokenIsExpired(rtoken):
      print(f"Refresh token has expired too {expirationTime(rtoken)}")
      exit(-1)
    data = { "grant_type": "refresh_token", "client_id": CLIENT_ID, "refresh_token": rtoken }
    response = requests.post(refresh_token_url, data=data)
    r = response.json()
    if response.status_code != 200:
        # print(yaml.dump(r))
        exit(-1)
    return r['access_token']

Two other helper functions, expirationTime and tokenIsExpired can be used to get the expiration time of a given token and check if a token is expired or not:

In [ ]:

def expirationTime(token):
    decoded = jwt.decode(jwt=token, options={"verify_signature": False})
    return datetime.fromtimestamp(decoded['exp'])

def tokenIsExpired(token):
    return datetime.now() > expirationTime(token)

Finally, the function executeServiceRequest will issue the HTTP request to the given service path. It will check if the ACCESS_TOKEN is expired and in that case it will refresh it. It will return back the response from the service

In [ ]:

def executeServiceRequest(path):
    global ACCESS_TOKEN
    if tokenIsExpired(ACCESS_TOKEN):
        print(f"ACCESS token expired in {expirationTime(ACCESS_TOKEN)}")
        print("Refreshing token...")
        ACCESS_TOKEN = refreshToken(REFRESH_TOKEN)
        print(f"new ACCESS token expiration {expirationTime(ACCESS_TOKEN)}")

    print(f"Call to Service {SERVICE_URL}{path}")
    response = requests.post(f"{SERVICE_URL}{path}", headers={"Authorization": f"Bearer {ACCESS_TOKEN}"})
    r = response.json()
    if response.status_code != 200:
        print(f"\nResponse: {r}")
    else:
        raise Exception("Unable Contact service")
    return r

So let's try to use the NetME service, in particular the network generation feature, available at the network_generation path.

So we just need to call the executeServiceRequest function with the correct path.

Please also add any other header or payload data the requested service requires

In [ ]:

response = executeServiceRequest('/network_generation')

ACCESS token expired in 2024-03-13 15:07:34
Refreshing token...
new ACCESS token expiration 2024-03-14 14:25:28
Call to Service https://netme-sobigdata.d4science.org/network_generation

Response: {'detail': [{'type': 'missing', 'loc': ['body'], 'msg': 'Field required', 'input': None, 'url': 'https://errors.pydantic.dev/2.6/v/missing'}]}