Jupyter Notebooks Samples¶

import os
import requests
import json
dlpath = '/datalake/v3/generateAccessToken'
gw = os.environ['GATEWAY_ENDPOINT'] + 'gateway/'
# increment_value = 1
headers = {
'Content-Type': 'application/json'
}   
payload="{ \"subtenantId\":\"\" } "
dl_url = gw + dlpath
response = requests.post(dl_url, data=payload, headers=headers)
#print(response.status_code)
dl = json.loads(response.text)
os.environ["AWS_ACCESS_KEY_ID"] = dl['credentials']['accessKeyId']
os.environ["AWS_SECRET_ACCESS_KEY"] = dl['credentials']['secretAccessKey']
os.environ["AWS_SESSION_TOKEN"] = dl['credentials']['sessionToken']    

# 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'
}
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.
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
response = requests.post(url, headers=HEADERS, data=body)
try:
    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')   
!echo "This is a test!" >> test.txt
fileName='test.txt'
signedURL = getSignedURL(fileName, OUTPUT_FOLDER)
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 ./mylocalfile.txt s3://$outputFolder+'/myfile.txt'

!pip install azure_cli
%store -r inputFolder
%store -r outputFolder
%store -r realInputFolder
%store -r realOutputFolder

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

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

!aws s3 cp $inputPath . --recursive --endpoint-url $endpoint

!aws s3 cp ./mdp-logs.csv $outputPath --recursive --endpoint-url $endpoint