PeopleSoft to Salesforce

#   _____           _              _____        __      
#  |  __ \         (_)            |_   _|      / _|     
#  | |__) |___  ___ _ _ __   ___    | |  _ __ | |_ ___  
#  |  _  // _ \/ __| | '_ \ / _ \   | | | '_ \|  _/ _ \ 
#  | | \ \  __/ (__| | |_) |  __/  _| |_| | | | || (_) |
#  |_|  \_\___|\___|_| .__/ \___| |_____|_| |_|_| \___/ 
#                    | |                                
#                    |_|

# Project Name - PeopleSoft to Salesforce - Courses

# Recipe URL - https://app.lingk.io/a/10932/tf/18834

# Description - 5 - PeopleSoft to Salesforce - Courses
                # Select all courses from CRSE_CATALOG(parent table), CRSE_OFFER tables in PeopleSoft.
                # This recipe is using HTTP V2 connector (with pagination) to get courses from peopleSoft.
                # Select all accounts from Salesforce where record type = 'university department'.
                # Reject record if CRSE_ID, CRSE_OFFER_NBR (salesforce external id fields) and ACAD_ORG (account lookup field) are blank.
                # Reject non matching account records.
                # Transformed fields and insert/Update record into SF by PeopleSoft_Course_Unique_Identifier__c.
                # Insert courses 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. Courses

## Salesforce Objects Used:
#1. hed__Course__c
#2. Accounts

# 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 Course 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
- 
    name: peopleSoftCourseData
    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,
            "noEffectiveDateLogic": false,
            "noEffectiveStatusLogic": false,
            "includeFieldTypes": false,
            "records": [
              {
                "recordName": "CRSE_CATALOG",
                "sqlWhereClause": "EXISTS (SELECT 1 FROM PS_CRSE_OFFER BE WHERE BE.CRSE_ID = A.CRSE_ID AND BE.INSTITUTION = '{{env.inst-code}}')"
              },
              {
                "recordName": "CRSE_OFFER",
                "parentRecordName": "CRSE_CATALOG"
              },
              {
                "recordName": "CRSE_ATTRIBUTES",
                "parentRecordName": "CRSE_CATALOG"
              }
            ]
          }
    schema:     
      fields:
        - name: 'ALLOW_MULT_ENROLL'
          type: 'string'
        - name: 'COMPONENT_PRIMARY'
          type: 'string'
        - name: 'CONSENT'
          type: 'string'
        - name: 'COURSE_TITLE_LONG'
          type: 'string'
        - name: 'CRSE_CONTACT_HRS'
          type: 'long'
        - name: 'CRSE_COUNT'
          type: 'long'
        - name: 'CRSE_ID'
          type: 'string'
        - name: 'CRSE_ATTRIBUTES'
          type: 'object'
          objectSchema:
            name: objectSchemaSchema
            fields:
              - name: 'fields'
                type: 'array'
                items:
                  name: 'arrayElement'
                  type: 'object'
                  objectSchema:
                    name: objectSchemaSchema
                    fields:
                      - name: 'CRSE_ID'
                        type: 'string'
                      - name: 'CRSE_ATTR'
                        type: 'string'
                      - name: 'CRSE_ATTR_VALUE'
                        type: 'string'
        - name: 'CRSE_OFFER'
          type: 'object'
          objectSchema:
            name: objectSchemaSchema
            fields:
              - name: 'fields'
                type: 'array'
                items:
                  name: 'arrayElement'
                  type: 'object'
                  objectSchema:
                    name: objectSchemaSchema
                    fields:
                      - name: 'ACAD_ORG'
                        type: 'string'
                      - name: 'CRSE_ID'
                        type: 'string'
                      - name: 'CATALOG_NBR'
                        type: 'string'
                      - name: 'CRSE_OFFER_NBR'
                        type: 'long'
                      - name: 'EFFDT'
                        type: 'string'
                      - name: 'INSTITUTION'
                        type: 'string'  
                      - name: 'SUBJECT'
                        type: 'string'        
                      - name: 'ACAD_CAREER'
                        type: 'string'                       
              - name: 'objectName'
                type: 'string'
              - name: 'objectType'
                type: 'string'
        - name: 'DESCR'
          type: 'string'
        - name: 'EFFDT'
          type: 'string'
        - name: 'SSR_COMPONENT'
          type: 'string'
        - name: 'SSR_DROP_CONSENT'
          type: 'string'
        - name: 'UNITS_ACAD_PROG'
          type: 'long'
        - name: 'UNITS_MAXIMUM'
          type: 'long'
        - name: 'UNITS_MINIMUM'
          type: 'long'
        - name: 'UNITS_REPEAT_LIMIT'
          type: 'long'
        - name: 'rowNumber'
          type: 'long'

