CPD-02 User Management.ipynb

Kernel: Python 3.10

User management

Using the API documented at: IBM Cloud Pak for Data Platform API
This is the latest despite CPD 4.6 being out.

Capabilities demonstrated:

• Use of a zip file to add python functions to the environment

• List all users

• List available groups

• List available roles

• Get the CPD configuration information and display available permissions

• Create a role

• Create a group

• Create a user

• Add the user to a group

• Add the new role to the user (does not work!!!)

• Delete the user

• Delete the new group

• Delete the new role

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

import zipfile
from io import BytesIO

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
#cpd_url = "https://cpd-cpd.ai-governance-12345a678e90addd123c4567c8f9a012-3456.us-east.containers.appdomain.cloud/"
cpd_url = "https://cloud-pak-for-data/"

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

print("Select the next empty cell.\n")
if platform == "cpdaas" :
    print("From the data tab on the right, use 'Insert to code', 'StreamingBody object' for the file 'cpdalllibs.zip'")
    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")

Support functions

# 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(getUsers)
print("\nShow the source of a function:\n")
print(inspect.getsource(getUsers))

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 all users

This returns an array of users with a lot of information.
Only part of this information is displayed. See the documentation for details.

users = []
resp = getUsers(headersAPI, cpd_url)
if resp.status_code > 202:  # if error
    print("getUsers status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    users = resp.json()
# Assuming it worked...
myFormat = "{:18} {:10} {:24} {:28} {:8} {}"
print(myFormat.format("Display name", "uid", "username", "email", "role", "status"))
for user in users :
    print(myFormat.format(user['displayName'], user['uid'], user['username'], 
                          user['email'],user['role'], user['current_account_status']))
    #print("\tPermissions: {}".format(",".join([item for item in user['permissions'] ])))
    #print("\tUser_roles: {}".format(",".join([item for item in user['user_roles'] ])))
    print("\tGroups: {}".format(",".join([item['name'] for item in user['groups']])))

List available groups

resp = getGroups(headersAPI, cpd_url)
if resp.status_code > 202:  # if error
    print("getGroups status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    groups = resp.json()['results']
myFormat = "{:15} {:<10} {:13}"
print(myFormat.format("Name", "Group ID", "Members count"))
for group in groups :
    print(myFormat.format(group['name'], group['group_id'], group['members_count']))

List available roles

resp = getRoles(headersAPI, cpd_url)
if resp.status_code > 202:  # if error
    print("getRoles status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    roles = resp.json()['rows']
# Assuming it worked
role_format = "{:25} {:<11} {}"
print(role_format.format("Role name", "Users count", "User groups count"))
for role in roles :
    print(role_format.format(role['doc']['role_name'], role['doc']['users_count'], role['user_groups_count']))

# Print the permissions for the Administrator role
admin = [item for item in roles if item['doc']['role_name'] == "Administrator"][0]
print("\nAdministrator's permissions:")
print(", ".join([item for item in admin['doc']['permissions']]))

Get the configuration information

List the available permissions in the CPD configuration

resp = getConfig(headersAPI, cpd_url)
if resp.status_code > 202:  # if error
    print("getConfig status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    config = resp.json()
# list the permissions
print("Permissions available in CPD:")
print(", ".join([item['key'] for item in config['permissions']]))

Create a role

Create a simple role using the same permissions as the existing User role.
Note that these permissions are a subset of the ones listed from the config permissions list.

usr = [item for item in roles if item['doc']['role_name'] == "User"][0]
permissions = [item for item in usr['doc']['permissions']]

print("Group 'User' permissions:")
print(", ".join([item for item in permissions]))

data = {
  "role_name": "Mytemprole",
  "description": "This is a temporary role.",
  "permissions": permissions
}

resp = cre8Role(headersAPI, cpd_url, data)
if resp.status_code > 202:  # if error
    print("getConfig status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    print("Role {} created".format(data['role_name']))
    role_id = resp.json()['id']

## Display the newly created role
resp = getRoles(headersAPI, cpd_url)
if resp.status_code > 202:  # if error
    print("getRoles status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    roles = resp.json()['rows']
new_role = [item for item in roles if item['doc']['role_name'] == data['role_name']][0]
print(json.dumps(new_role, indent=4))

Create a Group

data = {
  "name": "MyGroup",
  "description": "Temporary group",
  "account_id": 1000, # default
  "role_identifiers": [new_role['id']]
}
resp = cre8Group(headersAPI, cpd_url, data)
if resp.status_code > 202:  # if error
    print("cre8Group status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    new_group = resp.json()
    print("Group {} created".format(data['name']))

# Get the new group information
resp = getGroups(headersAPI, cpd_url)
if resp.status_code > 202:  # if error
    print("getGroups status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    groups = resp.json()['results']
new_group = [item for item in groups if item['name'] == "MyGroup"][0]
print(json.dumps(new_group, indent=4))

Create a user

Note that while we create the user, we also identify the role to use. In this case, "User".

It is possible to add multiple roles to a user at creation time. At the moment it seems to be limited to the roles that were part of the cluster installation. They are:

- Reporting Administrator
- Developer
- Data Scientist
- User
- Administrator
- Data Steward
- Data Quality Analyst
- Data engineer
- Business Analyst

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]
}

resp = 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 = resp.json()['uid'] # needed to user to a group

# List information on the new user
resp = getUser(headersAPI, cpd_url, uid)
if resp.status_code > 202:  # if error
    print("getUsers status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    new_user = resp.json()
print(json.dumps(new_user, indent=4))

Add the user to a group

resp = addUserToGroup(headersAPI, cpd_url, new_group['group_id'], uid)
if resp.status_code > 202:  # if error
    print("addUserToGroup Status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    print("Group added to the user")

# List the user
resp = getUser(headersAPI, cpd_url, uid)
if resp.status_code > 202:  # if error
    print("getUsers status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    new_user = resp.json()
print(json.dumps(new_user, indent=4))

Update the user and add a role

returns status code: 400, reason: Bad Request

data = {
    "user_roles": [
        "Data Steward"
    ]
}
print("Updating user with role {}".format(new_role['doc']['role_name']))
resp = addRoleToUser(headersAPI, cpd_url, user_email, data) 
if resp.status_code > 202:  # if error
    print("addRoleToUser Status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    print("User updated")

# List the user
resp = getUser(headersAPI, cpd_url, uid)
if resp.status_code > 202:  # if error
    print("getUsers status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    new_user = resp.json()
print(json.dumps(new_user, indent=4))

Delete the 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))

# display the users
users = []
resp = getUsers(headersAPI, cpd_url)
if resp.status_code > 202:  # if error
    print("getUsers status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    users = resp.json()
# Assuming it worked...
myFormat = "{:18} {:10} {:24} {:28} {:8} {}"
print(myFormat.format("Display name", "uid", "username", "email", "role", "status"))
for user in users :
    print(myFormat.format(user['displayName'], user['uid'], user['username'], 
                          user['email'],user['role'], user['current_account_status']))

Delete the previously created group

resp = deleteGroup(headersAPI, cpd_url, new_group['group_id'])
if resp.status_code > 202:  # if error
    print("deleteRole Status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    print("Group {} deleted".format(new_group['group_id']))

Delete the previously created role

resp = deleteRole(headersAPI, cpd_url, new_role['id'])
if resp.status_code > 202:  # if error
    print("deleteRole Status code: {}, reason: {}".format(resp.status_code, resp.reason))
else :
    print("Role {} deleted".format(new_role['doc']['role_name']))

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.