Ingest Historical Events (Using the Tidepool API)
Example Ingest Script from CSV
import requests # you might have to `pip install requests` if you don't have it installed
import csv
from time import sleep
from uuid import uuid4
from datetime import datetime
#################################
# Set up URLs + request headers #
#################################
DATA_FILEPATH = 'PATH_TO_YOUR_DATA.csv'
# submission and validation endpoints - reference https://docs.tidepool.so/reference/
API_BASE = 'https://shoreline.tidepool.so/api/v1/events'
# API keys are project specific - reference https://docs.tidepool.so/docs/manage-api-keys
headers = {
"X-Tidepool-Api-Key": "YOUR_API_KEY_HERE",
"Content-Type": "application/json",
}
############################################
# Load in data, batch it up, and format it #
############################################
# A little utility to batch data, since the API takes up to 100 events per batch
def make_batches(all_entries, n=100):
return [all_entries[i : i + n] for i in range(0, len(all_entries), n)]
# Function to populate + verify a formatted event for upload
def make_event_from_dict(entry):
# Set all required fields from the entry
event = {}
event['text'] = entry['text']
# An iso timestamp with timezone, such as '2023-08-10T21:26:27.302Z'
event['timestamp'] = entry['timestamp']
# These fields must be set, but here are sane defaults if you don't have them on hand
event['id'] = entry.get('id') or str(uuid4())
event['session_id'] = entry.get('session_id') or str(uuid4())
event['user_id'] = entry.get('user_id') or str(uuid4())
event['event_name'] = entry.get('event_name') or 'USER_MESSAGE'
return event
# Note: your CSV must have a header row with column names for csv.DictReader to work
with open(DATA_FILEPATH, 'r') as f:
formatted_events = [make_event_from_dict(d) for d in csv.DictReader(f)]
batched_events = make_batches(formatted_events)
######################
# Make the requests! #
######################
# Validate a single batch of data before going through the rest.
validate_resp = requests.post(f'{API_BASE}/validate', json=batched_events[0], headers=headers)
validate_resp.raise_for_status()
# Submit all the batches!
for i, batch in enumerate(batched_events):
track_response = requests.post(f'{API_BASE}/track', json=batch, headers=headers)
track_response.raise_for_status()
print(f"Submitted batch {i + 1} of {len(batched_events)}")
sleep(0.2)
1. Get an API Key
Reference the Manage API Keys guide for details.
Log in to https://app.tidepool.so and generate an API key from the API Keys section of the settings tab. Note that API keys are project scoped - each key will only allow you to add data to the project it was linked to at creation.
2. Format and Batch your Data
Depending on how your data is stored, you may need to update the schema or reformat it to match what the Tidepool API expects.
Each event should be formatted as a JSON object in the schema described here.
A batch of events is a list of up to 100 event objects. For bulk ingest of historical events, use the maximum batch size possible to improve performance.
3. Validate your Payload
Use the events/validate endpoint to confirm your payload and authorization are set up correctly. This is effectively a test upload that does not submit data and will return a 200 status code if successful.
4. Submit Events
Submit the events using the events/track endpoint. Successful batch submissions will return a 201 status code.
Updated 4 months ago