Jupyter Notebook Code Examples¶

import os
import requests
import json

# The IDL endpoint
dlpath = '/datalake/v3/generateAccessToken'

# The base gateway that provides transparent authentication
gw = os.environ['GATEWAY_ENDPOINT'] + 'gateway/'

headers = {
'Content-Type': 'application/json'
}   
payload="{ \"subtenantId\":\"\" } "
dl_url = gw + dlpath

# Perform the request
response = requests.post(dl_url, data=payload, headers=headers)

# The response should be a JSON object
idlSession = json.loads(response.text)

# Update the environments with the temporary credentials
os.environ["AWS_ACCESS_KEY_ID"] = idlSession['credentials']['accessKeyId']
os.environ["AWS_SECRET_ACCESS_KEY"] = idlSession['credentials']['secretAccessKey']
os.environ["AWS_SESSION_TOKEN"] = idlSession['credentials']['sessionToken']

# Now operations against the IDL S3 bucket can be performed directly
# ...

# upgrade pip and install required libraries
%pip install --upgrade pip 
%pip install requests --force-reinstall --upgrade 
%pip install pandas --force-reinstall --upgrade 
%pip install pyarrow --force-reinstall --upgrade 

import datetime
import requests
import os
import json
import re

HEADERS = {
'Accept': '*/*',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
'Content-Type': 'application/json'
}

# The base gateway that provides transparent authentication
GATEWAY = os.environ['GATEWAY_ENDPOINT'] + '/gateway/'

OUTPUT_FOLDER = 'OUTPUT_FOLDER'

# Get a signed URL for down/upload of data. The function
# attempts for 5 times to obtain the URL and then raises an exception.
# The same function can also generate an upload signed url
def getSignedURL(fileName, folder, attempt=0, upload=True): 
  if upload:
    IDLpath = 'datalake/v3/generateUploadObjectUrls'
  else:
    IDLpath = 'datalake/v3/generateDownloadObjectUrls'

  IDLFilePath = '/%s/%s' % (folder, fileName)
  url = GATEWAY + IDLpath
  body='{"paths": [{"path": "%s"}]}' % IDLFilePath

  # Send request to receive a signed url
  response = requests.post(url, headers=HEADERS, data=body)

  try:
    # Return the signed url
    return json.loads(response.text)['objectUrls'][0]['signedUrl']
  except KeyError:
    if attempt < 5:
        attempt += 1
        return getSignedURL(fileName, attempt, upload)
    else:
        raise Exception('Failed to get a signed URL')

# Create a text file using bash
!echo "This is a test!" >> test.txt


fileName='test.txt'
signedURL = getSignedURL(fileName, OUTPUT_FOLDER)

# upload the test file using the signed url
requests.put(signedURL, headers=HEADERS, data=fileName)

key_id = os.environ["CLIENT_ID"]
key_secret = os.environ["CLIENT_SECRET"]
tenant = os.environ["tenant"]
subTenant = os.environ.get("subtenant")
authServer = os.environ["TOKEN_ENDPOINT"]
coreGateway= os.environ["CORE_GATEWAY"]

class JsonWebToken():
    # Handles getting JWT from environment controller using HMAC.
    # Refreshes token whenever it gets expired.

    def __init__(self):
        self.token = self.__refreshToken()

    def __refreshToken(self):
        #Gets new JWT token using HMAC
        sign, timestamp = self.__getHmacSignature()
        headers = {
            'Authorization': 'HMAC-SHA256 Credential=' + key_id + '&SignedHeaders=host;x-msg-timestamp&Signature=' + sign,
            'x-msg-timestamp': timestamp
        }
        response = requests.get(authServer, headers=headers, verify=False)
        print(response.status_code)
        return json.loads(response.text)["access_token"]

    def __getHmacSignature(self):
        # Creates HMAC Signature for getting JWT
        payload, timestamp = self.__getPayload()
        sign = hmac.new(key_secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
        return sign, timestamp

    def __isExpired(self):
        # Checks whether existing JWT is expired or not.
        decoded = jwt.decode(self.token, verify=False, options={'verify_signature': False})
        current = int(datetime.now().timestamp()) + 10
        if decoded['exp'] <= current:
            return True
        return False

    def getToken(self):
        # Gets token if expired. This method should be used by users of this class.
        if self.__isExpired():
        print("token is expired. getting new one...")
        self.__refreshToken()
        return self.token

    def __getPayload(self):
        timestamp = datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')
        payload = "GET\n" + urlparse(authServer).netloc + "\n" + timestamp + "\n" + tenant
        print(payload)
        if subTenant != None:
            payload = payload + "\n" + subTenant

        return payload, timestamp

    jwtobj = JsonWebToken()

    # Generate Upload URL

    # create random string
    random_string = ''.join(random.choices(string.ascii_lowercase, k = 10))
    print(random_string)
    filename = 'readme.json'

    path = random_string + "/" + filename
    body = json.dumps({
    "paths": [
        {
            "path": path
        }
    ],
    "subtenantId": "",
    "isMicrosoftRoutingEnabled": False
    })

    with open(filename, 'w') as f:
        random_content = "{ 'random_string': '"+ ''.join(random.choices(string.ascii_lowercase, k = 50)) + "' }"
        f.write(random_content)

    upload_url = coreGateway + "/api/datalake/v3/generateUploadObjectUrls"

    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Authorization': 'Bearer ' + jwtobj.getToken()
    }

    response = requests.post(upload_url, data=body, headers=headers)

    print("response code: " + str(response.status_code))
    print(response.content)

    signedUploadUrl = json.loads(response.content)['objectUrls'][0]['signedUrl']
    print("Signed Upload URL: " + signedUploadUrl)

!aws s3 cp /tmp/mylocalfile.txt s3://$outputFolder+'/myfile.txt'

!pip install azure_cli
# reads the inputFolder and ouputFolder variables, preparing them for use 
%store -r inputFolder
%store -r outputFolder

input = inputFolder 
output = outputFolder
!az storage blob download --account-name prlstorageanls -c jobmanager -f data.csv -n $input'/data.csv' --auth-mode login
!az storage blob upload --account-name prlstorageanls -c jobmanager -f data.csv  -n $output'/output.csv' --auth-mode login 

inputPath='s3://' + os.environ.get('inputFolder') + '/'
outputPath='s3://' + os.environ.get('outputFolder') + '/'   
endpoint=os.environ.get('AWS_ENDPOINT_URL') + '/'

# Create the directory
directory_path = '/tmp/upload_dir'
os.makedirs(directory_path, exist_ok=True)
os.chmod(directory_path, stat.S_IRWXU)

!aws s3 cp $inputPath /tmp/upload_dir --recursive --endpoint-url $endpoint

!aws s3 cp /tmp/upload_dir $outputPath --recursive --endpoint-url $endpoint