genesys-notifications 客户端

这是一个用于接收genesys WebSocket通知的简单库。

有关背景信息，请参阅 https://developer.genesys.cloud/api/rest/v2/notifications/notification_service

该库提供以下功能：

• 通道连接和主题订阅
• 通知的异步迭代器
• 失败时自动重新连接通道
• 在无法恢复时提供有用的自定义异常
• 在过期或维护关闭时自动延长或轮换通道寿命

该库的用途故意限制为通过封装上述功能，使单个通知通道管理更方便。

它 不

• 刷新访问令牌
• 管理多个通道[^2]。
• 处理收到的主题通知[^3]
• 考虑最大20个通道的限制
• 考虑每个通道订阅1000个主题的限制
• 考虑每个通道一个连接的限制

用法

首先根据Genesys文档的说明创建Genesys通道。然后

1. 导入并实例化一个通道，传入URI和主题。
2. 等待通道连接并订阅。
3. 遍历结果以接收通知。

示例

>>> uri = "wss://streaming.mypurecloud.com/channels/streaming-0-fmtdmf8cdis7jh14udg5p89t6z"
>>> topics = ["v2.analytics.queues.tdmf8cd-k3h43h-udg5p89.observations"]
>>> from genesys_notifications import Channel
>>> from genesys_notifications.exceptions import InitializationFailure, RecoveryFailure
>>> async def test(uri, topics):
...    try:
...       notifications = await Channel(uri, topics)
...    except InitializationFailure:
...       # Handle websocket connection opening and topic subscription failures;
...       # ConnectionFailure and SubscriptionFailure can be caught separately as well.
...    try:
...       async for n in notifications:
...    except RecoveryFailure:
...       # Handle cases when the channel is not able to automatically recover from issues:
...       # in case of problems, it will try once to reconnect & resubscibe before giving up, and
...       # in case of scheduled expiry or ad-hoc Genesys-side close notification, it will try to
          # extend its lifetime by resubscribing to topics.
...    else:
...       print(n)
>>>
>>> import asyncio
>>> asyncio.run(test(uri, topics)) 
{
    "topicName": "channel.metadata",
    "eventBody":{
        "message":"WebSocket Heartbeat"
    }
}

请参阅exceptions模块以获取所有可用的异常。异常的原因始终可用在其reason属性中。请参阅exceptions.REASON枚举以获取可能的原因。

^[1]: 使用extend=False初始化通道以禁用自动生命周期扩展，并通过捕获ChannelExpiring来手动处理[^2]：除非调用channel.rollover(uri)；否则，在底层初始化一个新的WebSocket连接，并透明地替换原始连接[^3]：仅包括支持提供的功能所需的最小配置；订阅确认、通道关闭通知等。