Search…
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!