Bench ICA Python Library

This tutorial demonstrates how to use the ICA Python library packaged with the JupyterLab image for Bench Workspaces.
See the JupyterLab documentation for details about the JupyterLab docker image provided by Illumina.
The tutorial will show how authentication to the ICA API works and how to search, upload, download and delete data from a project into a Bench Workspace. The python code snippets are written for compatibility with a Jupyter Notebook.
Python modules
This snipppet defines the required python modules for this tutorial
1
# Wrapper modules
2
import icav2
3
from icav2.api import project_data_api
4
from icav2.model.problem import Problem
5
from icav2.model.project_data import ProjectData
6
​
7
# Helper modules
8
import random
9
import os
10
import requests
11
import string
12
import hashlib
13
import getpass
Copied!
Authentication
This snippet shows how to authenticate using following methods:
ICA Username & Password
ICA API Token
1
# Authenticate using User credentials
2
username = input("ICA Username")
3
password = getpass.getpass("ICA Password")
4
tenant = input("ICA Tenant name")
5
url = os.environ['ICA_URL'] + '/rest/api/tokens'
6
r = requests.post(url, data={}, auth=(username,password),params={'tenant':tenant})
7
token = None
8
apiClient = None
9
if r.status_code == 200:
10
    token = r.content
11
    configuration = icav2.Configuration(
12
        host = os.environ['ICA_URL'] + '/rest',
13
        access_token = str(r.json()["token"])
14
        )
15
    apiClient = icav2.ApiClient(configuration, header_name="Content-Type",header_value="application/vnd.illumina.v3+json")
16
    print("Authenticated to %s" % str(os.environ['ICA_URL']))
17
else:
18
    print("Error authenticating to %s" % str(os.environ['ICA_URL']))
19
    print("Response: %s" % str(r.status_code))
20
​
21
## Authenticate using ICA API TOKEN
22
configuration = icav2.Configuration(
23
    host = os.environ['ICA_URL'] + '/rest'
24
)
25
configuration.api_key['ApiKeyAuth'] = getpass.getpass()
26
apiClient = icav2.ApiClient(configuration, header_name="Content-Type",header_value="application/vnd.illumina.v3+json")
Copied!
Data Operations
These snippets show how to manage data in a project. Operations shown are:
Create a Project Data API client instance
List all data in a project
Create a data element in a project
Upload a file to a data element in a project
Download a data element from a project
Search for matching data elements in a project
Delete matching data elements in a project
1
# Retrieve project ID from the Bench workspace environment
2
projectId = os.environ['ICA_PROJECT']
Copied!
1
# Create a Project Data API client instance
2
projectDataApiInstance = project_data_api.ProjectDataApi(apiClient)
Copied!
List Data
1
# List all data in a project
2
pageOffset = 0
3
pageSize = 30
4
try:
5
    projectDataPagedList = projectDataApiInstance.get_project_data_list(project_id = projectId, page_size = str(pageSize), page_offset = str(pageOffset))
6
    totalRecords = projectDataPagedList.total_item_count
7
    while pageOffset*pageSize < totalRecords:
8
        for projectData in projectDataPagedList.items:
9
            print("Path: "+projectData.data.details.path + " - Type: "+projectData.data.details.data_type)
10
        pageOffset = pageOffset + 1
11
except icav2.ApiException as e:
12
    print("Exception when calling ProjectDataAPIApi->get_project_data_list: %s\n" % e)
Copied!
Create Data
1
# Create data element in a project
2
data = icav2.model.create_data.CreateData(name="test.txt",data_type = "FILE")
3
​
4
try:
5
    projectData = projectDataApiInstance.create_data_in_project(projectId, create_data=data)
6
    fileId = projectData.data.id
7
except icav2.ApiException as e:
8
    print("Exception when calling ProjectDataAPIApi->create_data_in_project: %s\n" % e)
Copied!
Upload Data
1
## Upload a local file to a data element in a project
2
# Create a local file in a Bench workspace
3
filename = '/tmp/'+''.join(random.choice(string.ascii_lowercase) for i in range(10))+".txt"
4
content = ''.join(random.choice(string.ascii_lowercase) for i in range(100))
5
f = open(filename, "a")
6
f.write(content)
7
f.close()
8
​
9
# Calculate MD5 hash (optional)
10
localFileHash = md5Hash = hashlib.md5((open(filename, 'rb').read())).hexdigest()
11
​
12
try:
13
    # Get Upload URL
14
    upload = projectDataApiInstance.create_upload_url_for_data(project_id = projectId, data_id = fileId)
15
    # Upload dummy file
16
    files = {'file': open(filename, 'r')}
17
    data = open(filename, 'r').read()
18
    r = requests.put(upload.url , data=data)
