CPD-04 Services.ipynb

Kernel: Python 3.10

Services and permissions

This notebook shows how we can get information about services using API calls.

The use of cpd-cli is also demonstrated.

The best source of information on how to use the API calls appears to be: Using the Volumes API

import json
import sys
import requests
import warnings
import os
from datetime import datetime
import inspect

import zipfile
from io import BytesIO

warnings.filterwarnings("ignore") # one of "error", "ignore", "always", "default", "module", or "once"

# Determine if you are running the notebook on CPDaaS or on a CPD instance
platform = "cpdaas"
if "USER_ID" in os.environ :
    platform = "cpd"

Make sure to set the variables in the next cell

cpd_url, username, password

# cluster URL, make sure it ends with "/", and no "zen" ending
# example:
#cpd_url = "https://cpd-cpd.ai-governance-12345a678e90addd123c4567c8f9a012-3456.us-east.containers.appdomain.cloud/"
cpd_url = "<CPD instance URL>"

username = "<username>"
password = "<password>"

Download and install cpd-cli

Also setup the environment to use the command

# Download the cpd-cli utility
url = "https://github.com/IBM/cpd-cli/releases/download/v12.0.4/cpd-cli-linux-EE-12.0.4.tgz"
filename = 'cpd-cli-linux-EE-12.0.4.tgz'
r = requests.get(url)

f = open(filename,'wb')
nb_bytes = f.write(r.content)
f.close()
!tar xzf cpd-cli-linux-EE-12.0.4.tgz
!rm -rf cpd-cli-linux-EE-12.0.4.tgz
!ln -s cpd-cli-linux-EE-12.0.4-57/cpd-cli cpd-cli
!ln -s cpd-cli-linux-EE-12.0.4-57/plugins plugins
!ln -s cpd-cli-linux-EE-12.0.4-57/LICENSES LICENSES

Make sure to add the proper API key

os.environ['CPD_API_KEY'] = "<YOUR_API_KEY>"

os.environ['CPD_PROFILE_URL']="https://cpd-cpd.ai-governance-94074a334e51addd457c5646c0f9a073-0000.us-east.containers.appdomain.cloud"
os.environ['CPD_ADMIN_USER'] = "user"
os.environ['CPD_PROFILE_NAME'] = "user"
os.environ['CPD_API_KEY'] = "1dLe9NHcxBzPx0KQdHMEQ5NEsdUzDNuZ2iG8FzB1"

!./cpd-cli config users set ${CPD_ADMIN_USER} --username admin --apikey ${CPD_API_KEY}
!./cpd-cli config profiles set ${CPD_PROFILE_NAME} --user ${CPD_ADMIN_USER} --url ${CPD_PROFILE_URL}
!./cpd-cli user-mgmt version

Support functions

print("Select the next empty cell.\n")
if platform == "cpdaas" :
    print("From the code snipet '</>' tab on the right, use 'Read data', 'Select data from project'")
    print("'Data asset', and finally 'cpdalllibs.zip'. Load it as 'StreamingBody object' and click on 'Insert code to cell'")
    print("Make sure the inserted code references 'streaming_body_1' in a line like:")
    print("streaming_body_1 = cos_client.get_object(Bucket=bucket, Key=object_key)['Body']")
else :
    print("From the data tab on the right, use 'Insert to code', 'load IO object' for the file 'cpdalllibs.zip'")
    print("Make sure the inserted code references 'raw_data_1' in a line like:")
    print("raw_data_1 = wslib.load_data('cpdalllibs.zip')")
print("\nExecute the cell")

# Load the python support functions
!rm -rf cpdalllibs
if platform == "cpdaas" :
    myzip = zipfile.ZipFile(BytesIO(streaming_body_1.read()))
else :
    myzip = zipfile.ZipFile(BytesIO(raw_data_1.read()))
    
myzip.extractall('.')

sys.path.append(".")
from cpdalllibs.cpdlibfns import *
importcpd()

# Test if we have access
help(getServiceInstances)

Get an access token

Note that the token usually lasts for only one hour.

An access token is used to identify a user in API requests.

