> ## Documentation Index
> Fetch the complete documentation index at: https://avala.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK

> Avala Python SDK 完整指南

<Info>Python SDK 需要 **Python 3.9+**。</Info>

## 安装

```bash theme={null}
pip install avala
```

## 快速开始

```python theme={null}
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)
```

## 创建账户

`signup` 函数创建新的 Avala 账户并返回 API 密钥。不需要认证。

```python theme={null}
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}")
```

也提供异步变体：

```python theme={null}
from avala import async_signup

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

## 认证

SDK 使用 Avala API 密钥进行认证，通过 `X-Avala-Api-Key` 头在每个请求中发送。

您可以直接提供密钥或让 SDK 从环境变量读取。

**方式 1：直接传递密钥**

```python theme={null}
from avala import Client

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

**方式 2：使用环境变量**

```bash theme={null}
export AVALA_API_KEY="your-api-key"
```

```python theme={null}
from avala import Client

# 自动从环境变量读取 AVALA_API_KEY
client = Client()
```

## 异步支持

SDK 附带基于 `httpx` 构建的完全异步客户端。在异步应用中使用 `AsyncClient` 进行非阻塞 I/O。

```python theme={null}
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())
```

**使用异步上下文管理器：**

```python theme={null}
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())
```

## 使用数据集

<Note>
  Python SDK 目前对数据集是**只读的**——您可以列出、检索和浏览项，但不能创建或删除数据集。上传和修改操作请使用 [REST API](/docs/sdks/rest-api#file-uploads)。
</Note>

### 列出数据集

```python theme={null}
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}")
```

### 获取数据集

```python theme={null}
dataset = client.datasets.get("550e8400-e29b-41d4-a716-446655440000")

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

## 使用项目

### 列出项目

```python theme={null}
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}")
```

### 获取项目

```python theme={null}
project = client.projects.get("770a9600-a40d-63f6-c938-668877660000")

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

## 使用任务

### 列出任务

```python theme={null}
tasks = client.tasks.list(project="770a9600-a40d-63f6-c938-668877660000", status="pending")

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

### 获取任务

```python theme={null}
task = client.tasks.get("990c1800-b62f-85a8-e150-880099880000")

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

## 使用导出

<Note>
  Python SDK 支持对代理、Webhook、存储配置、推理提供者、质量目标、共识配置和组织的完整 CRUD 操作。数据集和项目目前是只读的——修改操作请使用 [REST API](/docs/sdks/rest-api)。
</Note>

### 创建导出

```python theme={null}
export = client.exports.create(project="770a9600-a40d-63f6-c938-668877660000")

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

### 轮询完成状态

```python theme={null}
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}")
```

## 使用组织

### 列出组织

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

### 创建组织

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

### 管理成员

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

## 使用切片

### 列出切片

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

## 浏览数据集项

### 列出数据集中的项

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

### 列出序列

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

## 类型提示

SDK 完全类型化。所有响应对象都是具有完整类型标注的 Pydantic 模型，开箱即用地提供自动补全和类型检查。

```python theme={null}
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]
```

## 错误处理

SDK 抛出类型化异常，让您精确处理不同的失败模式。

```python theme={null}
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}")
```

| 异常                    | 描述                                            |
| --------------------- | --------------------------------------------- |
| `AvalaError`          | 所有 Avala API 错误的基础异常。                         |
| `AuthenticationError` | API 密钥无效或缺失（HTTP 401）。                        |
| `NotFoundError`       | 请求的资源不存在（HTTP 404）。                           |
| `RateLimitError`      | 已超过 API 速率限制（HTTP 429）。包含 `retry_after` 属性。   |
| `ValidationError`     | 请求载荷验证失败（HTTP 400/422）。包含 `details` 属性和字段级错误。 |
| `ServerError`         | 服务器返回内部错误（HTTP 5xx）。                          |

## 分页

列表方法返回 `CursorPage` 对象。您可以直接遍历项或手动控制分页。

```python theme={null}
# 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)
```

## 车队管理

<Info>
  车队管理处于预览阶段。此处描述的 API 可能会更改。
</Info>

`fleet` 命名空间提供对设备注册、录制、事件、规则和警报的访问。

```python theme={null}
# 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"}]
)
```

完整示例请参阅[车队仪表板](/docs/visualization/fleet/fleet-dashboard)指南。

## 配置

您可以在初始化时自定义客户端行为。

```python theme={null}
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)
)
```

| 参数            | 类型      | 默认值                           | 描述                    |
| ------------- | ------- | ----------------------------- | --------------------- |
| `api_key`     | `str`   | `AVALA_API_KEY` 环境变量          | 您的 Avala API 密钥。      |
| `base_url`    | `str`   | `https://api.avala.ai/api/v1` | API 基础 URL。           |
| `timeout`     | `float` | `30`                          | 请求超时时间（秒）。            |
| `max_retries` | `int`   | `2`                           | 暂时性故障（5xx、超时）的自动重试次数。 |