Trigger syncs with Dagster

Overview

Hightouch provides a Dagster extension to trigger syncs and gain visibility into sync execution trends. The Git repository contains the latest code.

Setup

Create an API key

The first step is to create a Hightouch API key in your Hightouch workspace settings.

From the API keys tab on the Settings page, select Add API key.
Enter a descriptive Name for your key.
Copy your API key and store it in a safe location. The key will only be displayed once.
Click Create API key.

Your API key provides read/write access to sensitive Hightouch resources and should be kept secure. If you believe your API key may have been compromised, revoke it immediately.

Install library

The simplest way to install the Hightouch Dagster extension is by using pip.

pip install dagster-hightouch

Configure library

To configure the library, provide your Hightouch API key as a resource in the resources.py file:

# resources.py
from dagster_hightouch.resources import ht_resource as hightouch_resource

ht_resource = hightouch_resource.configured(
    {"api_key": "555555-4444-3333-2222-1111111111"},
)

Set sync schedule type

For the Dagster extension to trigger syncs via the Hightouch API, you must set the syncs' schedules to Manual.

You can do this by going to the Syncs overview page, selecting the particular sync you want to trigger with Dagster, and opening the Schedule tab. Here, make sure the Schedule type is set to Manual. You should do this for each sync you want to trigger with Dagster.

You can also find the sync ID on this page, which you need for scheduling syncs in the next step.

Schedule syncs

You can schedule syncs by using the hightouch_sync_op function with the sync IDs of the syncs you want to schedule.

This example creates a job to trigger two syncs—one for Salesforce account objects and one for organization objects—every thirty minutes.

# The hightouch_sync_op calls the Hightouch API to trigger a sync, and then polls the API until the sync either completes or returns an error.

from dagster import ScheduleDefinition, get_dagster_logger, job
from dagster_hightouch.ops import hightouch_sync_op
from .resources import ht_resource

# Set sync IDs as constants

SFDC_ACCOUNTS_SYNC_ID = "23620"
SFDC_ORGS_SYNC_ID = "39619"

# Define configured sync ops

run_ht_sync_accounts = hightouch_sync_op.configured(
    {"sync_id": SFDC_ACCOUNTS_SYNC_ID}, name="hightouch_sfdc_accounts"
)
run_ht_sync_orgs = hightouch_sync_op.configured(
    {"sync_id": SFDC_ORGS_SYNC_ID}, name="hightouch_sfdc_organizations"
)

# Create a job with the defined resources, specifying the dependencies

@job(
    resource_defs={
        "hightouch": ht_resource,
    }
)
def ht_sfdc_job():
    
    ht_orgs = run_ht_sync_orgs(start_after=ht_contacts)
    run_ht_sync_accounts(start_after=ht_orgs)

# Schedule it to run as often as you like with a cron expression
# This example shows every 30 min

every_30_schedule = ScheduleDefinition(job=ht_sfdc_job, cron_schedule="*/30 * * * *")

To learn more, check out the blog post about the extension.