实时

实时令牌、频道、发布与在线状态。

概览

基础路径: /v1/realtime

Python 命名空间: infra.realtime

text

# Python
from infrai import infra

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

方法

realtime.token.issue

POST /v1/realtime/token

签发限定频道与能力的客户端令牌。

Python

python

infra.realtime.token.issue(user_id=None, channels=None, ttl_seconds=None)

TypeScript

typescript

client.realtime.token(opts?: RealtimeTokenOptions): Promise<RealtimeToken>

参数

名称	类型	必填	说明
`user_id`	`string`	可选	连接用户的 id。
`channels`	`string[]`	可选	令牌可访问的频道。
`ttl_seconds`	`number`	可选	过期前的有效时长（秒）。

RealtimeToken { token, client_id, expires_at, endpoint }

示例

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

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

tok = infra.realtime.token.issue(user_id="u_123", channels=["orders"], ttl_seconds=300)
print("token:", (tok.get("token") or "")[:12], "...")

realtime.channel.create

POST /v1/realtime/channels

创建实时频道。

Python

python

infra.realtime.channel.create(name=..., type=None)

TypeScript

typescript

client.realtime.channel.create(opts): Promise<{ channel_id, name, type }>

参数

名称	类型	必填	说明
`name`	`string`	必填	频道名称。
`type`	`"public" \| "private" \| "presence"`	可选	频道可见性类型。

{ channel_id, name, type }

示例

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

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

ch = infra.realtime.channel.create(name="orders", type="presence")
print("channel_id:", ch.get("channel_id"))

realtime.publish

POST /v1/realtime/publish

向频道发布事件。

Python

python

infra.realtime.publish(channel=..., event=..., data=..., idempotency_key=None)

TypeScript

typescript

client.realtime.publish(opts): Promise<{ message_id }>

参数

名称	类型	必填	说明
`channel`	`string`	必填	频道名称。
`event`	`string`	必填	事件名称。
`data`	`unknown`	必填	要发布的事件负载。
`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

published = infra.realtime.publish(
    channel="orders",
    event="order.created",
    data={"order_id": "ord_42", "total_usd": 19.99},
)
print("delivered:", published.get("delivered") or published.get("status"))

realtime.presence.get

GET /v1/realtime/channels/{channel}/presence

获取频道当前的在线成员。

Python

python

infra.realtime.presence.get(channel=...)

TypeScript

typescript

// see ROUTES["realtime.presence.get"]

参数

名称	类型	必填	说明
`channel`	`string`	必填	频道名称。

{ members: Array<{ client_id, user_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

p = infra.realtime.presence.get(channel="orders")
print("members:", len(p.get("members", [])))