###### End - HTTP Reader connectors ######

###### Start - Salesforce Reader connectors ######

# Salesforce Account Reader
-
    name: sfAccountReader
    type: salesforceReader
    delayedRead: true
    properties:
      useBearerToken: true
      query: SELECT 
              Id, 
              RecordTypeId, 
              Name,
              PeopleSoft_Univ_Department_Code__c
             FROM Account 
    schema:
      fields:
        - name: Id
          type: String
        - name: Name
          type: string  
        - name: RecordTypeId
          type: string 
        - name: PeopleSoft_Univ_Department_Code__c
          type: string

# Salesforce Course Reader
-
    name: sfCourseReader
    type: salesforceReader
    delayedRead: true
    parameterizedBy: batchedCourseId
    properties:
      useBearerToken: true
      charsToBeEncoded: "-"
      query: 
        SELECT 
          Id,
          PeopleSoft_Course_Unique_Identifier__c
        FROM hed__Course__c
        WHERE PeopleSoft_Course_Unique_Identifier__c IN (
            {% for element in var.output %}
            '{{element.id}}'{% if not loop.last %},{% endif %}
            {% endfor %}
          ) 
    schema:
      fields:
        - name: 'Id'
          type: 'string'
        - name: 'PeopleSoft_Course_Unique_Identifier__c'
          type: 'string'

###### End - Salesforce Reader connectors ######

###### Start - Salesforce Writer connectors ######

# Salesforce Course Writer        
-
    name: sfCourseWriter
    type: salesforceBulkWriter
    onError: continue
    isSerial: true
    properties:
      onError: continue
      useBearerToken: true
      sfObject: hed__Course__c
      batchSize: 100

###### End - Salesforce Writer connectors ######

###### Start - Local Writer connectors ######

# Local File Writer
-   
    name: finalMappedRecordFile
    type: localFileWriter
    format: csvWriter
    properties:
      fileName: "finalMappedCourseRecords.csv"

###### End - Salesforce Writer connectors ######

#   _______        _        
#  |__   __|      | |       
#     | | __ _ ___| | _____ 
#     | |/ _` / __| |/ / __|
#     | | (_| \__ \   <\__ \
#     |_|\__,_|___/_|\_\___/

# Batch task used for parameterized by statements/connectors

tasks:

## This batch task is being used under the statements sections in order to send information back to a connector to form a batching 'for' loop
 # documentation - https://help.lingk.io/en/articles/241-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 ***********************************************************************************************

# Print and count course data from Peoplesoft
  #- statement: PRINT peopleSoftCourseData
  - statement: (peopleSoftCourseDataCount) => SELECT COUNT(*) FROM peopleSoftCourseData
  #- statement: PRINT peopleSoftCourseDataCount

# inline function - flatten the array to make it a basic table
  - statement: (inlineCourseOffer) => SELECT inline(CRSE_OFFER.fields) FROM peopleSoftCourseData
  #- statement: PRINT inlineCourseOffer

# inline function - flatten the array to make it a basic table
  - statement: (inlineCourseAttribute) => SELECT inline(CRSE_ATTRIBUTES.fields) FROM peopleSoftCourseData
  #- statement: PRINT inlineCourseAttribute

