Tutorials

Python Usage Examples

Common Dependencies

import io
import os
import requests

Define API paths for reference

We define the REST API paths that we will use:

class Paths:
    def __init__(self):
        self.files_path = f"{HOST}/v1/files"
        self.runs_path = f"{HOST}/v1/runs"

File Upload

Definition of functions for uploading files. These methods return a file_id (or a list of file_ids) identifying the files on the platform. These will be used in other calls.

def upload_file(app_paths: Paths, api_key: str, src_path: str, dest_path: str) -> str:
    logger.info(f"Uploading file {src_path}")
    with open(src_path, "rb") as f:
        image_bytes = f.read()
    image_stream = io.BytesIO(image_bytes)

    response = requests.post(
        url=app_paths.files_path,
        headers={"X-API-Key": f"{api_key}"},
        files={"files": (dest_path, image_stream, "multipart/form-data")},
    )
    assert response.status_code == 201, response.text
    return response.json().get("file_ids")[0]

def upload_directory(app_paths: Paths, api_key: str, src_path: str, dest_path: str) -> Dict[str, str]:
    if not os.path.isdir(src_path):
        logger.error("Source path should be a directory, not a file. Returning...")
        return []
    result: Dict[str, str] = {}
    for root, dirs, files in os.walk(src_path):
        for file in files:
            file_src_path = os.path.join(root, file)
            file_dest_path = os.path.join(dest_path, file)
            result[file] = upload_file(app_paths, api_key, file_src_path, file_dest_path)
            logger.info(f"File {file} uploaded to Veraquo")
    return result

File Processing

Definition of functions to process the files.

These methods return a run_id (or a list of run_ids) identifying each execution on the platform. These will be used in other calls to check their status, obtain results, etc.

def process_file(paths: Paths, api_key: str, file_id: str):
    # check file status
    file_info = requests.get(
        url=f"{paths.files_path}/{file_id}",
        headers={"X-API-Key": f"{api_key}"},
    )
    
    file_format: str = file_info.json().get("format")
    status = file_info.json().get("state")
    if status != "ready":
        print(f"Error: file {file_id} not ready")
    
    payload = {"file_ids": [file_id]}

    processing_response = requests.post(
            url=f"{paths.runs_path}/detection/{file_format}",
            headers={"X-API-Key": f"{api_key}"},
            json=payload,
        )
    run_id = inference_response.json().get("run_ids")[0]
    return run_id

We define a function to check the processing status:

def get_processing_status(paths: Paths, api_key: str, run_id: str):
    status_response = requests.get(
        url=f"{paths.runs_path}/{run_id}/status",
        headers={"X-API-Key": f"{api_key}"},
    )
    response_json = status_response.json()
    run_status = response_json.get("status")
    return run_status

We define a function to obtain the results of a processing operation. We can obtain the information from the analysis of the file's data:

def get_processing_results(paths: Paths, api_key: str, file_id: str):
    response = requests.get(url=f"{paths.files_path}/{file_id}", headers={"X-API-Key": f"{api_key}"})
    file_details = response.json()
    detection_results = file_details["detection_results"]
    return detection_results

Deleting Files

We define a function to delete the files after processing:

def delete_files_from_storage(paths: Paths, api_key: str, file_ids: List[str]):
    for file_id in file_ids:
        delete_url = f"{paths.files_path}/{file_id}"
        response = requests.delete(url=delete_url, headers={"X-API-Key": f"{api_key}"},)

        if response.status_code == 204:
            print(f"File {file_name} deleted from storage")
        else:
            print(f"Problem deleting file {file_name}: {response.text}")