How can I leverage the Zenysis API?
Generating a payload
- First, a Site Administrator will need to generate an API access token:
- Go to the “Admin” page, select a user for which you want to generate a token for (token will only allow same access as that user is allowed)
- Switch to the “API Token” tabs
- Click “Generate Access Token”
- Click “Copy” and save the token for future use
- Click “Save” - don’t miss this step, since that is what makes the token active
- A Site Administrator will share this access token with the user of interest so they can leverage this token on their own
- Create a query in AQT or access a query from a dashboard
- Click “Share” in the above navigation bar
- Click on the “Export Data” tab and select either the “Disaggregated Data” or the “AQT Data”, and copy the endpoint
- Now you can retrieve the data by making HTTP GET requests to the endpoint you just copied. To authenticate your requests add them to “Authorization” token as a Bearer method.
- Example using curl (replace $URL and $TOKEN with the URL from share dialogue and token respectively):
` `curl $URL -H "Authorization: Bearer $TOKEN" ``
Python and node.js examples can be found at the bottom of this article.
Pagination
When querying a lot of data it might be beneficial to receive them in smaller chunks, or pages. It can be achieved using the Pagination feature. Pages are requested using the page and per_page query string arguments.
For example, https://internal.zenysis.com/api2/query/table?h=e7df5561a9b04cfb5aac9b6eb97915df&page=2&per_page=100
Sorting
To use sorting you can specify a sort GET-parameter which should contain a JSON formatted object specifying what fields to sort on. The keys should be fields and values are booleans. If it’s false, sorting happens in an ascending order, if it’s true then it happens in descending order.
The object is URL Encoded so it can be correctly parsed by the HTTP stack. The initial object here was: {“timestamp”: false}
Filtering
Similarly to sorting, for filtering use a where GET parameter which is a JSON object. For instance, to only receive data for a particular County: https://internal.zenysis.com/api2/query/table?h=e7df5561a9b04cfb5aac9b6eb97915df&page=1&where=%7B%22CountyCity%22%3A%22Augusta%22%7D
The initial JSON here is: {“CountyCity”: “Augusta”}
Python example:
#1/usr/bin/env python'''Supply your token as ZEN_API_TOKEN env var, for instance, to read it from a file named token.txt:ZEN_API_TOKEN=$(cat token.txt) python rbm_api.py'''import jsonimport osimport urllib.parseimport requests# ✅ Handle missing API token gracefullyACCESS_TOKEN = os.getenv('ZEN_API_TOKEN')if not ACCESS_TOKEN: raise ValueError( "🔴 Error: API token is missing!\n" "➡️ Set ZEN_API_TOKEN as an environment variable:\n" " export ZEN_API_TOKEN='your_api_token_here'\n" "Or store it in a file (token.txt) and run:\n" " export ZEN_API_TOKEN=$(cat token.txt)\n" )# ✅ Ensure QUERY_LINK is properly formatted (List of URLs)QUERY_LINK = [ 'your_query_link']# ✅ Validate URLs before processingfor url in QUERY_LINK: parsed_url = urllib.parse.urlparse(url) if not parsed_url.scheme or not parsed_url.netloc: raise ValueError(f"🔴 Error: Invalid URL detected: {url}")# ✅ Process multiple URLs properlyBASE_API_URLS = [ urllib.parse.urlunparse(urllib.parse.urlparse(url)[:2] + ('/', '', None, None)) for url in QUERY_LINK]INDICATOR_MAPPING_QUERY_ENDPOINT = '/api/field_names'def get_user_session(): session = requests.Session() session.headers['Authorization'] = f'Bearer {ACCESS_TOKEN}' return sessiondef convert_mappings(unmapped_json): mappings = get_mappings() data = [] for chunk in unmapped_json['data']: new_chunk = {} for indicator in chunk: mapped_indicator = mappings.get(indicator.split('__')[0], indicator) new_chunk[mapped_indicator] = str(chunk[indicator]) data.append(new_chunk) return {"data": data}def write_response_to_file(response, map_indicators, output_file_path): if response.status_code != 200: print(f"❌ Error: API request failed with status code {response.status_code}") print("📜 Response Content:", response.text) # Debugging output raise ValueError(f"🔴 Invalid response from API. Check the request and try again.") try: with open(output_file_path, 'w') as output_file: json.dump( convert_mappings(response.json()) if map_indicators else response.json(), output_file, indent=4 ) print(f"✅ Successfully saved response to {output_file_path}") except Exception as e: raise ValueError(f"🔴 Error writing response to file {output_file_path}: {e}")def request_and_write_query(endpoint_url, map_indicators, output_file_path): try: with get_user_session() as session: print(f"🔄 Requesting data from: {endpoint_url}") response = session.get(endpoint_url) write_response_to_file(response, map_indicators, output_file_path) except requests.RequestException as e: raise ValueError(f"🔴 Network error while accessing {endpoint_url}: {e}")def request_and_write_queries(query_links, map_indicators, output_file_paths): if len(query_links) != len(output_file_paths): raise ValueError( f"🔴 Error: Number of output files must match the number of query links.\n" f"➡️ {len(query_links)} query links provided, but {len(output_file_paths)} output files specified." ) with get_user_session() as session: for i, query_link in enumerate(query_links): print(f"🔄 Processing query {i+1}/{len(query_links)}: {query_link}") try: response = session.get(query_link) write_response_to_file(response, map_indicators, output_file_paths[i]) except requests.RequestException as e: print(f"🔴 Failed to fetch data from {query_link}: {e}")def get_mappings(): with get_user_session() as session: endpoint_url = urllib.parse.urljoin(BASE_API_URLS[0], INDICATOR_MAPPING_QUERY_ENDPOINT) try: response = session.get(endpoint_url) if response.status_code != 200: print(f"❌ Error: Failed to retrieve mappings from {endpoint_url}") print("📜 Response Content:", response.text) raise ValueError(f"🔴 Invalid response: {response.status_code}") x = [a.replace('\r', '') for a in response.text.split('\n') if a] return {pair.split(',')[0]: pair.split(',')[1] for pair in x} except requests.RequestException as e: raise ValueError(f"🔴 Network error while retrieving mappings from {endpoint_url}: {e}")try: request_and_write_queries( QUERY_LINK, True, ['table_raw_data.json'] )except ValueError as e: print(f"⚠️ Critical Error: {e}")
node.js example:
import fs from 'fs';import fetch from 'node-fetch';import { parse, format } from 'url';// ✅ Handle missing API token gracefullyconst API_TOKEN = process.env.ZEN_API_TOKEN;if (!API_TOKEN) { console.error( "🔴 Error: API token is missing!\n" + "➡️ Set ZEN_API_TOKEN as an environment variable:\n" + " export ZEN_API_TOKEN='your_api_token_here'\n" + "Or store it in a file (token.txt) and run:\n" + " export ZEN_API_TOKEN=$(cat token.txt)\n" ); process.exit(1);}// ✅ Ensure QUERY_LINK is properly formatted (List of URLs)const QUERY_LINKS = [ 'your_query_link'];// ✅ Validate URLs before processingQUERY_LINKS.forEach((url) => { const parsedUrl = parse(url); if (!parsedUrl.protocol || !parsedUrl.host) { console.error(🔴 Error: Invalid URL detected: ${url}); process.exit(1); }});// ✅ Process multiple URLs properlyconst BASE_API_URLS = QUERY_LINKS.map((url) => { const parsedUrl = parse(url); return format({ protocol: parsedUrl.protocol, host: parsedUrl.host });});const INDICATOR_MAPPING_QUERY_ENDPOINT = '/api/field_names';// ✅ Function to get session headersfunction getSessionHeaders() { return { Authorization: Bearer ${API_TOKEN}, 'Accept': 'application/json' };}// ✅ Function to fetch and convert mappingsasync function getMappings() { const endpointUrl = `${BASE_API_URLS[0]}${INDICATOR_MAPPING_QUERY_ENDPOINT}`; try { const response = await fetch(endpointUrl, { method: 'GET', headers: getSessionHeaders() }); if (!response.ok) { console.error(❌ Error: Failed to retrieve mappings from ${endpointUrl}); console.error("📜 Response Content:", await response.text()); process.exit(1); } const textData = await response.text(); return Object.fromEntries(textData .split(/\r?\n/g) .filter(line => line.includes(',')) .map(line => line.split(',').map(value => value.trim())) ); } catch (error) { console.error(🔴 Network error while retrieving mappings from ${endpointUrl}: ${error}); process.exit(1); }}// ✅ Function to fetch query dataasync function fetchQueryData(queryLink) { try { console.log(🔄 Requesting data from: ${queryLink}); const response = await fetch(queryLink, { headers: getSessionHeaders() }); if (!response.ok) { console.error(❌ Error: API request failed with status code ${response.status}); console.error("📜 Response Content:", await response.text()); process.exit(1); } return await response.json(); } catch (error) { console.error(🔴 Network error while accessing ${queryLink}: ${error}); process.exit(1); }}// ✅ Function to map query datafunction convertMappings(rawQueryData, mappings) { return { data: rawQueryData.data.map(row => { const newRow = {}; Object.keys(row).forEach((key) => { const baseIndicator = key.split('__')[0]; newRow[mappings[baseIndicator] || key] = row[key]; }); return newRow; }) };}// ✅ Function to write data to a JSON fileasync function writeResponseToFile(data, mapIndicators, outputFilePath, mappings) { try { const finalData = mapIndicators ? convertMappings(data, mappings) : data; fs.writeFileSync(outputFilePath, JSON.stringify(finalData, null, 4)); console.log(✅ Successfully saved response to ${outputFilePath}); } catch (error) { console.error(🔴 Error writing response to file ${outputFilePath}: ${error}); process.exit(1); }}// ✅ Function to handle multiple query linksasync function requestAndWriteQueries(queryLinks, mapIndicators, outputFilePaths) { if (queryLinks.length !== outputFilePaths.length) { console.error( 🔴 Error: Number of output files must match the number of query links.\n + ➡️ ${queryLinks.length} query links provided, but ${outputFilePaths.length} output files specified. ); process.exit(1); } const mappings = await getMappings(); for (let i = 0; i < queryLinks.length; i++) { console.log(🔄 Processing query ${i + 1}/${queryLinks.length}: ${queryLinks[i]}); const rawData = await fetchQueryData(queryLinks[i]); await writeResponseToFile(rawData, mapIndicators, outputFilePaths[i], mappings); }}// ✅ Run the process(async () => { try { await requestAndWriteQueries( QUERY_LINKS, true, ['table_raw_data1.json', 'table_raw_data2.json'] ); } catch (error) { console.error(⚠️ Critical Error: ${error}); process.exit(1); }})();
You should provide the API token from step 1d as an ZEN_API_TOKEN environment variable. See comments in the beginning of the example scripts to see how to do so.
Updated on: 24/09/2025
Thank you!