Call Data Service API

数据服务 API 调用

1. Scenario Description

应用开发者调用 Dataphin 数据服务已发布并授权的 API，支持同步、异步和流式（SSE）三种调用模式。调用脚本内置 HMAC-SHA256 签名认证，不依赖官方 SDK，也不需要任何第三方库。

业务流程：

确认调用信息 → 同步调用 API → （可选）异步调用 → 验证调用成功

资源拓扑：

数据服务网关
├── 阿里云 API 网关模式（推荐脚本调用）
│   ├── HMAC-SHA256 签名（脚本内置）
│   └── AppKey/AppSecret → 脚本自动处理
├── 内置网关模式
│   ├── 参数认证（appkey/appsecret 请求参数）
│   └── AppKey/AppSecret → 请求参数
├── 同步调用（即时返回）
├── 异步调用
│   ├── 提交任务 → jobId
│   ├── 脚本自动轮询状态
│   └── 获取结果
└── 流式调用（SSE）
    └── 实时返回数据片段

前置条件：

• 至少一个 API 已发布到目标环境（S1 create-and-publish-api 产出）
• 应用已创建并获授权（S2 manage-app-and-bindauth 产出）
• 已获取 AppKey 和 AppSecret
• 已确认网关地址和 API 调用路径
• Python >= 3.9（仅需标准库）

与 S1/S2 的衔接：

• S1 产出 ApiId + API 路径（通过 get-data-service-api-document 查询）
• S2 产出 AppKey / AppSecret，本 Skill 消费这些凭证发起调用

与 S1/S2 的本质差异：

2. Installation

无需安装任何 SDK 或第三方库。

官方 Python/Java SDK 亦可用（控制台「数据服务 → 应用管理 → 调用说明 → SDK 下载」），但不是本 Skill 的前置条件；脚本签名逻辑与官方 SDK v5.5.0 逐字节一致。

3. Environment Variables

安全提示：不要将 AppKey/AppSecret 硬编码在代码中，务必使用环境变量。

4. Authentication

Pre-check: Credentials Required

# 检查 Python 环境（脚本要求 >= 3.9，无需第三方库）
python3 --version

# 确认应用凭证已获取（来自 S2 manage-app-and-bindauth 产出）
# appKey: 应用 AppKey
# appSecret: 应用 AppSecret
# host: 数据服务网关地址（从控制台"网络配置"获取）

凭证不可打印：任何时候不得将 AppKey/AppSecret 输出到终端或日志。

认证方式说明：

本 Skill 不使用 RAM 凭证，使用数据服务应用凭证（AppKey/AppSecret），签名由调用脚本内置完成。

如何判断当前网关类型：登录 Dataphin 控制台 → 数据服务 → 服务管理 → 网络配置。

5. App Authentication

本 Skill 不使用 RAM 认证，改用数据服务应用认证。

认证方式一：脚本自动签名（阿里云 API 网关）——推荐

scripts/call-data-service-api.py 内置 HMAC-SHA256 签名，用户只需提供 AppKey/AppSecret，脚本自动完成全部签名流程（nonce/timestamp 生成、签名串构造、签名计算、Header 设置）。签名串规范见 Python 调用模板 §1，App 认证说明见 App 认证参考。

认证方式二：参数认证（内置网关）

内置网关模式下，appkey 和 appsecret 作为 API 的公共参数传入。详见 App 认证参考。

AppKey/AppSecret 获取

AppKey/AppSecret 由 S2 manage-app-and-bindauth 创建应用时获取。也可通过以下 CLI 命令查询：

# 查看应用详情（需要是应用成员，或 SuperAdmin 权限）
aliyun dataphin-public get-data-service-app \
  --op-tenant-id <tenantId> --app-id <appId> \
  --profile <profile> --endpoint <endpoint>

# 列出所有应用（查看 AppId，需应用成员才能获取 AppKey/AppSecret）
aliyun dataphin-public list-data-service-apps \
  --op-tenant-id <tenantId> \
  --list-query PageNo=1 PageSize=20 \
  --profile <profile> --endpoint <endpoint>

常见认证错误

Permission Failure Handling

若遇到权限错误（HTTP 403 或错误码含 AppUnauthorized/Forbidden），请：

1. 确认应用已通过 S2 manage-app-and-bindauth 获得目标 API 的授权
2. 确认 stage 参数与 API 发布环境匹配（RELEASE = 生产，PRE = 开发）
3. 确认当前用户是应用成员（IsMember: true）
4. 联系项目管理员授权

详见 App 认证参考。

6. Parameter Confirmation

IMPORTANT: Parameter Confirmation 执行前必须确认以下业务参数：

host 获取说明：

网关地址从 Dataphin 控制台获取：数据服务 → 服务管理 → 网络配置。独立部署环境常见命名为 dataphin-dataservice.<租户基础域名>（反代 canonical 常落 dataphin-os-gateway.*）。该网关与管理面 OpenAPI 端点 dataphin-openapi.* 不是同一域名，也不在任何 OpenAPI 返回里。若无法直接拿到，可按命名规律探测确认（curl POST /list/{apiId} 返回 DPN-OLTP-* 即命中），详见 调用前置发现 §网关 host 发现。VPC/私有化部署中该域名可能不对外暴露公网地址，需联系运维确认。

