Passer au contenu principal
Le Python SDK nécessite Python 3.9+.

Installation

pip install avala

Démarrage rapide

from avala import Client

client = Client(api_key="your-api-key")

# List all datasets
datasets = client.datasets.list()
for dataset in datasets:
    print(dataset.name, dataset.uid)

Créer un compte

La fonction signup crée un nouveau compte Avala et retourne une clé API. Elle ne nécessite pas d’authentification.
from avala import signup

result = signup(
    email="dev@acme.com",
    password="SecurePass123!",
    first_name="Jane",       # optional
    last_name="Doe",         # optional
)

print(f"User: {result.user.email}")
print(f"API Key: {result.api_key}")
Une variante asynchrone est également disponible :
from avala import async_signup

result = await async_signup(email="dev@acme.com", password="SecurePass123!")

Authentification

Le SDK s’authentifie en utilisant votre clé API Avala, qui est envoyée via l’en-tête X-Avala-Api-Key à chaque requête. Vous pouvez fournir la clé directement ou laisser le SDK la lire depuis l’environnement. Option 1 : Passer la clé directement
from avala import Client

client = Client(api_key="your-api-key")
Option 2 : Utiliser une variable d’environnement
export AVALA_API_KEY="your-api-key"
from avala import Client

# Automatically reads AVALA_API_KEY from the environment
client = Client()

Support asynchrone

Le SDK inclut un client entièrement asynchrone construit sur httpx. Utilisez AsyncClient pour des E/S non bloquantes dans les applications asynchrones.
import asyncio
from avala import AsyncClient

async def main():
    client = AsyncClient(api_key="your-api-key")

    datasets = await client.datasets.list()
    for dataset in datasets:
        print(dataset.name)

    # Always close the client when done, or use it as a context manager
    await client.close()

asyncio.run(main())
Utilisation du gestionnaire de contexte asynchrone :
import asyncio
from avala import AsyncClient

async def main():
    async with AsyncClient() as client:
        datasets = await client.datasets.list()
        for dataset in datasets:
            print(dataset.name)

asyncio.run(main())

Travailler avec les jeux de données

Le Python SDK est actuellement en lecture seule pour les jeux de données — vous pouvez lister, récupérer et parcourir les éléments mais pas créer ou supprimer des jeux de données. Utilisez la REST API pour les téléchargements et les mutations.

Lister les jeux de données

datasets = client.datasets.list()

for dataset in datasets:
    print(f"{dataset.name} ({dataset.uid})")
    print(f"  Items: {dataset.item_count}")
    print(f"  Created: {dataset.created_at}")

Obtenir un jeu de données

dataset = client.datasets.get("550e8400-e29b-41d4-a716-446655440000")

print(dataset.name)
print(dataset.slug)
print(dataset.item_count)

Travailler avec les projets

Lister les projets

projects = client.projects.list()

for project in projects:
    print(f"{project.name} ({project.uid})")
    print(f"  Status: {project.status}")
    print(f"  Created: {project.created_at}")

Obtenir un projet

project = client.projects.get("770a9600-a40d-63f6-c938-668877660000")

print(project.name)
print(project.status)

Travailler avec les tâches

Lister les tâches

tasks = client.tasks.list(project="770a9600-a40d-63f6-c938-668877660000", status="pending")

for task in tasks:
    print(f"{task.uid}{task.name} ({task.status})")

Obtenir une tâche

task = client.tasks.get("990c1800-b62f-85a8-e150-880099880000")

print(task.name)
print(task.status)

Travailler avec les exports

Le Python SDK supporte les opérations CRUD complètes pour les agents, webhooks, configurations de stockage, fournisseurs d’inférence, objectifs de qualité, configuration de consensus et organisations. Les jeux de données et projets sont actuellement en lecture seule — utilisez la REST API pour les mutations sur ces ressources.

Créer un export

export = client.exports.create(project="770a9600-a40d-63f6-c938-668877660000")

print(f"Export started: {export.uid}")
print(f"Status: {export.status}")

Attendre la fin

import time

export = client.exports.create(project="770a9600-a40d-63f6-c938-668877660000")

while export.status != "completed":
    time.sleep(2)
    export = client.exports.get(export.uid)
    print(f"Status: {export.status}")

print(f"Download: {export.download_url}")

Travailler avec les organisations

Lister les organisations

orgs = client.organizations.list()
for org in orgs:
    print(f"{org.name} ({org.slug})")

Créer une organisation

org = client.organizations.create(name="My Team", visibility="private", industry="technology")
print(f"Created: {org.name} ({org.uid})")

Gérer les membres

members = client.organizations.list_members("my-team")
for member in members:
    print(f"{member.full_name} - {member.role}")