token = "invalid"
resp = getToken(username, password, cpd_url) # from cell-2
if resp.status_code > 202:  # if error
    print("getToken status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    resp_json = resp.json()
    token = resp_json['token']
if token != "invalid" :
    print("Got a token at {} GMT".format(datetime.now().time().isoformat("seconds")))

# Header to use in subsequent queries
headersAPI = {
    'accept': 'application/json',
    'Content-type': 'application/json',
    'Authorization': 'Bearer ' + token,
    'cache-control': 'no-cache'
}

List service instances with cpd-cli

cpd-cli service-instance list --profile ${CPD_PROFILE_NAME} --output=text --verbose

The text output format is the default

Show the command options

!./cpd-cli service-instance list -h --profile user

Get the result in json format

It returns a IPython.utils.text.SList that has to be converted to a dictionary

instances_slist = !./cpd-cli service-instance list --profile user --output=json
instance_json = json.loads(" ".join([item for item in instances_slist]))
print("\n".join(["{:25} | {}".format(item['display_name'], item['provision_status']) for item in instance_json['service_instances']]))

Show verbose

Show what we get with --verbose

What is revealing is the one before last debug statement that returns something like:
API GET <cpd_url>/zen-data/v3/service_instances?offset=0&limit=50&fetch_all_instances=true

This provides the API call needed to get the same result

# Show what happens when the `--verbose`` flag is used
!./cpd-cli service-instance list --profile ${CPD_PROFILE_NAME} --output=text --verbose

List service instances using an API call

For this we use a function available in the support function library

# display the function source code
print(inspect.getsource(getServiceInstances))

resp = getServiceInstances(headersAPI, cpd_url)
if resp.status_code > 202 :
    print("getServiceInstances status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    instances_json = resp.json()
    print("Number of instances: {}".format(len(instances_json['service_instances'])))
# print(json.dumps(instances_json, indent=2, sort_keys=False))
    print("{:<25} | {:16} | {}".format("Display name", "id", "provision status"))
    print("{:<25} | {:16} | {}".format("=" * 25, "=" * 16, "=" * 16))
    print("\n".join(sorted((["{:<25} | {:16} | {}".format(item['display_name'], item['id'], item['provision_status']) 
                             for item in instances_json['service_instances']])) ))

List the members of the OpenScale service

There does not seem to be a cpd-cli command equivalent

openscale = [item for item in instances_json['service_instances'] if item['display_name'] == "openscale-defaultinstance"][0]
resp = getSvcUsers(headersAPI, cpd_url, openscale['id'])
if resp.status_code > 202 :
    print("getSvcsUsers status code: {}, reason: {}".format(resp.status_code, resp.reason))
resp_json = resp.json()
# print(json.dumps(resp_json, indent=2))
print("{:17} | {}".format("Username", "Role"))
print("{:17} | {}".format("=" * 17, "=" * 8))
print("\n".join("{:17} | {}".format(item['UserName'], item['Role']) for item in resp_json['requestObj']))

Create a target user

user_email = "[email protected]"

data = {
    "username": user_email,
    "authenticator": "default",
    "deletable": True,
    "displayName": user_email.split("@")[0],
    "email": user_email,
    "role": "User", # available values: ['Admin', 'User']
    "user_roles": [
        'User'
    ],
    "password": user_email.split("@")[0]
}

user = cre8User(headersAPI, cpd_url, data)
if resp.status_code > 202:  # if error
    print("cre8User Status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    print("User added")
    uid = user.json()['uid'] # needed to user to a group

Add the target user to OpenScale

The possible roles are: Admin, Editor, and Viewer

data = {
   "serviceInstanceID": openscale['id'],
   "users":[
      {
         "display_name": user_email.split("@")[0],
         "role": "Editor",
         "uid": uid,
         "username": user_email
      }
   ]
}
resp = addSvcUser(headersAPI, cpd_url, data)
if resp.status_code > 202:  # if error
    print("cre8User Status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    print("User added to the service")

List members again to prove the user was added

resp = getSvcUsers(headersAPI, cpd_url, openscale['id'])
if resp.status_code > 202 :
    print("getSvcsUsers status code: {}, reason: {}".format(resp.status_code, resp.reason))
resp_json = resp.json()
# print(json.dumps(resp_json, indent=2))
print("{:17} | {}".format("Username", "Role"))
print("{:17} | {}".format("=" * 17, "=" * 8))
print("\n".join("{:17} | {}".format(item['UserName'], item['Role']) for item in resp_json['requestObj']))

Delete a target user from a service

data = {
   "serviceInstanceID": openscale['id'],
   "users":[
      user.json()['uid']
   ]
}
resp = deleteSvcUser(headersAPI, cpd_url, data)
if resp.status_code > 202:  # if error
    print("deleteSvcUser Status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    print("User {} deleted from the service.".format(user_email))

# List the members again to prove it was removed
resp = getSvcUsers(headersAPI, cpd_url, openscale['id'])
if resp.status_code > 202 :
    print("getSvcsUsers status code: {}, reason: {}".format(resp.status_code, resp.reason))
resp_json = resp.json()
# print(json.dumps(resp_json, indent=2))
print("{:17} | {}".format("Username", "Role"))
print("{:17} | {}".format("=" * 10, "=" * 8))
print("\n".join("{:17} | {}".format(item['UserName'], item['Role']) for item in resp_json['requestObj']))

Cleanup: remove the target user

resp = deleteUser(headersAPI, cpd_url, user_email)
if resp.status_code > 202:  # if error
    print("deleteUser Status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    print("User {} deleted.".format(user_email))

Author

Jacques Roy is a member of the IBM Enablement for Data and AI

Copyright © 2023. This notebook and its source code are released under the terms of the MIT License.