⚠️ 参数获取暗坑（逆向参数时必看）：

• 应用名不唯一：同租户可能多个同名应用，必须用 AppKey 唯一确定，不靠名字。
• AppKey 是字符串（如 "200000326"），比较用 str()。
• list-authorized-data-service-api-details 用 AppKeyStr（字符串），AppKey（整型）已弃用。
• returnFields 只能取已授权字段（步骤 C 产出），传未授权字段会报错。 完整清单见 调用前置发现 §暗坑清单。

**API 调用路径构造：**正确格式为 /{methodType}/{apiId}?appKey={appKey}&env={env}，不是 /api/<GroupId>/<ApiName>。methodType 有 5 种：list（列表查询）、get（单条查询）、create/update/delete（DML）。

⚠️ methodType 由 API 发布时的操作类型决定，不能仅凭 IsPagedQuery 推断：IsPagedQuery 只表示「是否分页」，list 与 get 都可能为 true（实测 GetCustomer 的 IsPagedQuery=true 但它是 get）。应按 API 的操作语义/命名判断（Get*/单条 → get；List*/Bulk*/列表 → list；Create*→create；Update*→update；Delete*→delete）。methodType 猜错 → 网关返回 403 The request api path /xxx/{apiId} not bind app {appKey}，据此换正确动词重试。

示例：http://<YOUR_GATEWAY_ENDPOINT>/list/10083?appKey=200000008&env=PROD

7. Observability

本子 Skill 的 session-id 继承自父 Skill alibabacloud-dataphin-skills，不重新生成。

调用脚本通过标准 user-agent 请求头标记（脚本读取环境变量 SKILL_SESSION_ID 自动注入）：

user-agent: AlibabaCloud-Agent-Skills/call-data-service-api/{SESSION_ID}

⚠️ 不要用 X-Ca-User-Agent 之类的 x-ca-* 头承载可观测标记：x-ca-* 前缀会被纳入签名串，多一个头就要同步进 x-ca-signature-headers，否则触发 SignatureDoesNotMatch。用普通 user-agent（不参与签名）最稳妥。

其中 {SESSION_ID} 为父 Skill 生成的 32 字符小写十六进制字符串。

8. Core Workflow

步骤 0：环境前置检查

在执行任何调用之前，确认 Python 版本与凭证环境变量就绪。无需下载或安装 SDK。

# 1. Python >= 3.9（脚本仅用标准库，无需 pip install）
python3 --version

# 2. 确认脚本存在（相对于本 Skill 目录）
ls ./scripts/call-data-service-api.py

# 3. 确认凭证环境变量（值不打印）
: "${DATAPHIN_APP_KEY:?未设置}" "${DATAPHIN_APP_SECRET:?未设置}" "${DATAPHIN_GATEWAY_HOST:?未设置}"

步骤 0.5：零参数发现（仅有「应用名 + API 名」时）

若用户只给了应用名 + 要调的 API 名（例：让应用「客户管理」查询客户列表），而没有直接给 appKey/appSecret/apiId/host，先走一条管理面反查链补齐四要素——否则会卡在找参数上（这是本 Skill 最常见的耗时点）：

host（网关地址）需单独确认，见 §6 host 获取说明。

完整可照抄命令、同名去重/AppKeyStr/字段授权等暗坑清单、以及网关 host 命名规律 + 探测法，详见 调用前置发现。

步骤 1：确认调用信息

在发起调用前，确认以下信息已就绪并导出为环境变量（凭证不打印）：

export DATAPHIN_APP_KEY=<来自 S2>
export DATAPHIN_APP_SECRET=<来自 S2>
export DATAPHIN_GATEWAY_HOST=<控制台「网络配置」获取>
export SKILL_SESSION_ID=<父 Skill 生成的 session-id>   # 可观测标记，可选
# 调用时还需：apiId（整数）、method（LIST/GET/CREATE/UPDATE/DELETE）、
#             stage（RELEASE=生产 / PRE=开发）、env（PROD / PRE）

步骤 1.5：查询 API 文档（如路径未知）

# 通过 CLI 查询 API 文档，获取 IsPagedQuery、请求方法、参数列表
aliyun dataphin-public get-data-service-api-document \
  --op-tenant-id <tenantId> --id <apiId> \
  --profile <profile> --endpoint <endpoint>

# 关键返回字段：
#   IsPagedQuery    → 是否分页（仅辅助，不能单独用来定 methodType，见下）
#   RequestParamList → 业务请求参数（应走 conditions 字段）
#   ResponseParamList → 响应参数
#   PublicParamList → 公共参数（appkey/appsecret，仅内置网关需要）

API 调用 URL 构造规则：/{methodType}/{apiId}?appKey={appKey}&env={env}

