PeopleSoft to Salesforce: External SchooAccount

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

# Project Name - PeopleSoft to Salesforce: External SchooAccount

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

# Description - 1 - School Account Recipe
              # This recipe retrieves external school accounts and writes back to Salesforce
              # Select all accounts from EXT_ORG_TBL(parent table) ORG_LOCATION, W9_ORG_CEEB, EXT_ORG_TBL_ADM in PeopleSoft.
              # If organization type is "School" then record type = 'Educational Institution' else "Business Institution".
              # Transform fields and insert/update record into SF by PeopleSoft_External_Org_ID__c
              # Insert account 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. Accounts

## Salesforce Objects Used:
#1. 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 School Account 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: peopleSoftSchoolAccountData
    type: httpV2Reader 
    properties: 
      httpMethod: POST
      headerJson: {
            "Authorization":"Basic {{env.vars.apiAccessToken}}",
            "Content-Type":"application/json"
      }
      acceptInvalidCerts: true
      url: '{{env.vars.midStateInstanceURL}}'
      paginationType: requestBody
      body: 
                          {
          "rowLimit":1000,
          "pageNumber":1,
          "records":[
            {
              "recordName":"EXT_ORG_TBL",
              "includeDescriptionsFor":[
                "EXT_ORG_TYPE"
              ],
              "criteriaFields":[
                {
                  "fieldName":"EXT_ORG_TYPE",
                  "fieldValue":"SCHL",
                  "operator":"="
                }
              ]
            },
            {
              "recordName":"ORG_LOCATION",
              "parentRecordName":"EXT_ORG_TBL",
              "useParentEffectiveDate":true
            },
            {
              "recordName":"W9_ORG_CEEB",
              "parentRecordName":"EXT_ORG_TBL"
            },
            {
              "recordName":"EXT_ORG_TBL_ADM",
              "parentRecordName":"EXT_ORG_TBL"
            }
          ]
        }

schema:     
      fields:
        - name: 'DESCR'
          type: 'string'
        - name: 'DESCRSHORT'
          type: 'string'
        - name: 'EFFDT'
          type: 'string'
        - name: 'EFF_STATUS'
          type: 'string'
        - name: 'EXT_ORG_ID'
          type: 'string'
        - name: 'EXT_ORG_TBL_ADM'
          type: 'object'
          objectSchema:
            name: objectSchemaSchema
            fields:
              - name: 'fields'
                type: 'array'
                items:
                  name: 'arrayElement'
                  type: 'object'
                  objectSchema:
                    name: objectSchemaSchema
                    fields:
                      - name: 'EFFDT'
                        type: 'string'
                      - name: 'EFF_STATUS'
                        type: 'string'
                      - name: 'EXT_CAREER'
                        type: 'string'
                      - name: 'EXT_ORG_ID'
                        type: 'string'
                      - name: 'LS_SCHOOL_TYPE'
                        type: 'string'
                      - name: 'SCHOOL_CODE'
                        type: 'string'
                      - name: 'SCHOOL_DISTRICT'
                        type: 'string'
              - name: 'objectName'
                type: 'string'
              - name: 'objectType'
                type: 'string'
        - name: 'EXT_ORG_TYPE'
          type: 'string'
        - name: 'LASTUPDDTTM'
          type: 'string'
        - name: 'LASTUPDOPRID'
          type: 'string'
        - name: 'ORG_CONTACT'
          type: 'long'
        - name: 'ORG_LOCATION'
          type: 'object'
          objectSchema:
            name: objectSchemaSchema
            fields:
              - name: 'fields'
                type: 'array'
                items:
                  name: 'arrayElement'
                  type: 'object'
                  objectSchema:
                    name: objectSchemaSchema
                    fields:
                      - name: 'ADDRESS1'
                        type: 'string'
                      - name: 'ADDRESS2'
                        type: 'string'
                      - name: 'ADDRESS3'
                        type: 'string'
                      - name: 'CITY'
                        type: 'string'
                      - name: 'COUNTRY'
                        type: 'string'
                      - name: 'COUNTY'
                        type: 'string'
                      - name: 'DESCR'
                        type: 'string'
                      - name: 'DESCRSHORT'
                        type: 'string'
                      - name: 'EFFDT'
                        type: 'string'
                      - name: 'EFF_STATUS'
                        type: 'string'
                      - name: 'EMAILID'
                        type: 'string'
                      - name: 'EXT_ORG_ID'
                        type: 'string'
                      - name: 'LASTUPDDTTM'
                        type: 'string'
                      - name: 'LASTUPDOPRID'
                        type: 'string'
                      - name: 'ORG_LOCATION'
                        type: 'long'
                      - name: 'POSTAL'
                        type: 'string'
                      - name: 'STATE'
                        type: 'string'
              - name: 'objectName'
                type: 'string'
              - name: 'objectType'
                type: 'string'
        - name: 'PROPRIETORSHIP'
          type: 'string'
        - name: 'SETID'
          type: 'string'
        - name: 'W9_ORG_CEEB'
          type: 'object'
          objectSchema:
            name: objectSchemaSchema
            fields:
              - name: 'fields'
                type: 'array'
                items:
                  name: 'arrayElement'
                  type: 'object'
                  objectSchema:
                    name: objectSchemaSchema
                    fields:
                      - name: 'EXT_ORG_ID'
                        type: 'string'
                      - name: 'SAD_SCHOOL_CEEB'
                        type: 'string'
              - name: 'objectName'
                type: 'string'
              - name: 'objectType'
                type: 'string'

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

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

