Sync with external workflows using Python
Sync data with Census

Sync data with Census

Census synchronizes customer insights from your cloud data warehouse to all the SaaS tools your team utilizes, including Salesforce, HubSpot, Customer.io, Slack, and Zendesk.

With Python Actions, Y42 enables programmatic triggering of Census syncs via its API.

Create Sync in Census

Syncs establish connections between your Data Warehouse sources and Destination Services. For detailed information, refer to the Census documentation (opens in a new tab).

Create a new sync or use an existing one. Choose as source for the Sync, one of the tables published by Y42 in your data warehouse.

Sync between BigQuery table and Google Sheets.

Sync between BigQuery table and Google Sheets.

Trigger Sync via its API

You can trigger each sync programmatically through its API. The sync configuration page provides the trigger API endpoint.

To initiate the sync, use the following POST request format:


_10
https://bearer:{api_token}@app.getcensus.com/api/v1/syncs/{sync_id}/trigger

Initiate the sync

Initiate the sync

Store the API Token as a Secret in Y42

You can securely store API tokens (or any other secrets) within Y42 using the Secrets menu.

Store the API Token as a Secret in Y42.

Store the API Token as a Secret in Y42.

Create Python Action asset

  1. Create the Python Action asset: Use the POST request method outlined earlier and reference the API Token from the Secrets vault.
trigger_sync_to_census.py

_26
import json
_26
import requests
_26
from requests.exceptions import ConnectionError
_26
_26
from y42.v1.decorators import data_action
_26
import logging
_26
_26
@data_action
_26
def trigger_sync_to_census(context, assets):
_26
_26
api_token = context.secrets.get('census_api_token')
_26
sync_id = '567890'
_26
_26
url = f'https://bearer:{api_token}@app.getcensus.com/api/v1/syncs/{sync_id}/trigger'
_26
_26
response = requests.post(url) # census does not require headers or a payload
_26
_26
# log the response from your POST request
_26
logging.info(response.status_code)
_26
response_payload = json.loads(response.text)
_26
_26
# throw error when job was not triggered successfully
_26
if response_payload['status'] == 'error':
_26
raise ConnectionError(f"Triggering the job {response_payload['data']['sync_run_id']} failed with error: {response_payload['message']}")
_26
else:
_26
logging.info(response_payload)

  1. Specify an upstream dependency (Optional, but recommended): In the Dependencies tab, you can specify an upstream dependency for the Python asset. This dependency acts as a trigger, activating the Python Action to synchronize data with a third-party application upon the successful completion of the upstream asset.

It's often practical to select the source table, which was utilized to establish the Sync, as the dependency.

Specify an upstream dependency to act as the trigger for the Python asset.

Specify an upstream dependency to act as the trigger for the Python asset.

Build asset

Build the asset using dag selectors or via the Build history tab.

trigger_exposure

_10
y42 build -s +exposure:trigger_sync_to_census

The + selector, when used in front of the asset name in the command, also triggers all upstream dependencies.

Build history tab.

Build history tab.

Verify Sync status in Census

After returning to the Census interface, the sync should be activated, having data to flowing to your configured destination. Congrats! 🎉

Sync successfully triggered.

Sync successfully triggered.

Sync with external workflows using PythonSync data with Hightouch