# The attribute_list column is retrieved as an array of attribute:value due to the collection_list function is concatenating CRSE_ATTR as the field and CRSE_ATTR_VALUE as the value
  - statement: 
      (attributeGroup) =>
          SELECT
            IFNULL(TRIM(CRSE_ID), '')                    `CRSE_ID`,
            collect_list(
                concat_ws(':', IFNULL(CRSE_ATTR, ''), IFNULL(CRSE_ATTR_VALUE, ''))
            )                                            `attribute_list`
          FROM inlineCourseAttribute
          GROUP BY TRIM(CRSE_ID)
  #- statement: PRINT attributeGroup

# Gets the data from the Salesforce Account connector and count the amount of records retrieved
  - statement: (sfAccountReader) => Select * From sfAccountReader WHERE RecordTypeId IN (SELECT Id FROM RecordType WHERE isactive = true AND SobjectType = 'Account' AND Name = 'University Department')
  - statement: (sfAccountReaderCount) => SELECT COUNT(*) FROM sfAccountReader
  #- statement: PRINT sfAccountReaderCount

# Retrieved the records from the inlineCourseOffer table where the CRSE_ID and CRSE_OFFER_NBR are empties
  - statement: (rejectedRecordDueToBlankCourseidAndAccount) =>
                    SELECT
                        *
                    FROM inlineCourseOffer
                    WHERE TRIM(CRSE_ID) = '' AND TRIM(CRSE_OFFER_NBR) = ''

#- statement: PRINT rejectedRecordDueToBlankCourseidAndAccount
  - statement: (rejectedRecordDueToBlankCourseidAndAccountCount) => SELECT COUNT(*) FROM rejectedRecordDueToBlankCourseidAndAccount
  #- statement: PRINT rejectedRecordDueToBlankCourseidAndAccountCount

# 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: (rejectedRecordDueToMissingAccount) =>
                    SELECT
                        co.*
                    FROM inlineCourseOffer co
                    LEFT ANTI JOIN sfAccountReader sf ON IFNULL(TRIM(co.ACAD_ORG), '') = IFNULL(TRIM(sf.PeopleSoft_Univ_Department_Code__c), '')

#- statement: PRINT rejectedRecordDueToMissingAccount
  - statement: (rejectedRecordDueToMissingAccountCount) => SELECT COUNT(*) FROM rejectedRecordDueToMissingAccount
  #- statement: PRINT rejectedRecordDueToMissingAccountCount

# Retrieves a list of records that matches the courses from inlineCourseOffer table, Salesforce Accounts and Peoplesoft data but is not attributeGroup table by CRSE_ID field
  # 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: (mappedCourses) => 
                    SELECT
                      concat_ws('-', TRIM(co.CRSE_ID), TRIM(co.CRSE_OFFER_NBR))                      `PeopleSoft_Course_Unique_Identifier__c`,
                      IFNULL(TRIM(c.DESCR), '')                                                      `Name`,

lpad(Replace(string(format_number(cast(co.CATALOG_NBR as long), '#,##0')), ',', '-'), 10, '0')
                                                                                                    `PeopleSoft_Catalog_Number__c`,
                      IFNULL(TRIM(c.COURSE_TITLE_LONG), '')                                          `hed__Extended_Description__c`,
                      acc.Id                                                                         `hed__Account__c`,
                      IFNULL(TRIM(c.UNITS_MINIMUM), '')                                              `hed__Credit_Hours__c`,
                      concat_ws(', ', attribute_list)                                                `Course_Attribute__c`,

lpad(Replace(string(format_number(cast(co.CRSE_ID as long), '#,##0')), ',', '-'), 7, '0')  
                                                                                                     `PeopleSoft_Course_ID__c`,
                      IFNULL(co.SUBJECT, '')                                                         `PeopleSoft_Subject__c`,
                      IFNULL(co.ACAD_CAREER,'')                                                      `PS_Acad_Career__c`

FROM inlineCourseOffer co
                    INNER JOIN sfAccountReader acc ON IFNULL(TRIM(co.ACAD_ORG), '') = IFNULL(TRIM(acc.PeopleSoft_Univ_Department_Code__c), '')
                    INNER JOIN peopleSoftCourseData c ON IFNULL(TRIM(co.CRSE_ID), '') = IFNULL(TRIM(c.CRSE_ID), '')                
                    LEFT JOIN attributeGroup att ON IFNULL(TRIM(co.CRSE_ID), '') = IFNULL(TRIM(att.CRSE_ID), '')
                      WHERE TRIM(co.CRSE_ID) != '' AND TRIM(co.CRSE_OFFER_NBR) != ''

