跳转至

客户端

SearchClient 是 AIMage SDK 的核心入口,负责与 AI-Mage Search API 的所有通信。

创建客户端

使用 search.client() 工厂函数创建客户端实例:

from aimage import search

client = search.client(
    token="your_token_here",
    service=SearchService.PROD,  # 可选,默认为生产环境
)

参数说明

参数 类型 默认值 说明
token str 必填 API 访问令牌
service SearchService | str SearchService.PROD 服务环境,可传入枚举或自定义 URL
**client_options Any 传递给底层 httpx.Client 的额外选项

服务环境

from aimage.search.settings import SearchService

# 使用预定义环境
client = search.client(token="...", service=SearchService.PROD)  # 生产环境
client = search.client(token="...", service=SearchService.DEV)  # 开发环境,仅开发者或受邀组织
client = search.client(token="...", service=SearchService.LOCAL)  # 本地环境

# 使用自定义 URL
client = search.client(token="...", service="https://custom-api.example.com/api/v1/")

SearchClient API

health() -> bool

检查 API 服务是否正常运行。

if client.health():
    print("服务正常")

如果服务异常,会抛出 RuntimeError 异常。


projects() -> Generator[Project, None, None]

获取当前用户可以访问的所有项目。返回一个生成器,支持惰性加载分页数据。

# 遍历所有项目
for project in client.projects():
    print(f"{project.name}: {project.description}")

# 转为列表
all_projects = list(client.projects())

close() -> None

关闭底层 HTTP 客户端连接。

client.close()

client 属性

访问底层的 httpx.Client 实例,用于高级定制。

http_client = client.client  # httpx.Client

上下文管理器

SearchClient 支持 with 语句,退出时自动关闭连接:

with search.client(token="...") as client:
    for project in client.projects():
        print(project.name)
# 连接已自动关闭

HTTP 配置

超时设置

默认超时配置为:

  • 总超时:30 秒
  • 连接超时:10 秒

可以通过 client_options 传递自定义超时:

import httpx

client = search.client(
    token="...",
    timeout=httpx.Timeout(60.0, connect=20.0),
)

自定义请求头

client = search.client(
    token="...",
    headers={"X-Custom-Header": "value"},
)

认证机制

SDK 使用 Bearer Token 认证。创建客户端时传入的 token 会被设置为 Authorization: Bearer <token> 请求头。

Token 管理

Token 在调用 search.client() 时会被保存到全局设置中(settings.TOKEN),后续的内部 API 调用都会自动使用该 Token。

错误处理

SDK 内部使用 httpx 进行 HTTP 请求,所有响应都会调用 raise_for_status() 检查状态码。常见的异常包括:

异常 说明
httpx.HTTPStatusError HTTP 请求返回非 2xx 状态码
httpx.ConnectError 无法连接到 API 服务
httpx.TimeoutException 请求超时
RuntimeError 健康检查失败
import httpx

try:
    client = search.client(token="invalid_token")
    projects = list(client.projects())
except httpx.HTTPStatusError as e:
    print(f"API 错误: {e.response.status_code}")
except httpx.ConnectError:
    print("无法连接到 API 服务")