bazhuayu-client-python — DataHub 官方 Python SDK

八爪鱼 DataHub 的官方 Python 客户端（bazhuayu-client）：发现、调用、消费 Data App。 自包含小项目，零平台侧依赖（运行时仅 httpx），只讲平台公开的 /v1 REST API。 仓库按 <品牌>-client-<语言> 命名，其他语言的客户端与之平级（如 bazhuayu-client-js）。

安装

pip install bazhuayu-client

用法

from bazhuayu_client import Client

# base_url 缺省即生产端点 https://api-datahub.bazhuayu.com,本地/预发环境按需传入
with Client(api_key="demo-key") as client:
    # 发现
    result = client.search("评论", limit=10)
    detail = client.get_app("demo/reviews-store-query")   # 或卡片里的 app_id

    # 调用（等到终态）并消费
    run = client.call("demo/partner-reviews-api", {"product": "p-9001"})
    for record in client.iterate_records(run["run_id"]):
        print(record["content"], record["rating"])

    # 盘点与对账
    for r in client.iterate_runs(run_kind="production", created_from="2026-08-01T00:00:00Z"):
        print(r["run_id"], r["state"], r["billing"])
    print(client.billing(group_by="data_app", tz_offset=480))   # 按本地日/按 App 看花在哪

    # 结果数据默认保留 90 天，要长期留存打保留标记
    client.set_dataset_retention(run["dataset_id"], retained=True)

call() 发起运行并轮询到终态；需要非阻塞语义时用 run() 拿 run_id 后自行 get_run() / cancel()。被点对点授权给自己的 App 不在市场结果里，用 search(shared_with="me") 查。全部方法与错误映射见 src/bazhuayu_client/client.py。

App 引用有两种形态，get_app() / run() / call() 及运行列表的 data_app 过滤处等价可用：<用户名>/<应用名> 两段式引用（取自卡片的 namespace 与 app_name，适合人读，发布者改名后旧引用失效）；不变标识 app_id （app_<hex>，取自卡片/详情的 app_id 字段，改名免疫）。写进配置、定时任务 等长期集成建议钉 app_id。

配置

Client() 的 base_url / api_key 缺省时回退环境变量 BAZHUAYU_BASE_URL / BAZHUAYU_API_KEY（模板见 .env.example；SDK 不自动加载 .env，由调用方 自行加载），再回退生产端点 https://api-datahub.bazhuayu.com / 匿名。

自带 token 接入（token_provider，进阶）

程序化接入的标准凭证是 API Key。已在别处持有合法 identity JWT 的调用方 （如网关转发、MCP OAuth 授权码流的登录态）可用 Client(token_provider=) 注入自管凭证，与 api_key 互斥：传返回当前有效 token 的无参 callable， 或实现 TokenProvider 协议（token(min_ttl=)，续期由实现方负责）。 对带 min_ttl 语义的 provider，提交 run 前默认要求 token 剩余有效期 ≥ 12h （不足则先向 provider 索取更新鲜的 token，可用 Client(run_token_min_ttl=) 调整）——平台入队时快照凭证、终态重放扣费，提交时 token 越新鲜，长运行 结束时扣费凭证过期的敞口越小。SDK 不提供账号密码换 token 的实现 （ROPC 已被 OAuth 2.1 废弃，平台不再支持账号密码明文接入）。

测试

uv run pytest -q    # 离线单元测试（MockTransport，不依赖服务端）