19
except icav2.ApiException as e:
20
    print("Exception when calling ProjectDataAPIApi->create_upload_url_for_data: %s\n" % e)
21
​
22
# Delete local dummy file
23
os.remove(filename)
Copied!
Download Data
1
## Download a data element from a project
2
try:
3
    # Get Download URL 
4
    download = projectDataApiInstance.create_download_url_for_data(project_id=projectId, data_id=fileId)
5
​
6
    # Download file
7
    filename = '/tmp/'+''.join(random.choice(string.ascii_lowercase) for i in range(10))+".txt"
8
    r = requests.get(download.url)
9
    open(filename, 'wb').write(r.content)
10
​
11
    # Verify md5 hash
12
    remoteFileHash = hashlib.md5((open(filename, 'rb').read())).hexdigest()
13
    if localFileHash != remoteFileHash:
14
        print("Error: MD5 mismatch")
15
​
16
    # Delete local dummy file
17
    os.remove(filename)
18
except icav2.ApiException as e:
19
    print("Exception when calling ProjectDataAPIApi->create_download_url_for_data: %s\n" % e)
Copied!
Search for Data
1
# Search for matching data elements in a project
2
try:
3
    projectDataPagedList = projectDataApiInstance.get_project_data_list(project_id = projectId, full_text="test.txt")
4
    for projectData in projectDataPagedList.items:
5
        print("Path: " + projectData.data.details.path + " - Name: "+projectData.data.id + " - Type: "+projectData.data.details.data_type)
6
except icav2.ApiException as e:
7
    print("Exception when calling ProjectDataAPIApi->get_project_data_list: %s\n" % e)
Copied!
Delete Data
1
# Delete matching data elements in a project
2
try:
3
    projectDataPagedList = projectDataApiInstance.get_project_data_list(project_id = projectId, full_text="test.txt")
4
    for projectData in projectDataPagedList.items:
5
        print("Deleting file "+projectData.data.details.path)  
6
        projectDataApiInstance.delete_data(project_id = projectId, data_id = projectData.data.id)
7
except icav2.ApiException as e:
8
    print("Exception %s\n" % e)
Copied!
Base Operations
These snippets show how to get a connection to a base database and run an example query. Operations shown are:
Create a python jdbc connection
Create a table
Insert data into a table
Query the table
Delete the table
Snowflake Python API documentation can be found here​
This snipppet defines the required python modules for this tutorial
1
# API modules
2
import icav2
3
from icav2.api import project_base_api
4
from icav2.model.problem import Problem
5
from icav2.model.base_connection import BaseConnection
6
​
7
# Helper modules
8
import os
9
import requests
10
import getpass
11
import snowflake.connector
Copied!
1
# Retrieve project ID from the Bench workspace environment
2
projectId = os.environ['ICA_PROJECT']
Copied!
1
# Create a Project Base API client instance
2
projectBaseApiInstance = project_base_api.ProjectBaseApi(apiClient)
Copied!
Get Base Access Credentials
1
# Get a Base Access Token
2
try:
3
    baseConnection = projectBaseApiInstance.create_base_connection_details(project_id = projectId)
4
except icav2.ApiException as e:
5
    print("Exception when calling ProjectBaseAPIApi->create_base_connection_details: %s\n" % e)
6
## Create a python jdbc connection
7
ctx = snowflake.connector.connect(
8
    account=os.environ["ICA_SNOWFLAKE_ACCOUNT"],
9
    authenticator=baseConnection.authenticator,
10
    token=baseConnection.access_token, 
11
    database=os.environ["ICA_SNOWFLAKE_DATABASE"],
12
    role=baseConnection.role_name,
13
    warehouse=baseConnection.warehouse_name
14
)
15
ctx.cursor().execute("USE "+os.environ["ICA_SNOWFLAKE_DATABASE"])
Copied!
Create a Table
1
## Create a Table
2
tableName = "test_table"
3
ctx.cursor().execute("CREATE OR REPLACE TABLE " + tableName + "(col1 integer, col2 string)")
Copied!
Add Table Record
1
## Insert data into a table
2
ctx.cursor().execute(
3
        "INSERT INTO " + tableName + "(col1, col2) VALUES " + 
4
        "    (123, 'test string1'), " + 
5
        "    (456, 'test string2')")
Copied!
Query Table
1
## Query the table
2
cur = ctx.cursor()
3
try:
4
    cur.execute("SELECT * FROM "+tableName)
5
    for (col1, col2) in cur:
6
        print('{0}, {1}'.format(col1, col2))
7
finally:
8
    cur.close()
Copied!
Delete Table
1
# Delete the table
2
ctx.cursor().execute("DROP TABLE " + tableName);
Copied!