#- statement: PRINT mappedCourses        
  - statement: (mappedCoursesCount) => SELECT COUNT(*) FROM mappedCourses
  #- statement: PRINT mappedCoursesCount

- statement: (distinctCourseId) => SELECT DISTINCT PeopleSoft_Course_Unique_Identifier__c `id` FROM mappedCourses
  #- statement: print distinctCourseId

# Batch described in task section
  # 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 distinctCourseId
            --outputBatchTable batchedCourseId
            --result resultStatus

- statement: REFRESH sfCourseReader
  - statement: (sfCourseReaderCount) => SELECT COUNT(*) FROM sfCourseReader
  #- statement: PRINT sfCourseReaderCount

# Retrieve the records from the mappedCourses table that matches with the existing salesforce records in order to get the list that needs to be updated
  - statement:
      (courseToUpdate) =>
          SELECT
            sf.Id,
            m.PeopleSoft_Course_Unique_Identifier__c,
            m.Name,
            m.PeopleSoft_Catalog_Number__c,
            m.hed__Extended_Description__c,
            m.hed__Account__c,
            m.hed__Credit_Hours__c,
            m.Course_Attribute__c,
            m.PeopleSoft_Course_ID__c,
            m.PeopleSoft_Subject__c,
            m.PS_Acad_Career__c
          FROM mappedCourses m
          INNER JOIN sfCourseReader sf ON IFNULL(TRIM(m.PeopleSoft_Course_Unique_Identifier__c), '') = IFNULL(TRIM(sf.PeopleSoft_Course_Unique_Identifier__c), '')

#- statement: PRINT courseToUpdate
  - statement: (courseToUpdateCount) => SELECT COUNT(*) FROM courseToUpdate
  #- statement: PRINT courseToUpdateCount

# 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:
      (courseToInsert) =>
          SELECT
            m.*
          FROM mappedCourses m
          LEFT ANTI JOIN sfCourseReader sf ON IFNULL(TRIM(m.PeopleSoft_Course_Unique_Identifier__c), '') = IFNULL(TRIM(sf.PeopleSoft_Course_Unique_Identifier__c), '')

#- statement: PRINT courseToInsert
  - statement: (courseToInsertCount) => SELECT COUNT(*) FROM courseToInsert
  #- statement: PRINT courseToInsertCount

###### Start - Update Salesforce Course ######

- statement: (updateResult, updateError) => UPDATE courseToUpdate INTO sfCourseWriter
  #- statement: PRINT updateResult
  #- statement: PRINT updateError

###### End - Update Salesforce Course ######

###### Start - Insert Salesforce Course ######

- statement: (insertResult, insertError) => INSERT courseToInsert INTO sfCourseWriter
  #- statement: PRINT insertResult
  #- statement: PRINT insertError

###### End - Insert Salesforce Course ######

#################################### LOGS - Debugging purpuses ######################################

#  - statement: PRINT peopleSoftCourseDataCount
#  - statement: PRINT sfAccountReaderCount
#  - statement: PRINT rejectedRecordDueToBlankCourseidAndAccountCount
#  - statement: PRINT rejectedRecordDueToMissingAccountCount
#  - statement: PRINT mappedCoursesCount
#  - statement: PRINT sfCourseReaderCount
#  - statement: PRINT courseToUpdateCount
#  - statement: PRINT courseToInsertCount

## Insert request with SF id in local file
  # 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: REFRESH sfCourseReader
  - statement: |
      (existingCoursesInfo) =>
          SELECT
            sf.Id,
            m.*
          FROM mappedCourses m
          LEFT JOIN sfCourseReader sf ON IFNULL(TRIM(m.PeopleSoft_Course_Unique_Identifier__c), '') = IFNULL(TRIM(sf.PeopleSoft_Course_Unique_Identifier__c), '')

- statement: INSERT existingCoursesInfo INTO finalMappedRecordFile

# Add more statements to convert, join, aggregrate, transform, and integrate your data