Voorbeeldstappen: exporteren naar Microsoft Fabric met behulp van planningsgegevensbron en aangepast cloudlaadprogramma

const azureCloudUtil = {
   
   requestNewToken: function (scope = EnvSettings.default_scope) {
       var tenant_id = EnvSettings.tenant_id;
       var client_id = EnvSettings.client_id;
       var client_secret = EnvSettings.client_secret;
       
       var method = 'POST';
       var body = 'grant_type=client_credentials&client_id='+ client_id +'&scope=' + scope;

       var headers = {
           "Content-Type": 'application/x-www-form-urlencoded',
           "Authorization": 'Basic ' + ai.util.encodeBase64(client_id + ':' + client_secret, "")
       };

       var response = null;
       try {
           var url = EnvSettings.token_url.replace('{tenant_id}', tenant_id);
           response = makeHttpsReq(method, url, body, headers, null, false);
          // ai.log.logInfo('requestNewToken HTTP response ', JSON.stringify(response.getBody()));
       }
       catch (exception) {
           ai.log.logError('requestNewToken: HTTPS Request failed', '' + exception);
           return null;
       }

       if (response.getHttpCode() == '200') {
           var responseBody = response.getBody();
           return JSON.parse(responseBody);
       } else {
           ai.log.logError('requestNewToken: responseBody', responseBody);
           ai.log.logError('requestNewToken: HttpCode: ' , response.getHttpCode());
           return null;
       }
   },
   getWorkspace: function (workspaceIdentifier) {
       var response = azureCloudUtil.getWorkspaces();
       if (!isSuccessStatusCode(response)) {
           throw new Error("Unable to fetch workspace information!");
       }

       var responseBody = response.getBody();
       var workspaceList = JSON.parse(responseBody);
       if (!workspaceList || !workspaceList.value || !Array.isArray(workspaceList.value)) {
           throw new Error("Invalid input JSON. Expected a structure with a 'value' array.");
       }

       const lowerWorkspaceIdentifier = workspaceIdentifier.toLowerCase();
       const foundWorkspace = (
           workspaceList.value.find(entry =>
               entry.id.toLowerCase() === lowerWorkspaceIdentifier ||
               entry.displayName.toLowerCase() === lowerWorkspaceIdentifier
           ) || null
       );

       if (foundWorkspace === null) {
           throw new Error(`The resource '${workspaceIdentifier}' could not be found or you may not have the required permissions to access it.`);
       }

       return foundWorkspace;
   },
   getWorkspaces: function () {
       var targetUrl = EnvSettings.ms_fabric_base_url + '/workspaces'
       var method = 'GET';
       
       return makeHttpsReq(method, targetUrl, '', null);
   },
   getWorkspaceItems: function (workspace_id) {
       var targetUrl = EnvSettings.ms_fabric_base_url + '/workspaces/' + workspace_id + '/items'
       var method = 'GET';

       return makeHttpsReq(method, targetUrl, '', null);
   },
   getLakehouse: function (workspace, lakehouse_identifier) {
       var response = azureCloudUtil.getWorkspaceItems(workspace.id);
       if (!isSuccessStatusCode(response)) {
           throw new Error("Unable to fetch workspace items!");
       }
       
       var responseBody = response.getBody();
       var workspaceItems = JSON.parse(responseBody);

       if (!workspaceItems || !workspaceItems.value || !Array.isArray(workspaceItems.value)) {
           throw new Error("Invalid input JSON. Expected a structure with a 'value' array.");
       }

       const lower_lakehouse_identifier = lakehouse_identifier.toLowerCase();

       var foundItem = (
           workspaceItems.value.find(entry =>
               entry.type === "Lakehouse" &&
               ( entry.id.toLowerCase() === lower_lakehouse_identifier ||
               entry.displayName.toLowerCase() === lower_lakehouse_identifier )
           )  || null);
       
       if (foundItem === null) {
           foundItem = (
           workspaceItems.value.find(entry =>
               entry.id.toLowerCase() === lower_lakehouse_identifier ||
               entry.displayName.toLowerCase() === lower_lakehouse_identifier
           )  || null);
       }
       
       if (foundItem === null) {
           throw new Error(`The resource '${lower_lakehouse_identifier}' could not be found or you may not have the required permissions to access it.`);
       }

       if (foundItem.type !== "Lakehouse") {
           throw new Error(lakehouse_identifier + ' is of type ' + foundItem.type + ', not Lakehouse.');
       }
       
       return foundItem;
   },
   uploadRowsIntoFabricTable: function(workspaceId, lakehouseId, tableName, inputCsvFile) {
       const targetUrl = `${EnvSettings.ms_fabric_base_url}/workspaces/${workspaceId}/lakehouses/${lakehouseId}/tables/${tableName}/load`;
       
       const headers = {
           "Content-Type": "application/json",
       };
       
       const body = {
           relativePath: `Files/${inputCsvFile}`,
           pathType: "File",
           mode: "overwrite",  // Mode can be "overwrite", "append", "update", or "upsert"
           formatOptions:{
                   header: true,
                   delimiter: ",",
                   format: "CSV"
           }
       };

       const triggerResponse = makeHttpsReq("POST", targetUrl, JSON.stringify(body), headers);

       if (isSuccessStatusCode(triggerResponse)) {
           ai.log.logInfo('uploadRowsIntoFabricTable: ' + tableName, `Copy from the CSV file 'Files/${inputCsvFile}' initiated.`);
       } else {
           ai.log.logError('uploadRowsIntoFabricTable: ' + tableName, 'An error occurred while starting the copy process.' + triggerResponse.getBody());
           return false;
       }
       
       const terminalStates = ["Succeeded", "Failed"];
       const recentJobStatusResponse = trackJobProgress(triggerResponse, terminalStates, 'Copy CSV file');
       if (isSuccessStatusCode(triggerResponse)) {
           ai.log.logInfo('uploadRowsIntoFabricTable: ' + tableName, `Successfully copied all rows from the CSV file 'Files/${inputCsvFile}'.`);
       } else {
           ai.log.logError('uploadRowsIntoFabricTable: ' + tableName, 'Failed to copy rows from the CSV file ' + inputCsvFile + '. ' + triggerResponse.getBody());
           return false;
       }
       
       return true;
   },
   createFileResource: function(workspace, lakehouse, inputCsvFile) {
       const encodedWorkspaceName = workspace.displayName.replace(/ /g, "%20");
       const targetUrl = `${EnvSettings.onelake_dfs_base_url}/${encodedWorkspaceName}/${lakehouse.displayName}.lakehouse/Files/${inputCsvFile}?resource=file`;

       const headers = {
           "Content-Type": "application/json",
       };
       

       //ai.log.logInfo('createFileResource: targetUrl' , targetUrl);
       
       const response = makeHttpsReq("PUT", targetUrl, '', headers, EnvSettings.azure_storage_scope);

       if (!isSuccessStatusCode(response)) {
           ai.log.logError('createFileResource: ' + inputCsvFile, 'Failed to create the file resource.' + response.getBody());
           return false;
       }

       ai.log.logInfo('createFileResource: ' + inputCsvFile, `Created a file resource.`);
       
       return true;
   },
   uploadFileContent: function (workspace, lakehouse, inputCsvFile, inputCsvFileBody, position) {
       const encodedWorkspaceName = workspace.displayName.replace(/ /g, "%20");
       const targetUrl = `${EnvSettings.onelake_dfs_base_url}/${encodedWorkspaceName}/${lakehouse.displayName}.lakehouse/Files/${inputCsvFile}?action=append&position=${position}`;
       const contentLength = getByteLength(inputCsvFileBody);
       const headers = {
           "x-ms-version": "2021-06-08",
           "Content-Type": "text/csv",
           "Content-Length": contentLength
       };
       
       const response = makeHttpsReq("PATCH", targetUrl, inputCsvFileBody, headers, EnvSettings.azure_storage_scope);

       if (!isSuccessStatusCode(response)) {
           ai.log.logError('uploadFileContent: ' + inputCsvFile, 'Failed to upload the file content.'+ response.getBody());
           return -1;
       }

       ai.log.logInfo('uploadFileContent: ' + inputCsvFile, `Uploaded file content.`);
       
       return contentLength;
   },
   flushAndFinalizedFileContent: function(workspace, lakehouse, inputCsvFile, fileContentLength) {
       const encodedWorkspaceName = workspace.displayName.replace(/ /g, "%20");
       const targetUrl = `${EnvSettings.onelake_dfs_base_url}/${encodedWorkspaceName}/${lakehouse.displayName}.lakehouse/Files/${inputCsvFile}?action=flush&position=${fileContentLength}`;

       const headers = {
           "Content-Type": "application/json",
           "x-ms-version": "2021-06-08"
       };

       const response = makeHttpsReq("PATCH", targetUrl, '', headers, EnvSettings.azure_storage_scope);

       if (isSuccessStatusCode(response)) {
           ai.log.logInfo('flushAndFinalizedFileContent: ' + inputCsvFile, `The file content flushed and finalized.`);
           return true;
       } else {
           ai.log.logError('flushAndFinalizedFileContent: ' + inputCsvFile, 'Failed to flush and finalize the file content.');
           return false;
       }
   }

}