-
    name: sfRecordTypeReader
    type: salesforceReader
    delayedRead: true
    properties:
      useBearerToken: true
      query: 
        SELECT 
          Id,
          Name
        FROM RecordType
        WHERE Name IN ('Educational Institution', 'Business Organization') AND SobjectType = 'Account'
    schema:
      fields: 
        - name: 'Id'
          type: 'string'
        - name: 'Name'
          type: 'string'

# Uses the accountId retrieved in the batch task in the statements section to pull the information needed from Salesforce accounts
# Within the whereclause it iterates over the results from the accountsIds retrieved via the batch task described above
-
    name: sfAccountReader
    type: salesforceReader
    delayedRead: true
    parameterizedBy: batchedAccountId
    properties:
      useBearerToken: true
      query: 
        SELECT 
          Id,
          Name,
          PeopleSoft_External_Org_ID__c
        FROM Account
        WHERE PeopleSoft_External_Org_ID__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_External_Org_ID__c'
          type: 'string'

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

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

-
    name: sfAccountWriter
    type: salesforceBulkWriter
    onError: continue
    isSerial: true
    properties:
      onError: continue
      useBearerToken: true
      sfObject: Account
      batchSize: 100

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

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

-   
    name: finalMappedRecordFile
    type: localFileWriter
    format: csvWriter
    properties:
      fileName: "finalMappedAccountRecords.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 ***********************************************************************************************

# Prints and count the accounts data from PeopleSoft
  #- statement: PRINT peopleSoftSchoolAccountData
  - statement: (peopleSoftSchoolAccountDataCount) => SELECT COUNT(*) FROM peopleSoftSchoolAccountData
  #- statement: PRINT peopleSoftSchoolAccountDataCount

# inline function - flatten the array to make it a basic table
  - statement: (inlineLocations) => SELECT inline(ORG_LOCATION.fields) FROM peopleSoftSchoolAccountData

## Using the ROW_NUMBER function assigns a unique, sequential number to each row, starting with one,
   # according to the ordering of rows within the partition from the information pulled in the previous statement for the address
  ## The PARTITION BY is dividing the rows into small partitions by address preference field
  - statement: |
      (distinctLocations) =>
          SELECT 
            *,
            ROW_NUMBER() OVER (PARTITION BY TRIM(EXT_ORG_ID) ORDER BY ORG_LOCATION) AS Row
          FROM inlineLocations
  #- statement: print distinctLocations

