SDK Azure Web PubSub Service pour Python

Messagerie en temps réel avec connexions WebSocket à grande échelle.

Installation

# Service SDK (côté serveur)
pip install azure-messaging-webpubsubservice

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

Variables d'environnement

AZURE_WEBPUBSUB_CONNECTION_STRING=Endpoint=https://<name>.webpubsub.azure.com;AccessKey=...  # Alternative à l'authentification Entra ID
AZURE_WEBPUBSUB_HUB=my-hub  # Requis pour toutes les méthodes d'authentification
AZURE_TOKEN_CREDENTIALS=prod # Requis uniquement si DefaultAzureCredential est utilisé en production

Service Client (côté serveur)

Authentification

from azure.messaging.webpubsubservice import WebPubSubServiceClient

# Chaîne de connexion
client = WebPubSubServiceClient.from_connection_string(
    connection_string=os.environ["AZURE_WEBPUBSUB_CONNECTION_STRING"],
    hub="my-hub"
)

# Entra ID
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential

# Dev local : DefaultAzureCredential. Production : définir AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Ou utiliser un credential spécifique directement en production :
# Voir https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

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

Générer un token d'accès client

# Token pour utilisateur anonyme
token = client.get_client_access_token()
print(f"URL: {token['url']}")

# Token avec ID utilisateur
token = client.get_client_access_token(
    user_id="user123",
    roles=["webpubsub.sendToGroup", "webpubsub.joinLeaveGroup"]
)

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

Envoyer à tous les clients

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

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

Envoyer à un utilisateur

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

Envoyer à un groupe

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

Envoyer à une connexion

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

Gestion des groupes

# Ajouter un utilisateur à un groupe
client.add_user_to_group(group="my-group", user_id="user123")

# Retirer un utilisateur d'un groupe
client.remove_user_from_group(group="my-group", user_id="user123")

# Ajouter une connexion à un groupe
client.add_connection_to_group(group="my-group", connection_id="abc123")

# Retirer une connexion d'un groupe
client.remove_connection_from_group(group="my-group", connection_id="abc123")

Gestion des connexions

# Vérifier si une connexion existe
exists = client.connection_exists(connection_id="abc123")

# Vérifier si un utilisateur a des connexions
exists = client.user_exists(user_id="user123")

# Vérifier si un groupe a des connexions
exists = client.group_exists(group="my-group")

# Fermer une connexion
client.close_connection(connection_id="abc123", reason="Session ended")

# Fermer toutes les connexions pour un utilisateur
client.close_all_connections(user_id="user123")

Accorder/révoquer des permissions

from azure.messaging.webpubsubservice import WebPubSubServiceClient

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

# Révoquer une permission
client.revoke_permission(
    permission="joinLeaveGroup",
    connection_id="abc123",
    target_name="my-group"
)

# Vérifier une permission
has_permission = client.check_permission(
    permission="joinLeaveGroup",
    connection_id="abc123",
    target_name="my-group"
)

Client SDK (client WebSocket Python)

from azure.messaging.webpubsubclient import WebPubSubClient

client = WebPubSubClient(credential=token["url"])

# Gestionnaires d'événements
@client.on("connected")
def on_connected(e):
    print(f"Connected: {e.connection_id}")

@client.on("server-message")
def on_message(e):
    print(f"Message: {e.data}")

@client.on("group-message")
def on_group_message(e):
    print(f"Group {e.group}: {e.data}")

# Connexion et envoi
client.open()
client.send_to_group("my-group", "Hello from Python!")

Service Client asynchrone

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()

Opérations client

Bonnes pratiques

1. Utiliser les rôles pour limiter les permissions des clients
2. Utiliser les groupes pour la messagerie ciblée
3. Générer des tokens de courte durée pour la sécurité
4. Utiliser les ID utilisateur pour envoyer aux utilisateurs sur plusieurs connexions
5. Gérer la reconnexion dans les applications client
6. Utiliser le type de contenu JSON pour les données structurées
7. Fermer les connexions proprement avec des raisons