function makeHttpsReq(httpMethodType, targetUrl, httpBody, httpHeaders, scope = EnvSettings.default_scope, addAuthHeader = true) {
   
   var method = httpMethodType;
   var headers = httpHeaders || {
       'Content-Type': 'application/json'
   };

   if (EnvSettings.use_attached_oauth === false && addAuthHeader === true) {
       
       var tokenResponse = azureCloudUtil.requestNewToken(scope);
       if (tokenResponse === null || tokenResponse === undefined
           || tokenResponse['access_token'] === null || tokenResponse['access_token'] === undefined ) {
           throw new Error('Unable to generate access token. Check for invalid credentials, incorrect scopes, or misconfigured endpoints.');
       }
       
       EnvSettings.accessTokenResponse = tokenResponse;
       headers = {
           'Content-Type': 'application/json',
           'Authorization': 'Bearer '+ EnvSettings.accessTokenResponse['access_token'],
       };
   }

   var response = null;
   try {
       // ai.log.logInfo('targetUrl with method', targetUrl + '  ' + method);
       // ai.log.logInfo('httpBody', JSON.stringify(httpBody));
       // ai.log.logInfo('headers', JSON.stringify(headers));

       if (EnvSettings.use_attached_oauth === true) {
          response = ai.https.authorizedRequest(targetUrl, method, '' + httpBody, headers);  
       } else {
          response = ai.https.request(targetUrl, method, '' + httpBody, headers);    
       }
       // ai.log.logInfo('Response Body : ', response.getBody());

       if (!isSuccessStatusCode(response)) {
           ai.log.logError('makeHttpsReq: HTTP response code', response.getHttpCode());
       }
   } catch (exception) {
       ai.log.logError('makeHttpsReq: HTTPS Request failed', '' + exception);
       throw new Error('An unexpected communication failure occurred. ' + exception);
   }

   return response;

}