# inline function - flatten the array to make it a basic table
  - statement: (inlineCeebs) => SELECT inline(W9_ORG_CEEB.fields) FROM peopleSoftSchoolAccountData
  #- statement: print inlineCeebs

# inline function - flatten the array to make it a basic table
  - statement: (inlineAdministration) => SELECT inline(EXT_ORG_TBL_ADM.fields) FROM peopleSoftSchoolAccountData
  #- statement: print inlineAdministration

- statement: REFRESH sfRecordTypeReader
  #- statement: PRINT sfRecordTypeReader

# CASE function works as a dictionary and retrieves a specific value depending on the type field value pulled from the table as long as it is not null.
  # 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: (mappedAccounts) => SELECT	
                                    IFNULL(TRIM(c.SAD_SCHOOL_CEEB), '')                               `CEEB_Code__c`,
                                    IFNULL(TRIM(s.EXT_ORG_ID), '')                                    `PeopleSoft_External_Org_ID__c`,

CASE WHEN IFNULL(TRIM(UPPER(s.EXT_ORG_TYPE)), '') = 'SCHL' AND IFNULL(TRIM(UPPER(s.PROPRIETORSHIP)), '') = 'PUBL' THEN 'Public'     
                                      WHEN IFNULL(TRIM(UPPER(s.EXT_ORG_TYPE)), '') = 'SCHL' AND IFNULL(TRIM(UPPER(s.PROPRIETORSHIP)), '') = 'PRIV' THEN 'Private'
                                      WHEN IFNULL(TRIM(UPPER(s.EXT_ORG_TYPE)), '') = 'SCHL' AND IFNULL(TRIM(UPPER(s.PROPRIETORSHIP)), '') = 'RELG' THEN 'Catholic'
                                      WHEN IFNULL(TRIM(UPPER(s.EXT_ORG_TYPE)), '') = 'SCHL' THEN IFNULL(s.PROPRIETORSHIP, '')
                                    Else '' END                                                       `School_Sub_Type__c`,

IFNULL(TRIM(l.CITY), '')                                          `BillingCity`,
                                    IFNULL(TRIM(l.POSTAL), '')                                        `BillingPostalcode`,
                                    IFNULL(TRIM(l.STATE), '')                                         `BillingState`,
                                    IFNULL(TRIM(l.ADDRESS1), '')                                      `BillingStreet`,
                                    IFNULL(TRIM(s.DESCR), '')                                         `Description`,
                                    IFNULL(TRIM(s.DESCR), '')                                         `Name`,

CASE WHEN IFNULL(TRIM(UPPER(s.EXT_ORG_TYPE)), '') = 'SCHL'
                                         THEN (SELECT FIRST(Id) FROM sfRecordTypeReader WHERE NAME = 'Educational Institution')
                                    ELSE (SELECT FIRST(Id) FROM sfRecordTypeReader WHERE NAME = 'Business Organization')
                                    END                                                               `RecordTypeId`,

CASE WHEN IFNULL(TRIM(UPPER(s.EXT_ORG_TYPE)), '') = 'SCHL' AND IFNULL(TRIM(UPPER(a.LS_SCHOOL_TYPE)), '') = 'HS'
                                          THEN 'High School'
                                         WHEN IFNULL(TRIM(UPPER(s.EXT_ORG_TYPE)), '') = 'SCHL' AND IFNULL(TRIM(UPPER(a.LS_SCHOOL_TYPE)), '') = 'PS'
                                          THEN 'Middle School'
                                         WHEN IFNULL(TRIM(UPPER(s.EXT_ORG_TYPE)), '') = 'SCHL' AND IFNULL(TRIM(UPPER(a.LS_SCHOOL_TYPE)), '') = 'NA'
                                          THEN ''
                                        WHEN IFNULL(TRIM(UPPER(s.EXT_ORG_TYPE)), '') = 'SCHL' THEN IFNULL(TRIM(a.LS_SCHOOL_TYPE), '')
                                    ELSE '' END                                                       `School_Level__c`

