HabitaXX Python SDK

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

完整 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")

try:
    client.health.check()
except habitaxx.APIConnectionError as e:
    print("网络连接异常:", e.__cause__)
except habitaxx.RateLimitError as e:
    print("触发平台流控频率限制 (429)")
except habitaxx.APIStatusError as e:
    print(f"API 响应非 200 错误码: {e.status_code}")
    print(e.response)

常见异常层级关系

• habitaxx.APIError: 所有 API 异常基类
  • habitaxx.APIConnectionError: 网络连接不可达或超时
  • habitaxx.APIStatusError: 平台返回错误状态码
    • habitaxx.BadRequestError (400)
    • habitaxx.AuthenticationError (401)
    • habitaxx.PermissionDeniedError (403)
    • habitaxx.NotFoundError (404)
    • habitaxx.RateLimitError (429)
    • habitaxx.InternalServerError (500+)

高级配置

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"},
)