Tutorials

Python usage example

This tutorial starts with a user login. Using those credentials, it completes a full workflow and creates an API Key. It then shows the limited workflow for an integration that uses that API Key.

Dependency

pip install requests

Client and authentication

import time
from pathlib import Path

import requests


class VeraquoAPI:
    BASE_URL = "https://veraquo.gradiant.org/api/v2"
    AUTH_URL = "https://veraquo.gradiant.org/api/auth/login"

    def __init__(self):
        self.base_url = self.BASE_URL
        self.session = requests.Session()

    def login(self, username: str, password: str) -> dict:
        response = self.session.post(
            self.AUTH_URL,
            data={
                "grant_type": "password",
                "username": username,
                "password": password,
            },
            headers={"user-agent": "custom"},
        )
        response.raise_for_status()
        login_info = response.json()
        self.session.headers.pop("X-API-Key", None)
        self.session.headers.update(
            {"Authorization": f"Bearer {login_info['access_token']}"}
        )
        return login_info

    def create_api_key(self, name: str, expiration_days: int | None = None) -> dict:
        response = self.session.post(
            f"{self.base_url}/api-keys",
            json={"name": name, "expiration_days": expiration_days},
        )
        response.raise_for_status()
        return response.json()

    def use_api_key(self, api_key: str) -> None:
        self.session.headers.pop("Authorization", None)
        self.session.headers.update({"X-API-Key": api_key})

    def upload_file(self, path: str) -> dict:
        file_path = Path(path)
        with file_path.open("rb") as file_handle:
            response = self.session.post(
                f"{self.base_url}/files",
                files={"file": (file_path.name, file_handle)},
            )
        response.raise_for_status()
        return response.json()

    def list_files(self) -> dict:
        response = self.session.get(f"{self.base_url}/files")
        response.raise_for_status()
        return response.json()

    def get_file(self, file_id: str) -> dict:
        response = self.session.get(f"{self.base_url}/files/{file_id}")
        response.raise_for_status()
        return response.json()

    def create_run(self, file_id: str, workers: list[str] | None = None) -> dict:
        response = self.session.post(
            f"{self.base_url}/runs",
            json={
                "file_id": file_id,
                "run_schemas": None,
                "workers": workers,
            },
        )
        response.raise_for_status()
        return response.json()

    def list_runs(self) -> dict:
        response = self.session.get(f"{self.base_url}/runs")
        response.raise_for_status()
        return response.json()

    def get_run(self, run_id: str) -> dict:
        response = self.session.get(f"{self.base_url}/runs/{run_id}")
        response.raise_for_status()
        return response.json()

    def wait_for_run(self, run_id: str, interval_seconds: int = 5) -> dict:
        while True:
            run = self.get_run(run_id)
            if run["status"] in {"ready", "ok", "alert", "error", "unprocessed"}:
                return run
            time.sleep(interval_seconds)

    def get_results(self, run_id: str, result: str | None = None) -> dict:
        params = {"result": result} if result is not None else None
        response = self.session.get(
            f"{self.base_url}/runs/{run_id}/results",
            params=params,
        )
        response.raise_for_status()
        return response.json()

    def get_result(self, run_id: str, result_id: str) -> dict:
        response = self.session.get(
            f"{self.base_url}/runs/{run_id}/results/{result_id}"
        )
        response.raise_for_status()
        return response.json()

    def download_file(self, file_id: str, destination: str) -> None:
        response = self.session.get(f"{self.base_url}/files/{file_id}/download")
        response.raise_for_status()
        Path(destination).write_bytes(response.content)

    def download_result_file(
        self, run_id: str, result_id: str, file_url: str, destination: str
    ) -> None:
        response = self.session.get(
            f"{self.base_url}/runs/{run_id}/results/{result_id}/files/{file_url}"
        )
        response.raise_for_status()
        Path(destination).write_bytes(response.content)

The login endpoint is /api/auth/login. The resource API uses the base URL https://veraquo.gradiant.org/api/v2.

Complete flow with user credentials

The user token allows you to use the application and manage API Keys.

api = VeraquoAPI()

api.login("{USERNAME}", "{PASSWORD}")

file_info = api.upload_file("./file.webp")
run_info = api.create_run(file_info["id"])
completed_run = api.wait_for_run(run_info["id"])

if completed_run["status"] in {"ready", "ok", "alert"}:
    results = api.get_results(completed_run["id"])
    print(results)
else:
    print(f"The run ended with status: {completed_run['status']}")

key_info = api.create_api_key("my-integration", expiration_days=90)
print(key_info["key_hash"])

The GET /runs/{run_id}/results response contains runs_results and total. Each item may include id, result, result_details, worker_name, result_file_urls, detail, created_at, started_at, completed_at, info, extra_details, and run_id.

If a result includes files, use download_result_file with the run_id, result id, and each value in result_file_urls. To download the original file, use download_file(file_info["id"], "./original.webp").

There is no separate /status endpoint. Query GET /runs/{run_id} until status is ready, ok, alert, error, or unprocessed, then retrieve the results with GET /runs/{run_id}/results.

Using an API Key

An API Key cannot be used to log in, query /users/me, or create, list, or manage API Keys. Those operations require a user token and return 401 Unauthorized when attempted with an API Key.

An API Key can only operate on resources created using that same key:

POST /files: upload files.
GET /files: list files uploaded with that key.
GET /files/{file_id} and GET /files/{file_id}/download: query or download one of its files.
POST /runs: create runs on its files.
GET /runs and GET /runs/{run_id}: list and query its runs.
GET /runs/{run_id}/results: list the results of one of its runs.
GET /runs/{run_id}/results/{run_result_id} and GET /runs/{run_id}/results/{run_result_id}/files/{file_url}: query a result and download its files.

The key_info object in the previous example contains the key created by the user. use_api_key switches the current client to API-key authentication and removes the user token before the next request:

api.use_api_key(key_info["key_hash"])

my_files = api.list_files()
file_info = api.upload_file("./another-file.webp")
run_info = api.create_run(file_info["id"])
completed_run = api.wait_for_run(run_info["id"])

if completed_run["status"] in {"ready", "ok", "alert"}:
    my_runs = api.list_runs()
    results = api.get_results(completed_run["id"])
    print(my_files)
    print(my_runs)
    print(results)

The server filters lists and queries so that an API Key can only access its own files, runs, and results. Do not reuse a file_id or run_id created with another key.

Choosing workers and schemas

You can request specific workers when creating a run. If you do not send workers, the system selects the workers allowed for the user.

run_info = api.create_run(
    file_info["id"],
    workers=["gradiant.org.syp.veraquo.heatmap.generation"],
)

run_schemas lets you send queue-specific configuration when the worker requires it. The structure depends on the workers available to your user; query GET /runs/schemas to obtain the run schema.

Python usage example​

Dependency​

Client and authentication​

Complete flow with user credentials​

Using an API Key​

Choosing workers and schemas​