methodType 由 API 操作类型决定（5 选 1），按操作语义/命名判断——不要只看 IsPagedQuery：

⚠️ IsPagedQuery=true 不等于 list：get 类 API 也可能 IsPagedQuery=true（实测 GetCustomer）。猜错 methodType → 403 ... not bind app {appKey}，换正确动词重试。

步骤 2：同步调用 API（推荐）

用附带脚本 scripts/call-data-service-api.py 调用，脚本内置 HMAC-SHA256 签名，无需 SDK、无需第三方库。完整说明见 Python 调用模板。

# LIST 同步调用（--method 决定路径动词，须大写）
python3 scripts/call-data-service-api.py call \
  --api-id 10083 --method LIST \
  --params '{"conditions":{},"returnFields":[],"pageStart":0,"pageSize":10,"keepColumnCase":true}' \
  --stage RELEASE --env PROD

# GET 单条查询（结果在 result 字段，不是 results）
python3 scripts/call-data-service-api.py call --api-id 10084 --method GET \
  --params '{"conditions":{"id":1}}'

# 私有化自签 HTTPS：加 --scheme HTTPS --ignore-ssl

成功时脚本输出 code == "DPN-OLTP-COMMON-000" 的 JSON 并以退出码 0 结束；业务失败退出码 1，缺少环境变量/参数错误退出码 2。

结果字段随 method 不同：LIST → results（数组）；GET → result（单个对象）。

步骤 3：异步调用（大数据量）

# 自动完成：提交 → 轮询 jobId → 合并分页结果 → closeJob
python3 scripts/call-data-service-api.py async-call \
  --api-id 10083 --method LIST --params-file query.json \
  --poll-interval 1 --timeout 600

详见 异步调用模板。

步骤 4：流式调用（SSE）与 DML

# 流式调用：逐帧输出 JSON
python3 scripts/call-data-service-api.py sse --api-id 10085 --method GET --params '{}'

# DML：--method 换 CREATE/UPDATE/DELETE，参数为 ManipulationParam
python3 scripts/call-data-service-api.py call --api-id 10083 --method CREATE \
  --params '{"conditions":{"id":1,"name":"test"}}'
# 批量操作用 batchConditions，详见 references/python-client-template.md

步骤 5：嵌入自有工程（可选）

如需在已有 Python 工程内调用（而非命令行），可照抄 Python 调用模板 的 requests 版最小客户端（约 60 行）。签名规范见该文档 §1。

步骤 6：验证调用成功

验证标准：

• 脚本退出码为 0
• 响应 code 字段为 "DPN-OLTP-COMMON-000"（不是 "0"）
• 返回数据含预期业务字段（LIST 看 results，GET 看 result）

9. Success Verification

采用三步验证法：

1. HTTP 状态检查：响应状态码为 200
2. 业务码检查：code == "DPN-OLTP-COMMON-000" 表示业务成功（注意：不是 "0"）
3. 数据完整性：返回 results 列表包含预期业务字段

10. Cleanup

本 Skill 无需清理资源。API 调用不创建持久化资源，无需回滚操作。

注意：async-call 在异步任务完成/失败后会自动调用 closeJob 关闭任务（放在 finally 中），无需手动清理。

11. Command Tables

本 Skill 用附带脚本调用数据服务网关，同时用少量 CLI 命令获取调用所需的元信息。

CLI 信息查询命令

调用脚本子命令

选项与请求参数详见 API 调用参考 及 Python 调用模板。

12. Best Practices

• 零依赖调用：用 scripts/call-data-service-api.py（纯标准库），无需安装 SDK 或 requests，签名与官方 SDK v5.5.0 逐字节一致
• 环境变量存储凭证：不要硬编码 AppKey/AppSecret，使用 DATAPHIN_APP_KEY / DATAPHIN_APP_SECRET 环境变量，且不打印
• API 调用路径：正确格式为 /{methodType}/{apiId}?appKey=xxx&env=xxx，不是 /api/<GroupId>/<ApiName>
• method 大写：--method 使用大写（LIST/GET/CREATE/UPDATE/DELETE），它决定网关路径动词
• scheme 选择：内置网关仅支持 HTTP；阿里云 API 网关支持 HTTPS（自签证书用 --ignore-ssl）
• 业务成功码：DPN-OLTP-COMMON-000（不是 0）
• 异步调用：大数据量查询用 async-call，脚本自动轮询并合并分页结果、关闭任务
• 签名自研需谨慎：x-ca-signature-headers 不含 x-ca-signature 自身、path 与签名串完全一致、JSON 请求不带 content-md5、content-type 不带 ; charset（详见 签名规范）
• 可观测标记走 user-agent：不要用 x-ca-* 头承载，避免影响签名
• 大整数 ID：API 返回的 19 位 snowflake ID 在 Python 中按字符串处理
• IN 类型参数：使用列表传递值，如 {"age": [10, 20, 30]}
• 分页稳定性：使用 ORDER BY 主键或联合主键，避免分页时数据重复或丢失