function isSuccessStatusCode(response) {
   
   if (response === null || response === undefined) {
       return false;
   }
   
   if (isNaN(response.getHttpCode())) {
       return false;
   }
   
   var responseCode = parseInt(response.getHttpCode());
   
   return responseCode >= 200 && responseCode < 300;
   
}

function sleepX(milliSeconds) {
  var startTime = new Date().getTime();
  while (new Date().getTime() < startTime + milliSeconds);
}

function trackJobProgress(triggerResponse, terminalStates, jobName) {

       var httpUrlLocation = triggerResponse.getHeaders()['Location'];
       var method = 'GET';

       const headers = {
           "Content-Type": "application/json",
       };

       var maxCheckLimit = EnvSettings.max_check_status_limit;
       var jobStatusResponse = null;
       var jsonJobStatus = null;
       var lcaseLatestJobStatus = null;
       do {
           if (lcaseLatestJobStatus !== null) {
               ai.log.logInfo(jobName, lcaseLatestJobStatus);
               sleepX(1000 * 5);
           }
           
           jobStatusResponse = makeHttpsReq(method, httpUrlLocation, '', headers);
           maxCheckLimit--;
           
           jsonJobStatus = {};
           lcaseLatestJobStatus = 'unknown';
           if (jobStatusResponse !== null && isSuccessStatusCode(jobStatusResponse)) {
               jsonJobStatus = JSON.parse(jobStatusResponse.getBody());
               lcaseLatestJobStatus = jsonJobStatus.status.toLowerCase();
           }
       } while(maxCheckLimit > 0 && jsonJobStatus !== null && !(terminalStates.some(word => word.toLowerCase() === lcaseLatestJobStatus)));
       
       if (maxCheckLimit > 0){
           ai.log.logInfo(jobName, 'The job has been completed.');
       } else {
           ai.log.logInfo(jobName, `Terminated status check as max limit ${maxCheckLimit} reached.`);
       }
       ai.log.logInfo(jobName + ' - final status', jsonJobStatus.status);
       if (jsonJobStatus.Error !== null && jsonJobStatus.Error !== undefined && jsonJobStatus.Error.trim().length() > 0) {
           ai.log.logError(jobName + ' - error', JSON.stringify(jsonJobStatus.Error));
       }
       
       return jsonJobStatus;
}

