Prefect integrations for interacting with Hightouch.
Requires an installation of Python 3.7+.
We recommend using a Python virtual environment manager such as pipenv, conda or virtualenv.
These tasks are designed to work with Prefect 2.0. For more information about how to use Prefect, please refer to the Prefect documentation.
pip install prefect-hightouch
Then, register to view the block on Prefect Cloud:
prefect block register -m prefect_hightouch.credentials
Trigger a sync run and wait for completion
from prefect import flow from prefect_hightouch import HightouchCredentials from prefect_hightouch.syncs import trigger_sync_run_and_wait_for_completion @flow def sync_flow(): hightouch_credentials = HightouchCredentials.load("hightouch-token") sync_metadata = trigger_sync_run_and_wait_for_completion( hightouch_credentials=hightouch_credentials, sync_id=12345, full_resync=True, max_wait_seconds=1800, poll_frequency_seconds=10, ) return sync_metadata sync_flow()
List, get, and trigger syncs
from prefect import flow from prefect_hightouch import HightouchCredentials, api_models from prefect_hightouch.syncs import ( list_sync, get_sync, list_sync_runs, trigger_run, trigger_run_custom, ) @flow def hightouch_sync_flow(): hightouch_credentials = HightouchCredentials.load("hightouch-token") # list all syncs syncs = list_sync( hightouch_credentials, order_by=api_models.ListSyncOrderBy.CREATEDAT ) # get first sync sync_id = syncs.id sync = get_sync(hightouch_credentials, sync_id) # list previous runs sync_runs = list_sync_runs(hightouch_credentials, sync_id) # trigger by id sync_run = trigger_run( hightouch_credentials, sync_id, json_body=api_models.TriggerRunInput(full_resync=False), ) # trigger by slug sync_slug = syncs.slug sync_run_2 = trigger_run_custom( hightouch_credentials, json_body=api_models.TriggerRunCustomInput( sync_slug=sync_slug, full_resync=False, ), ) return sync_runs hightouch_sync_flow()
Call API endpoints
If an API endpoint is not exposed as a task, you can call the underlying API endpoint functions, but note, these are not Prefect tasks.
from prefect_hightouch.credentials import HightouchCredentials from prefect_hightouch.api_client.api.default import list_destination credentials = HightouchCredentials.load(token="my-service-token") client = credentials.get_client() response = list_destination.sync_detailed(client=client) data = response.parsed.data
If you encounter any bugs while using
prefect-hightouch, feel free to open an issue in the prefect-hightouch repository.
Feel free to ⭐️ or watch
prefect-hightouch for updates too!
If you'd like to install a version of
prefect-hightouch for development, clone the repository and perform an editable install with
git clone https://github.com/PrefectHQ/prefect-hightouch.git cd prefect-hightouch/ pip install -e ".[dev]" # Install linting pre-commit hooks pre-commit install