Tutoriales

Ejemplo de uso con Python

El tutorial comienza con un login de usuario. Con esas credenciales se realiza un flujo completo y se crea una API Key. Después se muestra el flujo limitado de una integración que utiliza esa API Key.

Dependencia

pip install requests

Cliente y autenticación

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)

El endpoint de login es /api/auth/login. La API de recursos utiliza la base URL https://veraquo.gradiant.org/api/v2.

Flujo completo con credenciales de usuario

El token de usuario permite utilizar la aplicación y administrar las API Keys.

api = VeraquoAPI()

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

file_info = api.upload_file("./fichero.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"La ejecución terminó con estado: {completed_run['status']}")

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

La respuesta de GET /runs/{run_id}/results contiene runs_results y total. Cada elemento puede incluir id, result, result_details, worker_name, result_file_urls, detail, created_at, started_at, completed_at, info, extra_details y run_id.

Si el resultado incluye archivos, usa download_result_file con el run_id, el id del resultado y cada valor de result_file_urls. Para descargar el fichero original puedes usar download_file(file_info["id"], "./original.webp").

No existe un endpoint /status separado. Consulta GET /runs/{run_id} hasta que status sea ready, ok, alert, error o unprocessed, y después recupera los resultados con GET /runs/{run_id}/results.

Uso con una API Key

Una API Key no puede utilizarse para hacer login, consultar /users/me ni crear, listar o administrar API Keys. Esas operaciones requieren un token de usuario y devuelven 401 Unauthorized cuando se intenta utilizar una API Key.

Una API Key solo puede operar sobre los recursos que se hayan creado utilizando esa misma clave:

• POST /files: subir ficheros.
• GET /files: listar los ficheros subidos con esa clave.
• GET /files/{file_id} y GET /files/{file_id}/download: consultar o descargar uno de sus ficheros.
• POST /runs: crear ejecuciones sobre sus ficheros.
• GET /runs y GET /runs/{run_id}: listar y consultar sus ejecuciones.
• GET /runs/{run_id}/results: listar los resultados de una ejecución propia.
• GET /runs/{run_id}/results/{run_result_id} y GET /runs/{run_id}/results/{run_result_id}/files/{file_url}: consultar un resultado y descargar sus archivos.

El objeto key_info del ejemplo anterior contiene la clave creada por el usuario. use_api_key cambia el cliente actual a autenticación mediante API Key y elimina el token de usuario antes de enviar la siguiente petición:

api.use_api_key(key_info["key_hash"])

my_files = api.list_files()
file_info = api.upload_file("./otro-fichero.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)

El servidor filtra los listados y las consultas para que la API Key solo pueda acceder a sus propios ficheros, ejecuciones y resultados. No reutilices un file_id o run_id creado con otra clave.

Elegir workers y esquemas

Puedes solicitar workers concretos al crear la ejecución. Si no envías workers, el sistema selecciona los workers permitidos para el usuario.

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

run_schemas permite enviar configuración por cola cuando el worker lo requiere. La estructura depende de los workers disponibles para tu usuario; consulta GET /runs/schemas para obtener el esquema de ejecución.