Python: Realtime

Publish to channels, replay missed messages, see presence and channels, open WebSockets with tickets and manage end-to-end encryption keys with the Python SDK.

await sovaria.stream_publish("room:42", {"text": "Hello!"})
missed = await sovaria.stream_replay("room:42", after=last_seen, limit=100)

await sovaria.e2ee_bootstrap_key(
    channel: str,
    *, ttl_seconds: int = 86400,
    force_new: bool = False,
) -> dict[str, Any]

await sovaria.e2ee_get_active_key(
    channel: str,
    client_public_key: str | None = None,
) -> dict[str, Any]

await sovaria.e2ee_get_key(
    channel: str,
    key_id: str,
    client_public_key: str | None = None,
) -> dict[str, Any]

await sovaria.e2ee_list_keys(channel: str) -> dict[str, Any]

await sovaria.e2ee_revoke_key(
    channel: str,
    *, key_id: str | None = None,
    reason: str | None = None,
) -> dict[str, Any]

await sovaria.e2ee_rotate_key(
    channel: str,
    *, ttl_seconds: int = 86400,
) -> dict[str, Any]

await sovaria.issue_realtime_ws_ticket(channel: str) -> str

sovaria.realtime_ws_url(
    channel: str,
    *, ticket: str | None = None,
) -> str

await sovaria.realtime_ws_url_with_ticket(channel: str) -> str

await sovaria.stream_channels() -> dict[str, Any]

await sovaria.stream_presence(channel: str) -> dict[str, Any]

await sovaria.stream_publish(
    channel: str,
    payload: dict[str, Any],
    *, clock: dict[str, int] | None = None,
    clock_delta: dict[str, int] | None = None,
) -> dict[str, Any]

await sovaria.stream_replay(
    channel: str,
    *, after: int = 0,
    limit: int = 200,
) -> dict[str, Any]

await sovaria.stream_subscribe(
    channel: str,
    client_id: str,
) -> dict[str, Any]

await sovaria.stream_unsubscribe(
    channel: str,
    client_id: str,
) -> dict[str, Any]