# _____ _ _____ __ # | __ \ (_) |_ _| / _| # | |__) |___ ___ _ _ __ ___ | | _ __ | |_ ___ # | _ // _ \/ __| | '_ \ / _ \ | | | '_ \| _/ _ \ # | | \ \ __/ (__| | |_) | __/ _| |_| | | | || (_) | # |_| \_\___|\___|_| .__/ \___| |_____|_| |_|_| \___/ # | | # |_| # Project Name - Peoplesoft to Salesforce: Test and Score Sync # Recipe URL - https://app.lingk.io/a/10932/tf/19134 # Description - 12 - Peoplesoft to Salesforce: Test and Score Sync # Select all test from PeopleSoft STDNT_TEST table. # Reject non-matching records due to Contact. # Collect info from all tables and transform fields. # Insert/Update record into SF by PeopleSoft_Test_Unique_Identifier__c. # Insert test requested fields with their SF id in Lingk environment file. # Reject non-matching records due to test for hed__Test_Score__c object. # Collect info from all tables and transform fields. # Insert/Update record into SF by PeopleSoft_Test_Score_Unique_Identifier__c. # Insert test-score requested fields with their SF id in Lingk environment file. # Systems - Peoplesoft, Salesforce # Connectors - HttpV2, Salesforce # Data Flows - Single Direction # Connection Type - API to API # Data Models ## Peoplesoft Models Used: #1. PS_STDNT_TEST_COMP ## Salesforce Objects Used: #1. Contact #2. hed__Test__c #3. hed__Test_Score__c # Add Recipe notes / Change log information here! # _____ _ # / ____| | | # | | ___ _ __ _ __ ___ ___| |_ ___ _ __ ___ # | | / _ \| '_ \| '_ \ / _ \/ __| __/ _ \| '__/ __| # | |___| (_) | | | | | | | __/ (__| || (_) | | \__ \ # \_____\___/|_| |_|_| |_|\___|\___|\__\___/|_| |___/ # # CONNECTORS specify what data will be pulled into the in-memory database during processing connectors: # Configure HTTP & Salesforce credentials in your Environment before running this recipe # HTTP Setup - https://help.lingk.io/en/articles/304-http-connector-setup-guide # Salesforce Setup - https://help.lingk.io/en/articles/267-salesforce-connector-setup-guide ###### Start - HTTP Reader connectors ###### ## PeopleSoft Person Connector ## HTTP V2 connector in comparison with V1, handles pagination through the connector properties ## Authentication and URL values are provided via environment variables # documentation - https://help.lingk.io/en/articles/202-custom-environment-variables ## PeopleSoft Test Connector - name: peopleSoftTestAndScoreData type: httpV2Reader properties: httpMethod: POST headerJson: { "Authorization":"Basic {{env.vars.apiAccessToken}}", "Content-Type":"application/json" } acceptInvalidCerts: false url: '{{env.vars.midStateInstanceURL}}' paginationType: requestBody body: | { "isDebugMode": false, "rowLimit": 1000, "pageNumber": 1, "records": [ {"recordName": "STDNT_TEST", "sqlWhereClause": " EXISTS (SELECT 1 FROM PS_L_SF_STDNT_FLTR B WHERE A.EMPLID = B.EMPLID ) AND EXISTS (SELECT 1 FROM PS_STDNT_TEST_COMP C WHERE C.EMPLID = A.EMPLID AND C.TEST_ID = A.TEST_ID AND DATE_LOADED >= %datein('{{env.vars.testLoadedDate}}') )", "includeDescriptionsFor": [ "TEST_ID" ] }, { "recordName": "STDNT_TEST_COMP", "parentRecordName":"STDNT_TEST", "includeDescriptionsFor": [ "TEST_COMPONENT", "LS_DATA_SOURCE" ] }, { "recordName": "SA_TEST_TBL", "parentRecordName": "STDNT_TEST" } ] } schema: fields: - name: 'EMPLID' type: 'string' - name: 'SA_TEST_TBL' type: 'object' objectSchema: name: objectSchemaSchema fields: - name: 'fields' type: 'array' items: name: 'arrayElement' type: 'object' objectSchema: name: objectSchemaSchema fields: - name: 'DESCR' type: 'string' - name: 'DESCRSHORT' type: 'string' - name: 'EFFDT' type: 'string' - name: 'TEST_ID' type: 'string' - name: 'STDNT_TEST_COMP' type: 'object' objectSchema: name: objectSchemaSchema fields: - name: 'fields' type: 'array' items: name: 'arrayElement' type: 'object' objectSchema: name: objectSchemaSchema fields: - name: 'DATE_LOADED' type: 'string' - name: 'EMPLID' type: 'string' - name: 'LS_DATA_SOURCE' type: 'string' - name: 'LS_DATA_SOURCE_description' type: 'string' - name: 'SCORE' type: 'double' - name: 'SCORE_LETTER' type: 'string' - name: 'TEST_COMPONENT' type: 'string' - name: 'TEST_COMPONENT_description' type: 'string' - name: 'TEST_DT' type: 'string' - name: 'TEST_ID' type: 'string' - name: 'TEST_INDEX' type: 'double' - name: 'objectName' type: 'string' - name: 'objectType' type: 'string' - name: 'TEST_ID' type: 'string' - name: 'rowNumber' type: 'long' ###### End - HTTP Reader connectors ###### ###### Start - Salesforce Reader connectors ###### # Salesforce Contact Reader - name: sfContactReader type: salesforceReader delayedRead: true parameterizedBy: batchedStudentId properties: useBearerToken: true query: SELECT Id, University_Student_ID__c, Name FROM Contact WHERE University_Student_ID__c IN ( {% for element in var.output %} '{{element.id}}'{% if not loop.last %},{% endif %} {% endfor %} ) schema: fields: - name: 'Id' type: 'string' - name: 'University_Student_ID__c' type: 'string' - name: 'Name' type: 'string' # Salesforce Test Reader - name: sfTestReader type: salesforceReader delayedRead: true parameterizedBy: batchedTestId properties: useBearerToken: true query: SELECT Id, Name, PeopleSoft_Test_Unique_Identifier__c FROM hed__Test__c WHERE PeopleSoft_Test_Unique_Identifier__c IN ( {% for element in var.output %} '{{element.id}}'{% if not loop.last %},{% endif %} {% endfor %} ) schema: fields: - name: 'Id' type: 'string' - name: 'Name' type: 'string' - name: 'PeopleSoft_Test_Unique_Identifier__c' type: 'string' # Salesforce Test Score Reader - name: sfTestScoreReader type: salesforceReader delayedRead: true parameterizedBy: batchedTestScoreId properties: useBearerToken: true query: SELECT Id, Name, PeopleSoft_Test_Score_Unique_Identifier__c FROM hed__Test_Score__c WHERE PeopleSoft_Test_Score_Unique_Identifier__c IN ( {% for element in var.output %} '{{element.id}}'{% if not loop.last %},{% endif %} {% endfor %} ) schema: fields: - name: 'Id' type: 'string' - name: 'Name' type: 'string' - name: 'PeopleSoft_Test_Score_Unique_Identifier__c' type: 'string' ###### End - Salesforce Reader connectors ###### ###### Start - Salesforce Writer connectors ###### # Salesforce Test Writer - name: sfTestWriter type: salesforceBulkWriter onError: continue isSerial: true properties: onError: continue useBearerToken: true sfObject: hed__Test__c batchSize: 50 # Salesforce Test Score Writer - name: sfTestScoreWriter type: salesforceBulkWriter onError: continue isSerial: true properties: onError: continue useBearerToken: true sfObject: hed__Test_Score__c batchSize: 50 ###### End - Salesforce Writer connectors ###### ###### Start - Local Writer connectors ###### # Local File Writer - name: finalMappedRecordFile type: localFileWriter format: csvWriter properties: fileName: "{{executionContext.logFile.name}}.csv" ###### End - Local Writer connectors ###### # _______ _ # |__ __| | | # | | __ _ ___| | _____ # | |/ _` / __| |/ / __| # | | (_| \__ \ <\__ \ # |_|\__,_|___/_|\_\___/ # Batch task used for parameterized by statements/connectors tasks: - name: batchTask type: dataOperation function: batch parameters: inputBatchGroupBy: # required id: desc inputBatchSize: 200 # inputBatchSize # required (to test) inputBatchFields: "id" # comma delimited for other fields #inputBatchFields #required batchedColumnAlias: output # outputColumnName - optional with a default of "output" # inputTable: # optional # result: #optional # ______ _ # | ____| | | # | |__ ___ _ __ _ __ ___ __ _| |_ ___ # | __/ _ \| '__| '_ ` _ \ / _` | __/ __| # | | | (_) | | | | | | | | (_| | |_\__ \ # |_| \___/|_| |_| |_| |_|\__,_|\__|___/ writeFormats: - name: csvWriter type: delimited properties: quoteAllFields: true delimiter: ',' header: true # _____ _ _ _ # / ____| | | | | | # | (___ | |_ __ _| |_ ___ _ __ ___ ___ _ __ | |_ ___ # \___ \| __/ _` | __/ _ \ '_ ` _ \ / _ \ '_ \| __/ __| # ____) | || (_| | || __/ | | | | | __/ | | | |_\__ \ # |_____/ \__\__,_|\__\___|_| |_| |_|\___|_| |_|\__|___/ # STATEMENTS specify how the data should be processed while in memory statements: #******************************************************************** D I S C L A I M E R *********************************************************************************************** # * # Note that in an effort to keep recipes optimized for DPH (Data Processing Hours), print statements should be commented out after development has concluded for a recipe. * # For more information on DPH optimization, please visit the following help article - https://help.lingk.io/en/articles/212-minimizing-data-processing-hours-on-the-lingk-platform * # * #******************************************************************** D I S C L A I M E R *********************************************************************************************** #- statement: print peopleSoftTestAndScoreData - statement: (peopleSoftTestAndScoreDataCount) => SELECT COUNT(*) FROM peopleSoftTestAndScoreData #- statement: PRINT peopleSoftTestAndScoreDataCount # inline function - flatten the array to make it a basic table - statement: (inlineTestComponent) => SELECT inline(STDNT_TEST_COMP.fields) FROM peopleSoftTestAndScoreData #- statement: PRINT inlineTestComponent # inline function - flatten the array to make it a basic table - statement: (inlineTestTable) => SELECT inline(SA_TEST_TBL.fields) FROM peopleSoftTestAndScoreData #- statement: PRINT inlineTestTable ## Get Salesforce Student Info - statement: (distinctStudentId) => SELECT DISTINCT TRIM(EMPLID) `id` FROM peopleSoftTestAndScoreData # This statement executes a batch process task where the name, inputTable, outputTable and the results are the parameters needed - statement: | execute task --name batchTask --inputTable distinctStudentId --outputBatchTable batchedStudentId --result resultStatus - statement: REFRESH sfContactReader - statement: (sfContactReaderCount) => SELECT COUNT(*) FROM sfContactReader #- statement: PRINT sfContactReaderCount ## Reject non-matching records due to # Contact # The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table) - statement: (rejectedRecordByNonMatchingContact) => SELECT * FROM inlineTestComponent s LEFT ANTI JOIN sfContactReader sf ON IFNULL(TRIM(s.EMPLID), '') = IFNULL(TRIM(sf.University_Student_ID__c), '') - statement: (rejectedRecordByNonMatchingContactCount) => SELECT COUNT(*) FROM rejectedRecordByNonMatchingContact #- statement: PRINT rejectedRecordByNonMatchingContactCount ## collect info from all tables required for Test. # The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table) - statement: (completeInfoForTest) => SELECT IFNULL(TRIM(t.EMPLID), '') `EMPLID`, IFNULL(TRIM(t.TEST_ID), '') `TEST_ID`, IFNULL(TRIM(t.TEST_COMPONENT), '') `TEST_COMPONENT`, IFNULL(TRIM(t.TEST_DT), '') `TEST_DT`, IFNULL(TRIM(t.LS_DATA_SOURCE), '') `LS_DATA_SOURCE`, IFNULL(TRIM(t.DATE_LOADED), '') `DATE_LOADED`, IFNULL(TRIM(t.LS_DATA_SOURCE_description), '') `LS_DATA_SOURCE_description` FROM peopleSoftTestAndScoreData ts LEFT JOIN inlineTestComponent t ON TRIM(ts.EMPLID) = TRIM(t.EMPLID) AND TRIM(ts.TEST_ID) = TRIM(t.TEST_ID) #- statement: PRINT completeInfoForTest - statement: (completeInfoForTestCount) => SELECT COUNT(*) FROM completeInfoForTest #- statement: PRINT completeInfoForTestCount ## Map Test info with SF fields - statement: (mappedTest) => SELECT concat_ws('-', TRIM(s.EMPLID), TRIM(s.TEST_ID), TRIM(s.TEST_COMPONENT), TRIM(s.TEST_DT), TRIM(s.LS_DATA_SOURCE)) `PeopleSoft_Test_Unique_Identifier__c`, to_date(IFNULL(s.TEST_DT, '')) `hed__Test_Date__c`, IFNULL(s.TEST_ID, '') `Test_Type_Text__c`, sf.Id `hed__Contact__c`, to_date(IFNULL(s.DATE_LOADED, '')) `hed__Date_Received__c`, IFNULL(s.LS_DATA_SOURCE_description, '') `hed__Source__c` FROM completeInfoForTest s INNER JOIN sfContactReader sf ON TRIM(s.EMPLID) = TRIM(sf.University_Student_ID__c) WHERE TRIM(s.EMPLID) != '' AND TRIM(s.TEST_ID) != '' AND TRIM(s.TEST_COMPONENT) != '' AND TRIM(s.TEST_DT) != '' AND TRIM(s.LS_DATA_SOURCE) != '' #- statement: PRINT mappedTest - statement: (mappedTestCount) => SELECT COUNT(*) FROM mappedTest #- statement: PRINT mappedTestCount - statement: (distinctTestId) => SELECT DISTINCT PeopleSoft_Test_Unique_Identifier__c `id` FROM mappedTest # This statement executes a batch process task where the name, inputTable, outputTable and the results are the parameters needed - statement: | execute task --name batchTask --inputTable distinctTestId --outputBatchTable batchedTestId --result resultStatus - statement: REFRESH sfTestReader - statement: (sfTestReaderCount) => SELECT COUNT(*) FROM sfTestReader #- statement: PRINT sfTestReaderCount - statement: (testToUpdate) => SELECT sf.Id, m.hed__Test_Date__c, m.Test_Type_Text__c, m.hed__Contact__c, m.hed__Date_Received__c, m.hed__Source__c FROM mappedTest m INNER JOIN sfTestReader sf ON IFNULL(TRIM(m.PeopleSoft_Test_Unique_Identifier__c), '') = IFNULL(TRIM(sf.PeopleSoft_Test_Unique_Identifier__c), '') #- statement: PRINT testToUpdate - statement: (testToUpdateCount) => SELECT COUNT(*) FROM testToUpdate #- statement: PRINT testToUpdateCount # The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table) - statement: (testToInsert) => SELECT m.* FROM mappedTest m LEFT ANTI JOIN sfTestReader sf ON IFNULL(TRIM(m.PeopleSoft_Test_Unique_Identifier__c), '') = IFNULL(TRIM(sf.PeopleSoft_Test_Unique_Identifier__c), '') #- statement: PRINT testToInsert - statement: (testToInsertCount) => SELECT COUNT(*) FROM testToInsert #- statement: PRINT testToInsertCount # - statement: (updateResults, updateErrors) => UPDATE testToUpdate INTO sfTestWriter # - statement: PRINT updateResults # - statement: PRINT updateErrors # - statement: (insertResults, insertErrors) => INSERT testToInsert INTO sfTestWriter # - statement: PRINT insertResults # - statement: PRINT insertErrors ###### Start - Request with SF id in local file ###### - statement: REFRESH sfTestReader - statement: (logFile) => SELECT "finalMappedTestRecords" name - statement: INSERT logFile INTO executionContext - statement: | (existingTestInfo) => SELECT sf.Id, m.* FROM mappedTest m INNER JOIN sfTestReader sf ON IFNULL(TRIM(m.PeopleSoft_Test_Unique_Identifier__c), '') = IFNULL(TRIM(sf.PeopleSoft_Test_Unique_Identifier__c), '') - statement: INSERT existingTestInfo INTO finalMappedRecordFile ###### End - Request with SF id in local file ###### ###### Start - Test Score ###### ## Reject non-matching records due to # Test # The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table) - statement: (rejectedRecordByNonMatchingTest) => SELECT * FROM inlineTestComponent s LEFT ANTI JOIN sfTestReader sf ON concat_ws('-', TRIM(s.EMPLID), TRIM(s.TEST_ID), TRIM(s.TEST_COMPONENT), TRIM(s.TEST_DT), TRIM(s.LS_DATA_SOURCE)) = IFNULL(TRIM(sf.PeopleSoft_Test_Unique_Identifier__c), '') - statement: (rejectedRecordByNonMatchingTestCount) => SELECT COUNT(*) FROM rejectedRecordByNonMatchingTest #- statement: PRINT rejectedRecordByNonMatchingTestCount ## collect info from all tables required for Test Score. # The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table) - statement: (completeInfoForTestScore) => SELECT IFNULL(TRIM(t.EMPLID), '') `EMPLID`, IFNULL(TRIM(t.TEST_ID), '') `TEST_ID`, IFNULL(TRIM(t.TEST_COMPONENT), '') `TEST_COMPONENT`, IFNULL(TRIM(t.TEST_DT), '') `TEST_DT`, IFNULL(TRIM(t.LS_DATA_SOURCE), '') `LS_DATA_SOURCE`, IFNULL(TRIM(t.TEST_COMPONENT_description), '') `TEST_COMPONENT_description`, IFNULL(TRIM(t.SCORE), '') `SCORE` FROM peopleSoftTestAndScoreData ts LEFT JOIN inlineTestComponent t ON TRIM(ts.EMPLID) = TRIM(t.EMPLID) AND TRIM(ts.TEST_ID) = TRIM(t.TEST_ID) #- statement: PRINT completeInfoForTestScore - statement: (completeInfoForTestScoreCount) => SELECT COUNT(*) FROM completeInfoForTestScore #- statement: PRINT completeInfoForTestScoreCount ## Map Test Score info with SF fields # the Concat_ws function adds different values using the "-" char - statement: (mappedTestScore) => SELECT concat_ws('-', TRIM(s.EMPLID), TRIM(s.TEST_ID), TRIM(s.TEST_COMPONENT), TRIM(s.TEST_DT), TRIM(s.LS_DATA_SOURCE)) `PeopleSoft_Test_Score_Unique_Identifier__c`, sf.Id `hed__Test__c`, IFNULL(s.TEST_COMPONENT_description, '') `Test_Component__c`, IFNULL(s.SCORE, '') `hed__Score__c` FROM completeInfoForTestScore s INNER JOIN sfTestReader sf ON TRIM(sf.PeopleSoft_Test_Unique_Identifier__c) = concat_ws('-', TRIM(s.EMPLID), TRIM(s.TEST_ID), TRIM(s.TEST_COMPONENT), TRIM(s.TEST_DT), TRIM(s.LS_DATA_SOURCE)) WHERE TRIM(s.EMPLID) != '' AND TRIM(s.TEST_ID) != '' AND TRIM(s.TEST_COMPONENT) != '' AND TRIM(s.TEST_DT) != '' AND TRIM(s.LS_DATA_SOURCE) != '' #- statement: PRINT mappedTestScore - statement: (mappedTestScoreCount) => SELECT COUNT(*) FROM mappedTestScore #- statement: PRINT mappedTestScoreCount - statement: (distinctTestScoreId) => SELECT DISTINCT PeopleSoft_Test_Score_Unique_Identifier__c `id` FROM mappedTestScore # This statement executes a batch process task where the name, inputTable, outputTable and the results are the parameters needed - statement: | execute task --name batchTask --inputTable distinctTestScoreId --outputBatchTable batchedTestScoreId --result resultStatus - statement: REFRESH sfTestScoreReader - statement: (sfTestScoreReaderCount) => SELECT COUNT(*) FROM sfTestScoreReader #- statement: PRINT sfTestScoreReaderCount - statement: (testScoreToUpdate) => SELECT sf.Id, m.hed__Test__c, m.Test_Component__c, m.hed__Score__c FROM mappedTestScore m INNER JOIN sfTestScoreReader sf ON IFNULL(TRIM(m.PeopleSoft_Test_Score_Unique_Identifier__c), '') = IFNULL(TRIM(sf.PeopleSoft_Test_Score_Unique_Identifier__c), '') #- statement: PRINT testScoreToUpdate - statement: (testScoreToUpdateCount) => SELECT COUNT(*) FROM testScoreToUpdate #- statement: PRINT testScoreToUpdateCount # The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table) - statement: (testScoreToInsert) => SELECT m.* FROM mappedTestScore m LEFT ANTI JOIN sfTestScoreReader sf ON IFNULL(TRIM(m.PeopleSoft_Test_Score_Unique_Identifier__c), '') = IFNULL(TRIM(sf.PeopleSoft_Test_Score_Unique_Identifier__c), '') #- statement: PRINT testScoreToInsert - statement: (testScoreToInsertCount) => SELECT COUNT(*) FROM testScoreToInsert #- statement: PRINT testScoreToInsertCount # - statement: (updateResults, updateErrors) => UPDATE testScoreToUpdate INTO sfTestScoreWriter # - statement: PRINT updateResults # - statement: PRINT updateErrors # - statement: (insertResults, insertErrors) => INSERT testScoreToInsert INTO sfTestScoreWriter # - statement: PRINT insertResults # - statement: PRINT insertErrors #################################### LOGS - Debugging purpuses ###################################### # - statement: PRINT peopleSoftTestAndScoreDataCount # - statement: PRINT sfContactReaderCount # - statement: PRINT rejectedRecordByNonMatchingContactCount # - statement: PRINT completeInfoForTestCount # - statement: PRINT mappedTestCount # - statement: PRINT sfTestReaderCount # - statement: PRINT testToUpdateCount # - statement: PRINT testToInsertCount # - statement: PRINT rejectedRecordByNonMatchingTestCount # - statement: PRINT completeInfoForTestScoreCount # - statement: PRINT mappedTestScoreCount # - statement: PRINT sfTestScoreReaderCount # - statement: PRINT testScoreToUpdateCount # - statement: PRINT testScoreToInsertCount ###### Start - Request with SF id in local file ###### - statement: REFRESH sfTestScoreReader - statement: (logFile) => SELECT "finalMappedTestScoreRecords" name - statement: INSERT logFile INTO executionContext - statement: | (existingTestScoreInfo) => SELECT sf.Id, m.* FROM mappedTestScore m INNER JOIN sfTestScoreReader sf ON IFNULL(TRIM(m.PeopleSoft_Test_Score_Unique_Identifier__c), '') = IFNULL(TRIM(sf.PeopleSoft_Test_Score_Unique_Identifier__c), '') - statement: INSERT existingTestScoreInfo INTO finalMappedRecordFile ###### End - Request with SF id in local file ###### ###### Start - Update Dates env vars ###### - statement: | (lastModifiedDateTime) => SELECT date_format(from_utc_timestamp(current_timestamp(), 'America/New_York'), "'YYYY-MM-dd'") AS `testLoadedDate` - statement: UPDATE lastModifiedDateTime INTO env.vars #date_format(current_timestamp(), 'YYYY-MM-dd') # date_format(current_timestamp(), 'YYYY-MM-dd') ###### End - Update Dates env vars ###### # Add more statements to convert, join, aggregrate, transform, and integrate your data