Travailler avec les slices

Lister les slices

slices = client.slices.list("my-org")
for s in slices:
    print(f"{s.name}: {s.item_count} items")

Parcourir les éléments d’un jeu de données

Lister les éléments d’un jeu de données

items = client.datasets.list_items("my-org", "my-dataset")
for item in items:
    print(f"{item.uid}: {item.key}")

Lister les séquences

sequences = client.datasets.list_sequences("my-org", "my-dataset")
for seq in sequences:
    print(f"{seq.uid}: {seq.key} ({seq.number_of_frames} frames)")

Annotations de type

Le SDK est entièrement typé. Tous les objets de réponse sont des modèles Pydantic avec des annotations de type complètes, vous offrant l’autocomplétion et la vérification de type dès l’installation.
from avala.types import Dataset, Project, Export, Task

def process_dataset(dataset: Dataset) -> None:
    print(dataset.name)        # str
    print(dataset.uid)         # str
    print(dataset.item_count)  # int
    print(dataset.created_at)  # Optional[datetime]

def process_task(task: Task) -> None:
    print(task.uid)            # str
    print(task.name)           # Optional[str]
    print(task.status)         # Optional[str]
    print(task.project)        # Optional[str]

Gestion des erreurs

Le SDK lève des exceptions typées pour que vous puissiez gérer précisément les différents modes de défaillance.
from avala import Client
from avala.errors import (
    AvalaError,
    NotFoundError,
    RateLimitError,
    ValidationError,
)

client = Client()

try:
    dataset = client.datasets.get("nonexistent")
except NotFoundError as e:
    print(f"Dataset not found: {e.message}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds.")
except ValidationError as e:
    print(f"Invalid request: {e.message}")
    for detail in e.details:
        print(f"  - {detail}")
except AvalaError as e:
    # Catch-all for any other Avala API error
    print(f"API error ({e.status_code}): {e.message}")
ExceptionDescription
AvalaErrorException de base pour toutes les erreurs de l’API Avala.
AuthenticationErrorClé API invalide ou manquante (HTTP 401).
NotFoundErrorLa ressource demandée n’existe pas (HTTP 404).
RateLimitErrorVous avez dépassé la limite de débit de l’API (HTTP 429). Inclut un attribut retry_after.
ValidationErrorLa charge utile de la requête a échoué la validation (HTTP 400/422). Inclut un attribut details avec les erreurs par champ.
ServerErrorLe serveur a retourné une erreur interne (HTTP 5xx).

Pagination

Les méthodes de liste retournent un objet CursorPage. Vous pouvez itérer directement sur les éléments ou contrôler la pagination manuellement.
# Iterate through items on the current page
for dataset in client.datasets.list():
    print(dataset.name)

# Manual pagination — control page size and access cursor
page = client.datasets.list(limit=10)
for dataset in page.items:
    print(dataset.name)

# Fetch the next page
if page.has_more:
    next_page = client.datasets.list(limit=10, cursor=page.next_cursor)

Gestion de flotte

La gestion de flotte est en aperçu. Les APIs décrites ici peuvent changer.
L’espace de noms fleet fournit l’accès au registre d’appareils, aux enregistrements, événements, règles et alertes.
# List online devices
devices = client.fleet.devices.list(status="online")

# Create a timeline event on a recording
event = client.fleet.events.create(
    recording_id="rec_abc123",
    timestamp="2026-01-15T10:30:00Z",
    type="anomaly",
    label="Gripper force spike",
    metadata={"force_n": 45.2}
)

# Create a recording rule
rule = client.fleet.rules.create(
    name="High Latency Alert",
    condition={"type": "threshold", "topic": "/diagnostics/latency", "field": "data.value", "operator": "gt", "value": 100},
    actions=[{"type": "tag", "value": "high-latency"}, {"type": "notify", "channel_id": "ch_your_channel_id"}]
)
Consultez le guide Tableau de bord de flotte pour des exemples complets.

Configuration

Vous pouvez personnaliser le comportement du client lors de l’initialisation.
from avala import Client

client = Client(
    api_key="your-api-key",
    base_url="https://api.avala.ai/api/v1",  # Default
    timeout=60,        # Request timeout in seconds (default: 30)
    max_retries=3,     # Number of retries on transient errors (default: 2)
)
ParamètreTypeDéfautDescription
api_keystrVariable d’env AVALA_API_KEYVotre clé API Avala.
base_urlstrhttps://api.avala.ai/api/v1L’URL de base de l’API.
timeoutfloat30Délai d’attente de requête en secondes.
max_retriesint2Nombre de réessais automatiques en cas d’erreurs transitoires (5xx, délais d’attente).