Examples

import requests
import os
 
API_KEY = os.environ.get("AVALA_API_KEY")
BASE_URL = "https://api.avala.ai/api/v1"
 
headers = {"X-Avala-Api-Key": API_KEY}

def list_all_datasets():
    """Fetch all datasets with pagination."""
    all_datasets = []
    url = f"{BASE_URL}/datasets/"
 
    while url:
        response = requests.get(url, headers=headers)
        data = response.json()
        all_datasets.extend(data["results"])
        url = data.get("next")
 
    return all_datasets
 
datasets = list_all_datasets()
for ds in datasets:
    print(f"{ds['name']} ({ds['item_count']} items)")

def create_dataset(name, data_type="image", visibility="private"):
    response = requests.post(
        f"{BASE_URL}/datasets/",
        headers=headers,
        json={
            "name": name,
            "data_type": data_type,
            "visibility": visibility
        }
    )
    return response.json()
 
dataset = create_dataset("My Training Data")
print(f"Created: {dataset['slug']}")

def get_dataset_with_items(owner, slug, limit=100):
    # Get dataset details
    dataset = requests.get(
        f"{BASE_URL}/datasets/{owner}/{slug}/",
        headers=headers
    ).json()
 
    # Get items
    items = requests.get(
        f"{BASE_URL}/datasets/{owner}/{slug}/items/",
        headers=headers,
        params={"limit": limit}
    ).json()
 
    return {
        "dataset": dataset,
        "items": items["results"]
    }

def list_active_projects():
    response = requests.get(
        f"{BASE_URL}/projects/",
        headers=headers,
        params={"status": "ACTIVE"}
    )
    return response.json()["results"]
 
projects = list_active_projects()
for proj in projects:
    print(f"{proj['name']} - {proj['task_type']}")

def get_project_progress(project_uid):
    response = requests.get(
        f"{BASE_URL}/projects/{project_uid}/metrics/",
        headers=headers
    )
    metrics = response.json()
 
    completion = (metrics["completed_tasks"] / metrics["total_tasks"]) * 100
    print(f"Progress: {completion:.1f}%")
    print(f"Acceptance rate: {metrics['acceptance_rate'] * 100:.1f}%")
 
    return metrics

def pause_project(project_uid):
    response = requests.post(
        f"{BASE_URL}/projects/{project_uid}/pause/",
        headers=headers
    )
    return response.json()
 
def resume_project(project_uid):
    response = requests.post(
        f"{BASE_URL}/projects/{project_uid}/resume/",
        headers=headers
    )
    return response.json()

import time
 
def export_project(project_uid, format="coco"):
    # Create export
    response = requests.post(
        f"{BASE_URL}/projects/{project_uid}/exports/",
        headers=headers,
        json={"format": format}
    )
    export = response.json()
    export_uid = export["uid"]
 
    # Poll until complete
    while True:
        status_response = requests.get(
            f"{BASE_URL}/exports/{export_uid}/",
            headers=headers
        )
        export = status_response.json()
 
        if export["status"] == "completed":
            return export["download_url"]
        elif export["status"] == "failed":
            raise Exception("Export failed")
 
        time.sleep(5)
 
# Usage
download_url = export_project("proj-uuid-001", "coco")
print(f"Download: {download_url}")

def export_completed_annotations(project_uid, labels=None):
    filters = {"status": "completed"}
    if labels:
        filters["labels"] = labels
 
    response = requests.post(
        f"{BASE_URL}/projects/{project_uid}/exports/",
        headers=headers,
        json={
            "format": "coco",
            "filters": filters
        }
    )
    return response.json()
 
# Export only car and truck annotations
export = export_completed_annotations(
    "proj-uuid-001",
    labels=["car", "truck"]
)

def list_org_members(org_slug):
    response = requests.get(
        f"{BASE_URL}/organizations/{org_slug}/members/",
        headers=headers
    )
    return response.json()["results"]
 
members = list_org_members("acme-ai")
for member in members:
    print(f"{member['user']['username']} - {member['role']}")

def invite_member(org_slug, email, role="member"):
    response = requests.post(
        f"{BASE_URL}/organizations/{org_slug}/invitations/",
        headers=headers,
        json={
            "email": email,
            "role": role
        }
    )
    return response.json()

import time
from requests.exceptions import RequestException
 
def api_call_with_retry(method, url, max_retries=3, **kwargs):
    """Make API call with retry logic for rate limits and errors."""
    for attempt in range(max_retries):
        try:
            response = requests.request(method, url, headers=headers, **kwargs)
 
            # Handle rate limits
            if response.status_code == 429:
                wait_time = 2 ** attempt
                print(f"Rate limited, waiting {wait_time}s...")
                time.sleep(wait_time)
                continue
 
            response.raise_for_status()
            return response.json()
 
        except RequestException as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)
 
    raise Exception("Max retries exceeded")
 
# Usage
datasets = api_call_with_retry("GET", f"{BASE_URL}/datasets/")