function getByteLength(str) {
   let byteLength = 0;
   
   if (str !== null && str !== undefined) {
   
       for (let i = 0; i < str.length; i++) {
           const charCode = str.charCodeAt(i);
   
           if (charCode <= 0x7f) {
               byteLength += 1;
           } else if (charCode <= 0x7ff) {
               byteLength += 2;
           } else if (charCode <= 0xffff) {
               byteLength += 3;
           } else {
               byteLength += 4;
           }
       }        
   }

   return byteLength;
}

 

var EnvSettings = {
   ms_fabric_base_url:'https://api.fabric.microsoft.com/v1',
   onelake_dfs_base_url:'https://onelake.dfs.fabric.microsoft.com',
   ms_fabric_workspace_url:'https://api.fabric.microsoft.com/v1/workspaces',
   default_scope: 'https://api.fabric.microsoft.com/.default',
   items_scope: 'https://api.fabric.microsoft.com/Workspace.ReadWrite.All https://api.fabric.microsoft.com/Item.ReadWrite.All',
   onelake_scope: 'https://onelake.dfs.fabric.microsoft.com/.default',
   azure_storage_scope: 'https://storage.azure.com/.default',
   batchSize: 1000,
   target_workspace_identifier:null,
   target_lakehouse_identifier: null,
   max_check_status_limit:25,
   use_attached_oauth: false,
   token_url: 'https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token',
   tenant_id:null,
   client_id:null,
   client_secret:null,
   accessTokenResponse: null,
};

 

'use strict';
/*******************************************************************************
* MainFunction
*******************************************************************************/
import { ai } from '/Script/Source/Integration2/CustomCloudScripts/CustomCloudScriptApi.js';


// *****************************************************************************
// Test Connection
// *****************************************************************************
function testConnection(context) {
   var dataSource = context.getDataSource();
   initSettings(dataSource);

   var testConnectionResult = false;
   var workspaceList  = azureCloudUtil.getWorkspaces();
   
   
   if (workspaceList === null || workspaceList === undefined) {
      ai.log.logInfo('Workspace List', 'Empty or undefined');
   } else {
       var workspaceItems = JSON.parse(workspaceList.getBody());

       if (!workspaceItems || !workspaceItems.value || !Array.isArray(workspaceItems.value)) {
           ai.log.logError('Invalid response', "Invalid response JSON. Expected a structure with a 'value' array.");
       } else {
           testConnectionResult = true;
       }
   
   }
   
   return testConnectionResult;
}

function exportData(context) {
   var reader = context.createTableReader();
   var dataSource = context.getDataSource();
   initSettings(dataSource);

   var sourceTableName = 'PersonnelModelSheet';
   var csvFileName = `${sourceTableName}.csv`;


   var workspace = azureCloudUtil.getWorkspace(EnvSettings.target_workspace_identifier);
   // ai.log.logInfo('workspace', workspace.id);
   
   var lakehouse = azureCloudUtil.getLakehouse(workspace, EnvSettings.target_lakehouse_identifier);
   // ai.log.logInfo('lakehouse', lakehouse);
   
   var totalLineCount = generateAndPushCSVData(reader, workspace, lakehouse, sourceTableName, csvFileName);
   
   ai.log.logInfo('exportData', 'Pushed ' + totalLineCount + ' lines');

}