FROM peopleSoftSchoolAccountData s
                                   LEFT JOIN distinctLocations l ON IFNULL(TRIM(s.EXT_ORG_ID), '') = IFNULL(TRIM(l.EXT_ORG_ID), '')
                                    AND l.Row = 1
                                   LEFT JOIN inlineCeebs c ON IFNULL(TRIM(s.EXT_ORG_ID), '') = IFNULL(TRIM(c.EXT_ORG_ID), '')
                                   LEFT JOIN inlineAdministration a ON IFNULL(TRIM(s.EXT_ORG_ID), '') = IFNULL(TRIM(a.EXT_ORG_ID), '')

#- statement: PRINT mappedAccounts   
  - statement: (mappedAccountsCount) => SELECT COUNT(*) FROM mappedAccounts
  #- statement: PRINT mappedAccountsCount

- statement: (distinctOrgId) => SELECT DISTINCT IFNULL(TRIM(PeopleSoft_External_Org_ID__c), '') `id` FROM mappedAccounts

# 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 distinctOrgId
            --outputBatchTable batchedAccountId
            --result resultStatus

#- statement: print sfAccountReader

- statement: (sfAccountReaderCount) => SELECT COUNT(*) FROM sfAccountReader           
  #- statement: PRINT sfAccountReaderCount

- statement: (accountsToUpdate) => SELECT 
                                      s.Id,
                                      m.CEEB_Code__c,
                                      m.School_Sub_Type__c,
                                      m.BillingCity,
                                      m.BillingPostalcode,
                                      m.BillingState,
                                      m.BillingStreet,
                                      m.Description,
                                      m.Name,
                                      m.RecordTypeId,
                                      m.School_Level__c
                                     FROM mappedAccounts m
                                     INNER JOIN sfAccountReader s ON IFNULL(TRIM(s.PeopleSoft_External_Org_ID__c), '') = IFNULL(TRIM(m.PeopleSoft_External_Org_ID__c), '')

# Count the accounts that needs to be updated
  #- statement: PRINT accountsToUpdate
  - statement: (accountsToUpdateCount) => SELECT COUNT(*) FROM accountsToUpdate
  #- statement: PRINT accountsToUpdateCount

# 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: (accountsToInsert) => SELECT 
                                      m.*
                                     FROM mappedAccounts m
                                     LEFT ANTI JOIN sfAccountReader s ON IFNULL(TRIM(s.PeopleSoft_External_Org_ID__c), '') = IFNULL(TRIM(m.PeopleSoft_External_Org_ID__c), '')

# Count the accounts that needs to be inserted
  #- statement: PRINT accountsToInsert
  - statement: (accountsToInsertCount) => SELECT COUNT(*) FROM accountsToInsert
  #- statement: PRINT accountsToInsertCount

###### Start - Accounts to be updated  #####

- statement: (updateResults, updateErrors) => UPDATE accountsToUpdate INTO sfAccountWriter
  #- statement: PRINT updateResults   
  #- statement: PRINT updateErrors

###### End - Accounts to be updated  #####

###### Start - Accounts to be inserted #####

- statement: (insertResults, insertErrors) => INSERT accountsToInsert INTO sfAccountWriter
  #- statement: PRINT insertResults   
  #- statement: PRINT insertErrors

###### End - Accounts to be inserted #####

#- statement: PRINT mappedAccountsCount
  #- statement: PRINT sfAccountReaderCount
  #- statement: PRINT accountsToUpdateCount
  #- statement: PRINT accountsToInsertCount

# 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 sfAccountReader
  - statement: |
      (existingAccountInfo) =>
          SELECT
            s.Id,
            m.*
          FROM mappedAccounts m
          LEFT JOIN sfAccountReader s ON IFNULL(TRIM(s.PeopleSoft_External_Org_ID__c), '') = IFNULL(TRIM(m.PeopleSoft_External_Org_ID__c), '')

- statement: INSERT existingAccountInfo INTO finalMappedRecordFile

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