Salesforce Upsert Example supporting Lookups

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

# Project Name - SALESFORCE UPSERT EXAMPLE SUPPORTING LOOKUPS

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

# Description - This recipe demonstrates upserting to multiple Salesforce objects with lookup relationships. To run this recipe, change the source and targets. Update the statements to match your SF data sets and click "Run".

# The process is as follows:
# 1. Read data from the source and target
# 2. Transform the source data
# 3. Make the insert and update data sets
# 4. Refresh the source data before creating child objects to get the SF object IDs for newly created SF objects

# Industry - Higher Ed
# Business Process -

# Systems - Salesforce
# Connectors - Salesforce
# Data Flows - Single Direction
# Connection Type - API to API

# Data Models
## Salesforce Objects Used:
#1. Account
#2. Campus__c
#3. Building__c

# Add Recipe notes / Change log information here!

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

# CONNECTORS specify what data will be pulled into the in-memory database during processing

connectors:

# Configure Salesforce credentials in your Environment before running this recipe
  # Salesforce Setup - https://help.lingk.io/en/articles/267-salesforce-connector-setup-guide

###### Start - SFTP Reader connectors ######

## Source Employee Data
-
    name: employeeData
    type: sftpReader
    format: comma
    properties:
      path: /sftplingk/employees.csv

###### End - SFTP Reader connectors ######

## Target data for comparing Salesforce data against Employee data

###### Start - Salesforce Reader connectors ######
-
    name: AccountReaderDepartment
    type: salesforceReader
    properties:
      useBearerToken: true
      query: SELECT Id, RecordTypeId, Name FROM Account where RecordTypeId in (SELECT Id FROM RecordType where isactive = true and SobjectType = 'Account' and Name = 'Test')
    schema:
      fields:
        - name: Id
          type: String
        - name: Name
          type: string  
        - name: RecordTypeId
          type: string

-
    name: Campus__cReader
    type: salesforceReader
    properties:
      useBearerToken: true
      query: SELECT Id, OwnerId, Name FROM Campus__c
    schema:
      fields:
        - name: Id
          type: String
        - name: OwnerId
          type: string
        - name: Name
          type: string

-
    name: Building__cReader
    type: salesforceReader
    properties:
      useBearerToken: true
      query: SELECT Id, Name, Campus__c FROM Building__c
    schema:
      fields:
        - name: Id
          type: String
        - name: Name
          type: string
        - name: Campus__c
          type: string

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

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

-
    name: Campus__c
    type: salesforceBulkWriter
    properties:
      onError: continue
      useBearerToken: true
      sfObject: Campus__c
      # batchSize: 50 # default 10000
      # upsert: true # default false - requires external Id in sfObject

-
    name: Building__c
    type: salesforceBulkWriter
    properties:
      onError: continue
      useBearerToken: true
      sfObject: Building__c
      # batchSize: 50 # default 10000
      # upsert: true # default false - requires external Id in sfObject

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

#   ______                         _       
#  |  ____|                       | |      
#  | |__ ___  _ __ _ __ ___   __ _| |_ ___ 
#  |  __/ _ \| '__| '_ ` _ \ / _` | __/ __|
#  | | | (_) | |  | | | | | | (_| | |_\__ \
#  |_|  \___/|_|  |_| |_| |_|\__,_|\__|___/

readFormats:
-
    name: comma
    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 ***********************************************************************************************

# Campus
# read the source table 
# Do transformations and build the flattened dataset

- statement: |
        (transformed) => SELECT distinct 
            (SELECT Id FROM AccountReaderDepartment LIMIT 1)    Owner__c
            ucase(`Work Location Name`)                         Name 
            FROM employeeData 
            WHERE `Work Location Name` != ''

# START: debugging, remove for production
  #- statement: print transformed

- statement: (count) => select count(*) from transformed
  #- statement: print count
# END: debugging, remove for production

# Build the set of data to be updated
# Note: You can add a where clause to limit updates by field changes, etc. versus blanket updates
  - statement: |
      (transformedToUpdate) => 
        SELECT 
          Campus__cReader.id,
          transformed.Owner__c,
          transformed.Name
        FROM transformed 
          INNER JOIN Campus__cReader ON transformed.Name = Campus__cReader.Name 
        WHERE transformed.Name != Campus__cReader.Name        
#       INNER JOIN on the keys

# START: debugging, remove for production
  #- statement: print transformedToUpdate

- statement: (count) => select count(*) from transformedToUpdate 
  #- statement: print count                  
# END: debugging, remove for production

# Build the set of data to be inserted
# Note: You can add a where clause to limit updates by field changes, etc. versus blanket updates
   # 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: |
          (transformedToInsert) => 
            SELECT 
              transformed.Name,
              transformed.Owner__c
            FROM transformed 
              LEFT ANTI JOIN Campus__cReader ON transformed.Name = Campus__cReader.Name 
#       INNER JOIN on the keys

# START: debugging, remove for production
  #- statement: print transformedToInsert

- statement: (count) => select count(*) from transformedToInsert
  #- statement: print count
# END: debugging, remove for production

# schema validation for the table which is to be inserted.
  - statement: (results, errors) => INSERT transformedToInsert INTO Campus__c

# schema validation for the table which is to be updated.
  - statement: (results, errors) => UPDATE transformedToUpdate INTO Campus__c

# Building

# this statement is needed get the latest data
  - statement: refresh Campus__cReader

# Do transformations and build the flattened dataset

- statement: |
      (transformed) => SELECT distinct 
            employeeData.`Cu Office Address Line 1 Text`    Name, 
            Campus__cReader.Id                              Campus__c 
            FROM employeeData  
              INNER JOIN Campus__cReader ON employeeData.`Work Location Name` = Campus__cReader.`Name`
            WHERE employeeData.`Cu Office Address Line 1 Text` != ""

# START: debugging, remove for production  
  #- statement: print transformed

- statement: (count) => select count(*) from transformed
  #- statement: print count 
# END: debugging, remove for production

- statement: |
      (transformedToUpdate) => 
          SELECT distinct
            Building__cReader.id id, 
            transformed.Name, 
            transformed.Campus__c 
          FROM transformed 
            INNER JOIN Building__cReader ON transformed.Name = Building__cReader.Name 
          WHERE
            transformed.Name != Building__cReader.Name

# START: debugging, remove for production   
  #- statement: print transformedToUpdate

- statement: (count) => select count(*) from transformedToUpdate
  #- statement: print count
# END: debugging, remove for production

# 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: |
        (transformedToInsert) => SELECT distinct 
              transformed.Name, 
              transformed.Campus__c 
            FROM transformed 
              LEFT ANTI JOIN Building__cReader ON transformed.Name = Building__cReader.Name

# START: debugging, remove for upda   
  #- statement: print transformedToInsert

- statement: (count) => select count(*) from transformedToInsert
  #- statement: print count
# END: debugging, remove for production

# schema validation for the table which is to be inserted.
  - statement: (results, errors) => INSERT transformedToInsert INTO Building__c

#### Start - Update Accounts ####

# schema validation for the table which is to be updated.
  - statement: (results, errors) => UPDATE transformedToUpdate INTO Building__c

#### End - Update Accounts ####

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