Query structured data
To browse the api. See section Datasets
API endpoints
To browse the api. See section Datasets
All response formats and status codes are documented in the API Explorer
Base url: https://api.veracity.com/veracity/dw/gateway/api/v2
Authentication: Authentication and subscription key is required for each api request, see details.
Query for datasets
You can query for dataset based on name, tags, schema id or specific schema versions. Full query payload:
{
"isBaseDataset": true,
"pageIndex": 0,
"pageSize": 0,
"sortColumn": "string", // optional, valid values: Name, CreatedOn, LastModifiedOn, CreatedBy, ConnectorName
"sortDirection": "Ascending",
"datasetName": "string",
"tags": {},
"createdAfter": "string",
"createdBefore": "string",
"schemaVersionIds": [
"string"
],
"schemaId": "string" // Get datasets belongs to all schema versions of the given schema
}
In this example datasets based on give schemaId is requested. Sort column is optional and valid values: Name, CreatedOn, LastModifiedOn, CreatedBy,
Python
import requests
import json
# from step 1
veracityToken = token
mySubscriptionKey = "<api key for service account>"
workspaceId = "<workspace id>"
schemaId = "<schema id>"
base_url = "https://api.veracity.com/veracity/dw/gateway/api/v2"
endpoint = f"/workspaces/{workspaceId}/datasets/query"
url = base_url + endpoint
headers = {
"Content-Type": "application/json",
"Ocp-Apim-Subscription-Key": mySubscriptionKey,
"Authorization": f"Bearer {veracityToken}",
"User-Agent": "python-requests/2.31.0"
}
payload = {
"pageIndex": 1,
"pageSize": 10,
"sortColumn": "CreatedOn",
"sortDirection" : "Descending",
"schemaId": schemaId
}
try:
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print(f"Error AS token: {e}")
result = response.json()
C#
public async Task QueryDataSet(string veracityToken, string workspaceId, string subscriptionKey)
{
string url = $"https://api.veracity.com/veracity/dw/gateway/api/v2/workspaces/{workspaceId}/datasets/query";
var token = veracityToken;
var postData = new Dictionary<string, string>
{
{"pageIndex", "1"},
{"pageSize", "10"},
{"sortColumn", "CreatedOn"},
{"sortDirection", "Descending"}
};
string jsonString = JsonConvert.SerializeObject(postData);
HttpContent content = new StringContent(jsonString, Encoding.UTF8, "application/json");
if (_httpClient == null)
_httpClient = new HttpClient();
_httpClient.DefaultRequestHeaders.Add("Ocp-Apim-Subscription-Key", subscriptionKey);
_httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", token);
try
{
var result = await _httpClient.PostAsync(url,content);
if (result.IsSuccessStatusCode)
{
var listOfDataset = result.Content.ReadAsStringAsync().Result;
}
}
catch (Exception ex)
{
}
}
Note: The response is set of datasets that meets the query filter. Datasets that was created as "views" from other datasets (by removing columns or creating a filter) will list columns used in the view and the query filter used to create the view. In the following example this datasets use the listed columns, even if the schema version has more columns and the filter used to create the view was WEEKLY_WIND_SPEED_AVG > 8.
"queries": [
{
"column": "WEEKLY_WIND_SPEED_AVG",
"filterType": "Greater",
"filterValues": [
"8"
]
}
],
"columns": [
"week",
"year",
"weekly_wind_speed_avg",
"weekly_wind_speed_max"
],
Query for data within a dataset
The request body contains the filters fiot the query:
- queryFilters: Schemas must have query filters defined to be able to use queryFilters such as column Greater or Less than a value.
- columnFilter: Define the columns to be listed in output, if not provided all columns will be listed
Request payload
{
"pageIndex": 0,
"pageSize": 0,
"queryFilters": [
{
"column": "string",
"filterType": "List",
"filterValues": [
"string"
]
}
],
"columnFilter": [
"string"
],
"sorting": {
"column": "string",
"order": "Ascending"
}
}
The following filter will return a projection of columns listed as columnFilter. In addition return datapoints in descending order on column Timestamp:
{
"pageIndex": 1,
"pageSize": 500,
"columnFilter": ["Timestamp", "DataChannelId", "Value", "_ValueNumeric"],
"sorting": {
"column": "Timestamp",
"order": "Descending"
}
}
Python
import requests
import json
from datetime import datetime, timedelta
mySubscriptionKey = "<api key for service account>"
workspaceId = "<workspace id>"
datasetId = "<dataset id>"
# from step 1
veracityToken = token
base_url = "https://api.veracity.com/veracity/dw/gateway/api/v2"
endpoint = f"/workspaces/{workspaceId}/datasets/{datasetId}/query"
url = base_url + endpoint
headers = {
"Content-Type": "application/json",
"Ocp-Apim-Subscription-Key": mySubscriptionKey,
"Authorization": f"Bearer {veracityToken}",
"User-Agent": "python-requests/2.31.0"
}
# column filter is used to readuse number of columns in response, and query filter is used to filter on content
payload = {
"pageIndex": 1,
"pageSize": 100,
"sorting": {
"column": "TS",
"order": "Ascending"
},
"columnFilter": ["TS", "Wind"]
# to use query filter, the schema used for the dataset must support it
}
try:
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print(f"Error reading dataset: {e}")
result = response.json()
C#
public async Task QueryDataInDataSet(string veracityToken, string workspaceId, string datasetId, string subscriptionKey)
{
string url = $"https://api.veracity.com/veracity/dw/gateway/api/v2/workspaces/{workspaceId}/datasets/{datasetId}/query";
var token = veracityToken;
var sorting = new Dictionary<string, string>
{
{"column", "TS"},
{"order", "Ascending"},
};
var postData = new Dictionary<string, object>
{
{"pageIndex", "1"},
{"pageSize", "1000"},
{"sorting", sorting },
{ "columnFilter", new string[]{"TS", "Wind" } }
};
string jsonString = JsonConvert.SerializeObject(postData);
HttpContent content = new StringContent(jsonString, Encoding.UTF8, "application/json");
if (_httpClient == null)
_httpClient = new HttpClient();
_httpClient.DefaultRequestHeaders.Add("Ocp-Apim-Subscription-Key", subscriptionKey);
_httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", token);
try
{
var result = await _httpClient.PostAsync(url, content);
if (result.IsSuccessStatusCode)
{
var filtereddataset = result.Content.ReadAsStringAsync().Result;
}
}
catch (Exception ex)
{
}
}
To add filters to dataset query
You can add the following query filter operators to your query:
| Data type | Filtertypes |
|---|---|
| String | List, StringContains, Equals |
| Decimal, Float, Double, Integrer | Equals,Greater,GreaterOrEqual,Less,LessOrEqual |
| DateTime, Date only | Greater,GreaterOrEqual,Less,LessOrEqual |
| Boolean | Equals |
Note: NonFromList not supported
Note: NonFromList not supported
You can combine different filters for a column (assuming it suports them). Add query filter operators in the filterType property in your request. See an example below.
The following filter will filter on timestamp and data channel ids. In addition return datapoints in descending order:
{
"pageIndex": 1,
"pageSize": 500,
"queryFilters": [
{
"column": "Timestamp",
"filterType": "Greater",
"filterValues": [
"2024-05-01"
]
},
{
"column": "Timestamp",
"filterType": "Less",
"filterValues": [
"2024-05-03"
]
},
{
"column": "DataChannelId",
"filterType": "Equals",
"filterValues": [
"HAYS_FC-RUN1-TT01",
"Emerson_700XA-MOL_PCT_0"
]
}
],
"columnFilter": ["Timestamp", "DataChannelId", "Value", "_ValueNumeric"],
"sorting": {
"column": "Timestamp",
"order": "Descending"
}
}
var queryFilters = new[]
{
new { column = "Timestamp", filterType = "Greater", filterValues = new[] { "2024-05-01" } },
new { column = "Timestamp", filterType = "Less", filterValues = new[] { "2024-05-03" } },
new { column = "DataChannelId", filterType = "Equals", filterValues = new[] { "HAYS_FC-RUN1-TT01", "Emerson_700XA-MOL_PCT_0" } }
};
public Task<string> QueryDataInDatasetAsync(
string workspaceId, string datasetId, string sortingColumn, string sortingOrder, object[]? queryFilters = null, int pageIndex = 0, int pageSize = 100)
{
var sorting = new Dictionary<string, string>
{
{"column", sortingColumn},
{"order", sortingOrder},
};
var body = new Dictionary<string, object>
{
["pageIndex"] = pageIndex,
["pageSize"] = pageSize
};
if (!string.IsNullOrWhiteSpace(sortingColumn))
body["sorting"] = sorting ;
if (queryFilters != null && queryFilters.Length > 0)
body["queryFilters"] = queryFilters;
return CallAsync(HttpMethod.Post, $"/workspaces/{workspaceId}/datasets/{datasetId}/query", body);
}
private async Task<string> CallAsync(HttpMethod method, string path, object? body = null)
{
var token = await tokenService.GetTokenAsync();
var request = new HttpRequestMessage(method, config.BaseUrl.TrimEnd('/') + path);
request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", token);
if (!string.IsNullOrWhiteSpace(config.ApiKey))
request.Headers.Add("Ocp-Apim-Subscription-Key", config.ApiKey);
if (body is not null)
{
var json = JsonSerializer.Serialize(body, CamelCase);
request.Content = new StringContent(json, Encoding.UTF8, "application/json");
}
//Send the HTTP request and get the response
using var response = await httpClient.SendAsync(request);
var content = await response.Content.ReadAsStringAsync();
if (!response.IsSuccessStatusCode)
throw new ApiException(response.StatusCode, content);
return content;
}
Note that not all columns are filterable and different columns may support different filters. So, check which filters are supported by the columns in your dataset.
You can do that by getting the default schema version with column information and information on supported query filters. To get it, call the https://api.veracity.com/veracity/dw/gateway/api/v1/workspaces/{workspaceId}/schemas?includeDefaultSchemaVersion=true.
Below you can see a sample response that tells which columns are filterable and what filter operators they support.
Use the response as follows:
- In the schema version, find the name of the column you want to filter.
- Under the column's name, look for
isFilterable. If it istrue, then you can filter this column. - Under
filterTypes, check the supported filter operators.
Download using paging
Downloading very large result sets (e.g. 1,000,000 rows) will always require paging, regardless of platform. Neither databases nor analytics engines are designed to stream arbitrarily large tables in a single operation. For example, Databricks itself only displays up to 10,000 rows at a time in interactive queries.
public async Task QueryDataInDataSetAsync(string veracityToken, string workspaceId, string datasetId, string subscriptionKey)
{
string url = $"https://api.veracity.com/veracity/dw/gateway/api/v2/workspaces/{workspaceId}/datasets/{datasetId}/query";
var token = veracityToken;
//select one of the columns in the schema that is sortable
var sorting = new Dictionary<string, string>
{
{"column", "timestamp"},
{"order", "Ascending"},
};
//The sortColumn must be one of the fixed metadata such as Name, Description, CreatedBy, CreatedOn.
var postData = new Dictionary<string, object>
{
{"pageIndex", "1"},
{"pageSize", "100000"},
{"sortColumn", "CreatedOn"},
{"sorting", sorting }
};
if (_httpClient == null)
_httpClient = new HttpClient();
_httpClient.DefaultRequestHeaders.Add("Ocp-Apim-Subscription-Key", subscriptionKey);
_httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", token);
try
{
//start to read first page
int pageIndex = 1;
bool continuePaging = true;
while (continuePaging)
{
postData["pageIndex"] = pageIndex;
string jsonString = JsonConvert.SerializeObject(postData);
HttpContent content = new StringContent(jsonString, Encoding.UTF8, "application/json");
var result = await _httpClient.PostAsync(url, content);
if (result.IsSuccessStatusCode)
{
var filtereddataset = result.Content.ReadAsStringAsync().Result;
var payload = JsonConvert.DeserializeObject<QueryResponse<object>>(filtereddataset);
await HandlePageAsync(pageIndex, payload.Data);
continuePaging = payload.Pagination.TotalPages > pageIndex;
pageIndex++;
//verify that token is still valid and renew if needed.
if (!TokenUtils.IsJwtTimeValid(veracityToken))
{
//renew token
}
}
else
break;
}
}
catch (Exception ex)
{
}
}
private static async Task HandlePageAsync<T>(int pageIndex, List<T> data)
{
// TODO: Replace with your processing (save to storage, transform, etc.)
Console.WriteLine($"Processed page {pageIndex}, items: {data?.Count ?? 0}");
await Task.CompletedTask;
}
}
public sealed class QueryResponse<T>
{
[JsonProperty("data")]
public List<T> Data { get; set; } = new();
[JsonProperty("pagination")]
public Pagination Pagination { get; set; } = new();
}
public sealed class Pagination
{
[JsonProperty("pageIndex")]
public int PageIndex { get; set; }
[JsonProperty("pageSize")]
public int PageSize { get; set; }
[JsonProperty("totalPages")]
public int TotalPages { get; set; }
[JsonProperty("totalCount")]
public int TotalCount { get; set; }
}
Token util
using System.IdentityModel.Tokens.Jwt;
public static class TokenUtils
{
/// <summary>
/// Returns true if token exists and current time is within nbf..exp (with optional skew).
/// </summary>
public static bool IsJwtTimeValid(string jwt, TimeSpan? clockSkew = null)
{
if (string.IsNullOrWhiteSpace(jwt)) return false;
var handler = new JwtSecurityTokenHandler();
if (!handler.CanReadToken(jwt)) return false;
var token = handler.ReadJwtToken(jwt);
var now = DateTimeOffset.UtcNow;
var skew = clockSkew ?? TimeSpan.FromMinutes(2); // account for minor clock differences
// exp and nbf are optional; treat missing as valid unless present
var nbf = token.Payload.Nbf.HasValue
? DateTimeOffset.FromUnixTimeSeconds(token.Payload.Nbf.Value) - skew
: DateTimeOffset.MinValue;
var exp = token.Payload.Exp.HasValue
? DateTimeOffset.FromUnixTimeSeconds(token.Payload.Exp.Value) + skew
: DateTimeOffset.MaxValue;
return now >= nbf && now <= exp;
}
/// <summary>
/// Returns the UTC expiration if present; otherwise null.
/// </summary>
public static DateTimeOffset? GetJwtExpiry(string jwt)
{
var handler = new JwtSecurityTokenHandler();
if (!handler.CanReadToken(jwt)) return null;
var token = handler.ReadJwtToken(jwt);
return token.Payload.Exp.HasValue
? DateTimeOffset.FromUnixTimeSeconds(token.Payload.Exp.Value)
: (DateTimeOffset?)null;
}
}
Using Databricks to download huge dataset
Based on Microsoft and Databricks best practices, the following approaches are recommended when exporting large datasets:
- Export to dataset to Filestorage (ADLS / Blob).
- Write the table to Azure Blob Storage/ADLS as Parquet or CSV files.
- Then download from cloud storage using Azure Storage Explorer or CLI.
- Use Databricks SQL Query Download.
- Run a query in Databricks SQL (select * from table).
- Click "Download" to get results as CSV (limited to query result size limits).
- Download CSV (all rows up to 5GB)
- Run a databricks workflow using Veracity api.
Query using SAS URI
Step 1: Get SAS uri for a dataset
Python
import requests
import json
mySubscriptionKey = dbutils.secrets.get(scope="secrets", key="bkal_subKey")
workspaceId = "473ef07d-e914-4db4-959f-6eb737e391a6"
datasetId = "7f0e5cc3-f5c6-43cb-9414-634cb1ae7f4f"
base_url = "https://api.veracity.com/veracity/dw/gateway/api/v2"
endpoint = f"/workspaces/{workspaceId}/datasets/{datasetId}/sas?durationInMinutes=60&type=dfs"
url = base_url + endpoint
headers = {
"Content-Type": "application/json",
"Ocp-Apim-Subscription-Key": mySubscriptionKey,
"Authorization": f"Bearer {veracityToken}",
"User-Agent": "python-requests/2.31.0"
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print(f"Error: {e}")
read_sas_uri = response.json()
Step 2: Query using SAS
Python
Each dataset is stored as a deltalake folder with transaction logs in folder _delta_logs and parquest file(s). You can read this using libraries.
from azure.storage.filedatalake import DataLakeServiceClient
from urllib.parse import urlparse
# Your SAS URI from step 2
sas_url = read_sas_uri
# Parse the URL - it points to a flder
parsed_url = urlparse(sas_url)
account_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
file_system_name = parsed_url.path.strip("/").split("/")[0]
folder_path = "/".join(parsed_url.path.strip("/").split("/")[1:]) if len(parsed_url.path.strip("/").split("/")) > 1 else ""
sas_token = parsed_url.query if parsed_url.query.startswith('sv=') else ''
# Create the service client
service_client = DataLakeServiceClient(account_url=account_url, credential=sas_token)
directoryClient = service_client.get_directory_client(file_system_name, folder_path) if folder_path else None
filesystem_client = service_client.get_file_system_client(file_system=file_system_name)
# List files in the folder
paths = filesystem_client.get_paths(path=folder_path, recursive=False)
pathsLst = list(paths)
# each dataset is stored as a deltalake folder with transaction logs in in folder _delta_logs and parquest file(s). You can read this using libraries.
print(f"\nContents of folder '{folder_path}':")
for path in pathsLst:
print(f"- {path.name} (Directory: {path.is_directory})")