azure-messaging-webpubsubservice-py

Python 用 Azure Web PubSub Service SDK

大規模な WebSocket 接続によるリアルタイムメッセージング。

インストール

# Service SDK (server-side)
pip install azure-messaging-webpubsubservice

# Client SDK (for Python WebSocket clients)
pip install azure-messaging-webpubsubclient

環境変数

AZURE_WEBPUBSUB_CONNECTION_STRING=Endpoint=https://<name>.webpubsub.azure.com;AccessKey=...
AZURE_WEBPUBSUB_HUB=my-hub

サービスクライアント（サーバー側）

認証

import os
from azure.messaging.webpubsubservice import WebPubSubServiceClient

# Connection string
client = WebPubSubServiceClient.from_connection_string(
    connection_string=os.environ["AZURE_WEBPUBSUB_CONNECTION_STRING"],
    hub="my-hub"
)

# Entra ID
from azure.identity import DefaultAzureCredential

client = WebPubSubServiceClient(
    endpoint="https://<name>.webpubsub.azure.com",
    hub="my-hub",
    credential=DefaultAzureCredential()
)

クライアントアクセストークンの生成

# Token for anonymous user
token = client.get_client_access_token()
# Treat token["url"] as a bearer secret; return it only to the authenticated intended client.

# Token with user ID
token = client.get_client_access_token(
    user_id="user123",
    roles=["webpubsub.sendToGroup.my-group", "webpubsub.joinLeaveGroup.my-group"]
)

# Token with groups
token = client.get_client_access_token(
    user_id="user123",
    groups=["group1", "group2"]
)

すべてのクライアントに送信

# Send text
client.send_to_all(message="Hello everyone!", content_type="text/plain")

# Send JSON
client.send_to_all(
    message={"type": "notification", "data": "Hello"},
    content_type="application/json"
)

ユーザーに送信

client.send_to_user(
    user_id="user123",
    message="Hello user!",
    content_type="text/plain"
)

グループに送信

client.send_to_group(
    group="my-group",
    message="Hello group!",
    content_type="text/plain"
)

接続に送信

client.send_to_connection(
    connection_id="abc123",
    message="Hello connection!",
    content_type="text/plain"
)

グループ管理

# Add user to group
client.add_user_to_group(group="my-group", user_id="user123")

# Remove user from group
client.remove_user_from_group(group="my-group", user_id="user123")

# Add connection to group
client.add_connection_to_group(group="my-group", connection_id="abc123")

# Remove connection from group
client.remove_connection_from_group(group="my-group", connection_id="abc123")

接続管理

# Check if connection exists
exists = client.connection_exists(connection_id="abc123")

# Check if user has connections
exists = client.user_exists(user_id="user123")

# Check if group has connections
exists = client.group_exists(group="my-group")

# Close connection
client.close_connection(connection_id="abc123", reason="Session ended")

# Close all connections for user
client.close_all_connections(user_id="user123")

権限の付与／取り消し

from azure.messaging.webpubsubservice import WebPubSubServiceClient

# Grant permission
client.grant_permission(
    permission="joinLeaveGroup",
    connection_id="abc123",
    target_name="my-group"
)

# Revoke permission
client.revoke_permission(
    permission="joinLeaveGroup",
    connection_id="abc123",
    target_name="my-group"
)

# Check permission
has_permission = client.check_permission(
    permission="joinLeaveGroup",
    connection_id="abc123",
    target_name="my-group"
)

クライアント SDK（Python WebSocket クライアント）

from azure.messaging.webpubsubclient import WebPubSubClient
from azure.messaging.webpubsubclient.models import CallbackType, WebPubSubDataType

with WebPubSubClient(token["url"]) as ws_client:
    ws_client.subscribe(CallbackType.CONNECTED, lambda e: print("Connected"))
    ws_client.subscribe(CallbackType.GROUP_MESSAGE, lambda e: print("Group message received"))
    ws_client.join_group("my-group")
    ws_client.send_to_group("my-group", "Hello from Python!", WebPubSubDataType.TEXT)

トークンの発行はサーバーで認可し、ユーザー ID と許可されたグループは、信頼できないリクエストフィールドではなく、認証済みセッションから取得してください。固定された SDK がサポートする、明示的で短い有効期限を設定し、アクセストークン URL やメッセージ本文のログ出力は避け、使用後はサービスクライアントと認証情報を閉じてください。これらの送信やグループ変更は接続中のクライアントに影響するため、例ではテスト用ハブを使用してください。

非同期サービスクライアント

from azure.messaging.webpubsubservice.aio import WebPubSubServiceClient
from azure.identity.aio import DefaultAzureCredential

async def broadcast():
    credential = DefaultAzureCredential()
    client = WebPubSubServiceClient(
        endpoint="https://<name>.webpubsub.azure.com",
        hub="my-hub",
        credential=credential
    )
    
    await client.send_to_all("Hello async!", content_type="text/plain")
    
    await client.close()
    await credential.close()

クライアント操作

操作	説明
get_client_access_token	WebSocket 接続 URL を生成
send_to_all	すべての接続にブロードキャスト
send_to_user	特定のユーザーに送信
send_to_group	グループのメンバーに送信
send_to_connection	特定の接続に送信
add_user_to_group	ユーザーをグループに追加
remove_user_from_group	ユーザーをグループから削除
close_connection	クライアントを切断
connection_exists	接続状態を確認

ベストプラクティス

1. ロールを使用して、クライアントの権限を制限します
2. グループを使用して、対象を絞ったメッセージングを行います
3. 有効期間の短いトークンを生成して、セキュリティを確保します
4. ユーザー ID を使用して、複数の接続にまたがるユーザーに送信します
5. クライアントアプリケーションで再接続を処理します
6. 構造化データにはJSONコンテンツタイプを使用します
7. 理由を指定して、接続を適切に閉じます

使用する場面

このスキルは、概要で説明されているワークフローまたは操作を実行する場合に適用できます。

制限事項

• このスキルは、タスクが上記の範囲に明確に一致する場合にのみ使用してください。
• 出力を、環境固有の検証、テスト、または専門家によるレビューの代替として扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中止して確認を求めてください。