Hightouch API: Hightouch Public Rest API to access syncs, models, sources and destinations
Create a new destination
import hightouch
from hightouch.models import shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = shared.DestinationCreate(
configuration={
'key': '<value>',
},
name='<value>',
slug='<value>',
type='<value>',
)
res = s.create_destination(req)
if res.one_of is not None:
# handle response
pass
operations.CreateDestinationResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
409,422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Create a new model
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.CreateModelRequest(
model_create=shared.ModelCreate(
is_schema=False,
name='<value>',
primary_key='<value>',
query_type='<value>',
slug='<value>',
source_id=8761.56,
),
)
res = s.create_model(req)
if res.one_of is not None:
# handle response
pass
operations.CreateModelResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
409,422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Create a new source
import hightouch
from hightouch.models import shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = shared.SourceCreate(
configuration={
'key': '<value>',
},
name='<value>',
slug='<value>',
type='<value>',
)
res = s.create_source(req)
if res.one_of is not None:
# handle response
pass
Parameter |
Type |
Required |
Description |
request |
shared.SourceCreate |
✔️ |
The request object to use for the request. |
operations.CreateSourceResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
409,422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Create a new sync
import hightouch
from hightouch.models import shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = shared.SyncCreate(
configuration={
'key': '<value>',
},
destination_id=8797.77,
disabled=False,
model_id=2438.5,
schedule=shared.SyncCreateSchedule(
schedule=shared.CronSchedule(
expression='<value>',
),
type='<value>',
),
slug='<value>',
)
res = s.create_sync(req)
if res.one_of is not None:
# handle response
pass
Parameter |
Type |
Required |
Description |
request |
shared.SyncCreate |
✔️ |
The request object to use for the request. |
operations.CreateSyncResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
409,422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Retrieve a destination based on its Hightouch ID
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.GetDestinationRequest(
destination_id=4856.96,
)
res = s.get_destination(req)
if res.destination is not None:
# handle response
pass
operations.GetDestinationResponse
Error Object |
Status Code |
Content Type |
errors.SDKError |
4xx-5xx |
/ |
Retrieve models from model ID
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.GetModelRequest(
model_id=7962.16,
)
res = s.get_model(req)
if res.model is not None:
# handle response
pass
operations.GetModelResponse
Error Object |
Status Code |
Content Type |
errors.SDKError |
4xx-5xx |
/ |
Retrieve source from source ID
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.GetSourceRequest(
source_id=4378.62,
)
res = s.get_source(req)
if res.source is not None:
# handle response
pass
operations.GetSourceResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Retrieve sync from sync ID
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.GetSyncRequest(
sync_id=7319.86,
)
res = s.get_sync(req)
if res.sync is not None:
# handle response
pass
operations.GetSyncResponse
Error Object |
Status Code |
Content Type |
errors.SDKError |
4xx-5xx |
/ |
Get the status of a sync sequence run.
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.GetSyncSequenceRunRequest(
sync_sequence_run_id='<value>',
)
res = s.get_sync_sequence_run(req)
if res.sync_sequence_status_output is not None:
# handle response
pass
operations.GetSyncSequenceRunResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
List the destinations in the user's workspace
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.ListDestinationRequest()
res = s.list_destination(req)
if res.object is not None:
# handle response
pass
operations.ListDestinationResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
List all the models in the current workspace including parent and related models
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.ListModelRequest()
res = s.list_model(req)
if res.object is not None:
# handle response
pass
operations.ListModelResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
List all the sources in the current workspace
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.ListSourceRequest()
res = s.list_source(req)
if res.object is not None:
# handle response
pass
operations.ListSourceResponse
Error Object |
Status Code |
Content Type |
errors.SDKError |
4xx-5xx |
/ |
List all the syncs in the current workspace
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.ListSyncRequest()
res = s.list_sync(req)
if res.object is not None:
# handle response
pass
operations.ListSyncResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
List all sync runs under a sync
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.ListSyncRunsRequest(
sync_id=8858.62,
)
res = s.list_sync_runs(req)
if res.object is not None:
# handle response
pass
operations.ListSyncRunsResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Trigger a new run for the given sync.
If a run is already in progress, this queues a sync run that will get
executed immediately after the current run completes.
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.TriggerRunRequest(
sync_id='<value>',
)
res = s.trigger_run(req)
if res.trigger_run_output is not None:
# handle response
pass
operations.TriggerRunResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Trigger a new run globally based on sync id or sync slug
If a run is already in progress, this queues a sync run that will get
executed immediately after the current run completes.
import hightouch
from hightouch.models import shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = shared.TriggerRunCustomInput()
res = s.trigger_run_custom(req)
if res.one_of is not None:
# handle response
pass
operations.TriggerRunCustomResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.TriggerRunIDGraphRequest(
graph_id='<value>',
)
res = s.trigger_run_id_graph(req)
if res.trigger_run_id_graph_output is not None:
# handle response
pass
operations.TriggerRunIDGraphResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Trigger a new run for the given sync sequence.
If a run is already in progress, this queues a sync sequence run that will be
executed immediately after the current run completes.
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.TriggerSequenceRunRequest(
sync_sequence_id='<value>',
)
res = s.trigger_sequence_run(req)
if res.trigger_sequence_run_output is not None:
# handle response
pass
operations.TriggerSequenceRunResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Update an existing destination
Patch a destination based on its Hightouch ID
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.UpdateDestinationRequest(
destination_update=shared.DestinationUpdate(),
destination_id=8928.88,
)
res = s.update_destination(req)
if res.one_of is not None:
# handle response
pass
operations.UpdateDestinationResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Update an existing model
Patch a model based on its Hightouch ID
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.UpdateModelRequest(
model_update=shared.ModelUpdate(),
model_id=1027.03,
)
res = s.update_model(req)
if res.one_of is not None:
# handle response
pass
operations.UpdateModelResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Update an existing source
Patch a source based on its Hightouch ID
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.UpdateSourceRequest(
source_update=shared.SourceUpdate(),
source_id=6585.68,
)
res = s.update_source(req)
if res.one_of is not None:
# handle response
pass
operations.UpdateSourceResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |
Update an existing sync
Patch a sync based on its Hightouch ID
import hightouch
from hightouch.models import operations, shared
s = hightouch.Hightouch(
security=shared.Security(
bearer_auth="<YOUR_BEARER_TOKEN_HERE>",
),
)
req = operations.UpdateSyncRequest(
sync_update=shared.SyncUpdate(),
sync_id=5066.51,
)
res = s.update_sync(req)
if res.one_of is not None:
# handle response
pass
operations.UpdateSyncResponse
Error Object |
Status Code |
Content Type |
errors.ValidateErrorJSON |
422 |
application/json |
errors.SDKError |
4xx-5xx |
/ |