function generateAndPushCSVData(reader, workspace, lakehouse, sourceTableName, csvFileName) {
   const tagetBatchSize = EnvSettings.batchSize;
   var headers = null;
   var totalColumnCount = null;
   var csvColumnCount = null;


   let row;
   var csvRows = [];
   var lineCount = 0;
   var totalByteSize = 0;
   var recentCallResult = false;
   var recentUploadByteSize = 0;
   while ((row = reader.readRow()) !== null && row !== undefined) {
       
       if (totalColumnCount === null) {
           
           totalColumnCount = row.length;
           csvColumnCount = row.length - 2;
           headers = createHeaderRow(csvColumnCount);

           if (!Array.isArray(headers) || headers.length === 0) {
               throw new Error("Headers must be a non-empty array.");
           }

           // Add headers as the first row of the CSV
           csvRows = [headers.join(",")];
           
           var result = azureCloudUtil.createFileResource(workspace, lakehouse, csvFileName);
           if (result === false) {
               return 0;
           }
       }

       if (!Array.isArray(row) || row.length < headers.length) {
           throw new Error("Each row must be an array with the same number of columns as the headers.");
       }

       // Escape values and join them with commas
       const escapedRow = row.map(value =>
           typeof value === "string" ? `"${value.replace(/"/g, '""')}"` : value
       );
       csvRows.push(escapedRow.join(","));
       lineCount++;

       // reached batch limit
       if (lineCount >= tagetBatchSize && (lineCount % tagetBatchSize) === 0){
           var csvData = prepareCsvData(csvRows, totalByteSize);
           recentUploadByteSize = azureCloudUtil.uploadFileContent(workspace, lakehouse, csvFileName, csvData, totalByteSize);
           if (recentUploadByteSize < 1) {
               throw new Error("An error occurred while uploading rows.");
           }
           totalByteSize += recentUploadByteSize;
           csvRows = [];
       }
   }
   

   if (csvRows.length > 0 ){
       var csvData = prepareCsvData(csvRows, totalByteSize);
       recentUploadByteSize = azureCloudUtil.uploadFileContent(workspace, lakehouse, csvFileName, csvData, totalByteSize);
       if (recentUploadByteSize < 1) {
           throw new Error("An error occurred while uploading rows.");
       }
       totalByteSize += recentUploadByteSize;

   }
   
   if (lineCount > 0) {
       recentCallResult = azureCloudUtil.flushAndFinalizedFileContent(workspace, lakehouse, csvFileName, totalByteSize);
       
       if (recentCallResult === true) {
           recentCallResult = azureCloudUtil.uploadRowsIntoFabricTable(workspace.id, lakehouse.id, sourceTableName, csvFileName);
       } else {
           ai.log.logError('generateAndPushCSVData - table load status ', 'Failed to flush and finalize the uploaded rows. Check the system and try again.');
       }
       
       if (recentCallResult === false) {
           return 0;
       }
   }
   
   return lineCount;
}

function prepareCsvData(csvRows, totalByteSize){
   if (csvRows === null || csvRows === undefined || csvRows.length === 0 ) {
       return "";
   }
   
   var csvData = null;
   if (totalByteSize === 0) {
       csvData = csvRows.join("\n");
   } else {
       csvData = ("\n" + csvRows.join("\n"));
   }
   
   return csvData;
}


function createHeaderRow(columnCount) {
   if (columnCount <= 0 || !Number.isInteger(columnCount)) {
       throw new Error("Column count must be a positive integer.");
   }

   const headerRow = Array.from({ length: columnCount }, (_, index) => `Column${index + 1}`);
   return headerRow;
}


function initSettings( dataSource ) {
   EnvSettings.target_workspace_identifier = getSetting(dataSource, "Target Workspace");
   EnvSettings.target_lakehouse_identifier = getSetting(dataSource, "Target Lakehouse");
   EnvSettings.tenant_id = getSetting(dataSource, "Tenant Id");
   EnvSettings.client_id = getSetting(dataSource, "Client Id");
   EnvSettings.client_secret = getSetting(dataSource, "Client Secret");
}


function getSetting(dataSource, propKey, defaultValue = null) {
   var setting = dataSource.getSetting(propKey)
   if (setting === null || setting === undefined) {
       return     defaultValue;
   }
   
   var value = dataSource.getSetting(propKey).getValue();
   
   if (value === null || value === undefined) {
       return     defaultValue;
   }
   
   return value.trim();
}