Skip to main content

Run Spectacles from the API

With the Spectacles API, you can programmatically create runs and get their results. This is useful if you want to trigger Spectacles as part of a data pipeline managed in another tool like Airflow or Prefect.

Let's say you have data transformation jobs running in Airflow. With the Spectacles API, you can run Spectacles after each job completes to verify that nothing has been broken in your Looker instance.

tip

This guide walks through the high-level steps to create and monitor a run using Python and the API. However, you should also check out the detailed reference documentation for the API here.

Getting your API key#

To use the Spectacles API, you will require an API key. You can generate or replace your API key in the Settings page for your Spectacles organisation.

caution

Only one API key can be active at a time, so save your API key somewhere securely when generating it. If you lose it, you'll need to replace it with a new key.

Example in Python#

Here's an example Python script that demonstrates authenticating, creating a run, and getting the results.

import time
import requests
# Define API key and IDs
api_key = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjp7InVpZCI6IldzdEdrNDBVRURoRWxDV1RZaVo5LWFwaSJ9LCJpYXQiOjE2MTgxNDg5MTN9.euyUoiLYtph26H6qPx9U7kqbEvKobbORzRZcMwvDYDD"
org_id = "WstGk40UEDhElCWTYABD"
project_id = "JDmbNFfdWj0B50B6zqE6"
suite_id = "ZhBiJDFGVRkOmpuO9x4m"
# Set the API key in header
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
# Create a run
create_run_url = "https://app.spectacles.dev/api/v1/runs"
payload = {"org_id": org_id, "suite_id": suite_id, "project_id": project_id}
create_run_response = requests.post(url=create_run_url, headers=headers, json=payload)
run_id = create_run_response.json()["run_id"]
# Naively wait for 5 seconds to check
time.sleep(5)
# Get the run's results
run_url = (
f"https://app.spectacles.dev/api/v1/org/{org_id}/proj/{project_id}/run/{run_id}"
)
run_response = requests.get(url=run_url, headers=headers)
print(run_response.json())