SDK 参考

一个 async with 管全生命周期：进入即认证、退出即释放、请求自动续期。

安装

pip install gausium-sdk

# 如需运行 Demo
pip install gausium-sdk[demo]

基本用法

from gausium_sdk import GausiumSDK

async with GausiumSDK(
    client_id="your_client_id",
    client_secret="your_client_secret",
    open_access_key="your_open_access_key",
) as sdk:
    # 获取机器人列表
    robots = await sdk.robot.list()

    # 查询机器人状态
    status = await sdk.robot.get_status("SN123")

    # 发送指令
    await sdk.command.send("SN123", command_type="pause")

GausiumSDK 类

GausiumSDK 是 SDK 的唯一入口。构造函数接收三个必填参数：client_id（应用 ID）、client_secret（应用密钥）、open_access_key（开放平台访问密钥）。内部创建 Auth 和 Client 实例，并通过 Python property 懒加载五个业务服务。

构造函数

class GausiumSDK:
    def __init__(self, client_id: str, client_secret: str, open_access_key: str):
        self.auth = Auth(client_id, client_secret, open_access_key)
        self.client = Client(self.auth)

服务属性（懒加载）

每个属性首次访问时才 import 对应的 Service 类并实例化，传入 Client 实例。后续访问直接返回缓存的实例。

上下文管理器

Auth 认证模块

Auth 类管理 OAuth2 全生命周期。使用自定义 grant_type urn:gaussian:params:oauth:grant-type:open-access-token 向 POST /gas/api/v1alpha1/oauth/token 获取初始 token，之后通过 refresh_token grant_type 自动续期。内部使用 asyncio.Lock 防止并发刷新。

TokenData 模型

class TokenData(BaseModel):
    token_type: str
    access_token: str
    expires_in: int          # API 返回的存活秒数
    refresh_token: str
    obtained_at: float       # 获取时刻（time.time()）

    @property
    def expires_at(self) -> float:
        """过期时刻 = obtained_at + expires_in"""
        return self.obtained_at + self.expires_in

Auth 常量与属性

Auth 方法

Client HTTP 客户端

Client 封装 httpx.AsyncClient，提供 get 和 post 两个方法。每次请求前自动调用 Auth.ensure_valid() 确保 token 有效，并通过 _headers() 注入 Authorization: Bearer <token> 和 Content-Type: application/json 请求头。

构造函数

class Client:
    def __init__(
        self,
        auth: Auth,
        base_url: str = "https://openapi.gs-robot.com",
        timeout: float = 30.0,
    ):
        self.auth = auth
        self.base_url = base_url
        self.timeout = timeout

连接池配置

Client 方法

请求流程

1. 调用 get() 或 post()
2. 内部先执行 _ensure_token() -> Auth.ensure_valid() -> 按需 login 或 refresh
3. 通过 _headers() 构造 Authorization: Bearer <token> 请求头
4. 使用 httpx.AsyncClient 发送请求
5. 调用 res.raise_for_status() 检查 HTTP 状态码
6. 返回 res.json() 解析后的字典

服务层

服务层按业务域划分为五个独立模块，每个服务接收 Client 实例并通过它发送 HTTP 请求。所有服务通过 GausiumSDK 的懒加载属性访问，首次访问时才 import 和实例化。

• RobotService（services/robot.py）-- 机器人信息服务，包含 6 个接口：列表查询、详情获取、状态查询等
• TaskService（services/task.py）-- 任务服务，包含 3 个接口：任务创建、查询、管理
• CommandService（services/command.py）-- 指令服务，包含 4 个接口：远程控制指令下发
• MapService（services/map.py）-- 地图服务，包含 5 个接口：地图上传、下载、查询
• CleaningService（services/cleaning.py）-- 清洁数据服务，包含 4 个接口：清洁报告、统计数据

# 服务通过属性访问，首次使用时自动加载
async with GausiumSDK(client_id, client_secret, open_access_key) as sdk:
    # 访问 sdk.robot 时才 import RobotService
    robots = await sdk.robot.list()

    # 访问 sdk.cleaning 时才 import CleaningService
    reports = await sdk.cleaning.get_reports("SN123")

最佳实践

架构总览

Auth 管 token 生命周期，Client 管 HTTP 请求和连接池，SDK 管服务编排和资源释放。三层各司其职，通过 async with 一键串联。

— GausiumSDK 设计原则