调度

定时任务、消息队列与出向 Webhook。

概览

基础路径: /v1/scheduling

Python 命名空间: infra.cron

text

# Python
from infrai import infra

# TypeScript
import { InfraiClient } from "@infrai/sdk";
const client = new InfraiClient({ apiKey: process.env.INFRAI_API_KEY! });
// → client.job

方法

scheduling.cron.create

POST /v1/scheduling/cron

创建按计划向 URL 发起 POST 的定时任务。

Python

python

infra.cron.create(cron_expr, task, *, name=None, timezone="UTC", retry=3, timeout_seconds=300, payload=None, idempotency_key=...)

TypeScript

typescript

client.job.cron.create(opts: CronOptions): Promise<CronRecord>

参数

名称	类型	必填	说明
`name`	`string`	必填	可读的任务名称。
`schedule`	`string`	必填	Cron 表达式，例如 0 9 * * *。
`url`	`string`	必填	接收定时 POST 的 URL。
`payload`	`unknown`	可选	每次运行附带的可选 JSON 负载。
`timezone`	`string`	可选	调度使用的 IANA 时区。
`retries`	`number`	可选	失败时的重试次数。
`idempotency_key`	`string`	可选	可选去重 key；相同重试将返回同一结果。

CronRecord { cron_id, name, schedule, url, enabled, next_run_at? }

示例

一次性前置(每个范例都假定已完成):

bash

pip install infrai
# one-time auth (no secret needed): anonymous account + trial, writes ~/.infrai/credentials
python -c "from infrai import infra; infra.activate()"
# returning user instead: export INFRAI_API_KEY=ifr_pk_proj_...

python

from infrai import infra

job = infra.cron.create(
    "0 9 * * *",                 # every day at 09:00
    "send-daily-digest",
    name="daily-digest",
    timezone="UTC",
    payload={"segment": "active-users"},
)
print("cron_id:", job.get("id") or job.get("cron_id"))

scheduling.cron.list

GET /v1/scheduling/cron

列出定时任务。

Python

python

infra.cron.list()

TypeScript

typescript

// see ROUTES["scheduling.cron.list"]

{ items: CronRecord[] }

示例

一次性前置(每个范例都假定已完成):

bash

pip install infrai
# one-time auth (no secret needed): anonymous account + trial, writes ~/.infrai/credentials
python -c "from infrai import infra; infra.activate()"
# returning user instead: export INFRAI_API_KEY=ifr_pk_proj_...

python

from infrai import infra

for j in infra.cron.list().get("items", []):
    print(j.get("cron_id"), j.get("schedule"), "next:", j.get("next_run_at"))

scheduling.queue.publish

POST /v1/scheduling/queue/publish

向队列发布消息，可选延迟。

Python

python

infra.queue.publish(queue, payload, *, delay_seconds=None, idempotency_key=None)

TypeScript

typescript

// see ROUTES["scheduling.queue.publish"]

参数

名称	类型	必填	说明
`queue`	`string`	必填	队列名称。
`payload`	`object`	必填	消息负载。
`delay_seconds`	`number`	可选	投递前的延迟（秒）。
`idempotency_key`	`string`	可选	可选去重 key；相同重试将返回同一结果。

{ message_id }

示例

一次性前置(每个范例都假定已完成):

bash

pip install infrai
# one-time auth (no secret needed): anonymous account + trial, writes ~/.infrai/credentials
python -c "from infrai import infra; infra.activate()"
# returning user instead: export INFRAI_API_KEY=ifr_pk_proj_...

python

from infrai import infra

msg = infra.queue.publish(
    "send-daily-digest",
    {"user_id": "u_123", "kind": "manual-trigger"},
    delay_seconds=0,
)
print("message_id:", msg.get("message_id") or msg.get("id"))

scheduling.webhook.send

POST /v1/scheduling/webhook/send

发送出向 Webhook，带重试与 HMAC 签名。

Python

python

infra.webhook.send(url, payload, ...)

TypeScript

typescript

// see ROUTES["scheduling.webhook.send"]

参数

名称	类型	必填	说明
`url`	`string`	必填	Webhook 的目标 URL。
`payload`	`object`	必填	消息负载。

{ delivery_id, state }

示例

一次性前置(每个范例都假定已完成):

bash

pip install infrai
# one-time auth (no secret needed): anonymous account + trial, writes ~/.infrai/credentials
python -c "from infrai import infra; infra.activate()"
# returning user instead: export INFRAI_API_KEY=ifr_pk_proj_...

python

from infrai import infra

d = infra.webhook.send("https://api.example.com/hooks/order",
                       {"order_id": "ord_42", "status": "paid"})
print("delivery:", d.get("delivery_id"), "| state:", d.get("state"))