SDK:TypeScript 和 Python
两个 SDK,每个接口一个方法:共 207 个带类型的方法,每次调用自动带幂等键,按服务端允许的方式重试,翻页可设条数和花费上限。
安装
npm install @dataxapi/sdk
pip install dataxapi
Node 20 及以上;Python 3.9 及以上。两个都没有依赖。它们都由目录生成,所以 207 个接口各是一个带类型的方法,注释里写着价格。
客户端
import { DataXAPI } from "@dataxapi/sdk";
const dx = new DataXAPI(); // reads DATAXAPI_API_KEY
const labelled = new DataXAPI({ maxRetries: 3, source: "my-app" });from dataxapi import DataXAPI dx = DataXAPI() # reads DATAXAPI_API_KEY labelled = DataXAPI(max_retries=3, source="my-app")
apiKeyapi_keyKey。默认读环境变量 DATAXAPI_API_KEY。baseUrlbase_url服务地址。默认读 DATAXAPI_BASE_URL,没有就是 https://api.dataxapi.com。timeoutMstimeout一次尝试最长等多久:TypeScript 里是毫秒,Python 里是秒。默认 65 秒。maxRetriesmax_retries可以重试的请求最多再发几次。默认 2。sourcesource记在每次调用上的来源标记。endUserend_user记在每次调用上的终端用户标记。调用
可以用接口对应的方法调,也可以用 call 按编号调。两种都是一次调用、一页、各扣各的;返回的就是服务给出的返回。每次调用都带着 SDK 自己生成的幂等键,SDK 内部重试时用的是同一个。
const result = await dx.tiktok.videos.search({ query: "space station" });
console.log(result.data.items, result.meta.billing.amount);
// By ID, refused before it runs if it could cost more than $0.002.
const capped = await dx.call("tiktok.videos.search", { query: "space station" }, { maxPrice: "0.002" });result = dx.tiktok.videos.search(query="space station") print(result["data"]["items"], result["meta"]["billing"]["amount"]) # By ID, refused before it runs if it could cost more than $0.002. capped = dx.call("tiktok.videos.search", query="space station", options={"max_price": "0.002"})
要连续翻页,dx.pages(…) 一次给一页,dx.items(…) 一次给一条,可以设页数、条数和金额的上限,见翻页。
同一件事,几个平台
角色是几个平台都能做的同一件事,用同一个名字:social.posts.search 在 tiktok 上是 tiktok.videos.search,在 youtube 上是 youtube.videos.search。按角色调用时,平台只是一个参数。一次查几个平台时,每个平台各是一次调用、各扣各的,一个失败不影响其余。
const one = await dx.callByRole("social.posts.search", "tiktok", { query: "space station" }); const outcomes = await dx.acrossPlatforms("social.posts.search", ["tiktok", "youtube"], { query: "space station" }); for (const each of outcomes) console.log(each.platform, each.result?.meta.billing.amount ?? each.error?.message);
one = dx.call_by_role("social.posts.search", "tiktok", query="space station") outcomes = dx.across_platforms("social.posts.search", ["tiktok", "youtube"], query="space station") for each in outcomes: print(each["platform"], each["error"] or each["result"]["meta"]["billing"]["amount"])
报错
服务回了一个报错。上面有 status、code、charged、retry、retryAfterSeconds、requestId、param、problems、reason(Python 里是 retry_after_seconds、request_id)。每个错误代码有自己的类,例如 NotFoundError、RateLimitedError。
没有收到服务自己的回答:连接失败或超时。这次调用有没有执行,不知道。
这次调用发出去时的全部内容:接口、参数和选项,包括幂等键。
为真表示有一次发送可能已经执行,但结果没拿到。
unsettled 为真时,用同样的接口和参数,带上 error.call.options 再发一遍。键没变,它就是同一次调用:已经结算的拿回保存的结果,不会扣第二次。
import { APIConnectionError, DataXAPI, DataXAPIError } from "@dataxapi/sdk";
const dx = new DataXAPI();
const params = { query: "space station" };
let page;
try {
page = await dx.call("tiktok.videos.search", params);
} catch (error) {
if ((error instanceof DataXAPIError || error instanceof APIConnectionError) && error.unsettled && error.call) {
// The same call under the key it went out with.
page = await dx.call("tiktok.videos.search", params, error.call.options);
} else throw error;
}from dataxapi import APIConnectionError, DataXAPI, DataXAPIError
dx = DataXAPI()
try:
page = dx.call("tiktok.videos.search", query="space station")
except (DataXAPIError, APIConnectionError) as error:
if not (error.unsettled and error.call):
raise
again = error.call
page = dx.call(again["capability"], again["params"], options=again["options"])账户、用量、调用记录、目录
这几个都不收费。目录不用 Key。
dx.account()dx.account()余额、这把 Key 的上限和已花、限流额度。dx.usage({ group_by: "capability" })dx.usage(group_by="capability")已结束的调用,按分组计数和求和。dx.calls.get(requestId)dx.calls.get(request_id)一次调用后来怎么样了,扣了多少。dx.calls.result(requestId, { view: "full" })dx.calls.result(request_id, view="full")24 小时内重取保存的结果,可以换档位。dx.calls.byKey(key)dx.calls.by_key(key)一个幂等键现在代表什么。dx.catalog.search({ query: "…" })dx.catalog.search("…")用一句话找接口,最多 5 个。