HabitaXX Python SDK

栖界 HabitaXX 开放平台官方 Python SDK，基于 Python 3.9+ 编写，包含完整的请求参数和响应类型定义，并同时提供同步与异步客户端。

完整 API 文档

各接口的参数定义与完整调用方法可参考 api.md。

安装

pip install habitaxx

如需使用异步并发与 aiohttp 传输适配层：

pip install 'habitaxx[aiohttp]'

核心授权与调用流程

栖界开放平台采用 两阶段鉴权 机制：

1. 获取 Access Token：使用平台颁发的 x-api-key、项目编号 project_no 和开发者用户标识 user_id，调用 /v1/auth/token 接口换取短效 access_token。
2. 调用业务接口：将换取到的 access_token 配置为 Client 的凭证，后续请求将自动在 Header 中携带 Authorization: Bearer <access_token> 调用具体能力接口（如宠物 AI 分析、鸟类识别等）。

1. 同步调用示例

import os
from habitaxx import Habitaxx

# 步骤 1：初始化客户端并换取 Access Token
# 可以通过环境变量 HABITAXX_API_KEY 注入，也可在代码中指定
API_KEY = os.environ.get("HABITAXX_API_KEY", "qj_live_your_api_key")
BASE_URL = os.environ.get("HABITAXX_BASE_URL", "https://open-api.habitaxx.com")

client = Habitaxx(
    api_key=API_KEY,
    base_url=BASE_URL,
)

# 调用授权接口换取短效 access_token
# 对应底层 HTTP 请求：
# POST /v1/auth/token
# Headers: x-api-key: <API_KEY>
# Body: {"project_no": "...", "user_id": "..."}
token_resp = client.auth.token(
    x_api_key=API_KEY,
    body={
        "project_no": "prj_your_project_no",
        "user_id": "user_123456",
    },
)

access_token = token_resp["data"]["access_token"]
print(f"成功获取 access_token: {access_token}")

# 步骤 2：更新客户端凭据为 access_token，后续业务请求自动附带 Authorization: Bearer <access_token>
client.api_key = access_token

# 调用健康检查接口或业务接口
health_status = client.health.check()
print("健康检查结果:", health_status)

# 调用通用 AI 宠物问诊/行为预测接口示例：
# response = client.predict.new(
#     data=[[0.1, 0.2, 0.3, 0.4, 0.5, 0.6]],
#     device_id="dev_1001",
# )

2. 异步调用示例 (Async)

import asyncio
import os
from habitaxx import AsyncHabitaxx

async def main():
    API_KEY = os.environ.get("HABITAXX_API_KEY", "qj_live_your_api_key")
    
    async with AsyncHabitaxx(
        api_key=API_KEY,
        base_url="https://open-api.habitaxx.com",
    ) as client:
        # 第一阶段：换取 Access Token
        token_resp = await client.auth.token(
            x_api_key=API_KEY,
            body={
                "project_no": "prj_your_project_no",
                "user_id": "user_123456",
            },
        )
        
        access_token = token_resp["data"]["access_token"]
        
        # 第二阶段：更新凭据并请求业务 API
        client.api_key = access_token
        health = await client.health.check()
        print("异步健康检查响应:", health)

asyncio.run(main())

异常与错误处理

当网络连接失败或平台返回 4xx/5xx HTTP 错误时，SDK 会抛出对应的结构化异常类：

import habitaxx
from habitaxx import Habitaxx

client = Habitaxx(api_key="your_api_key")

# 平台接口统一返回 HTTP 200，业务状态由响应体中的 code 与 error_code 标识
# 成功响应: code == 200, message == 'success'
# 业务失败: code != 200，并附带明确的 error_code 与追踪凭证 call_id
try:
    with open("bird.jpg", "rb") as f:
        res = client.bird.detect(file=f)
    if res.get("code") == 200:
        print("识别成功:", res.get("data"))
    else:
        print(f"业务异常 [{res.get('error_code')} - {res.get('code')}]: {res.get('message')}")
        print(f"排查追踪凭证 call_id: {res.get('call_id')}")
except habitaxx.APIConnectionError as e:
    print("网络连接异常或超时:", e.__cause__)
except habitaxx.APIError as e:
    print("请求执行异常:", e)

统一业务响应与异常处理

平台对外接口统一返回 HTTP 200 响应，业务层面的鉴权失效、参数错误、配额超限或服务繁忙均在统一响应体中呈现：

• 成功响应：code: 200, message: "success", data: { ... }
• 业务失败：包含全局唯一错误码 code、语义化标识 error_code、脱敏中文提示 message 及全局链路追踪 ID call_id
• 网络与传输异常：
  • habitaxx.APIConnectionError：本地网络不通、DNS 无法解析或网关连接超时
  • habitaxx.APIError：客户端底层请求生命周期异常

高级配置

1. 超时与重试

默认情况下，客户端会针对网络连接错误以及 408 Request Timeout、429 Rate Limit、5xx 服务端错误自动执行最多 2 次指数退避重试：

from habitaxx import Habitaxx

client = Habitaxx(
    api_key="your_api_key",
    timeout=20.0,  # 请求超时时间（秒）
    max_retries=3,  # 重试次数
)

2. 自定义 Headers / 请求级覆盖

可以通过 extra_headers 或 extra_query 动态传递自定义 Header：

response = client.health.check(
    extra_headers={"X-Custom-Trace-Id": "req-999"},
)