WebSockets

Starlette contient une classe WebSocket qui agit comme une requête HTTP, mais permet d'envoyer et de recevoir des données sur WebSocket.

WebSocket pour

Signé par: WebSocket(scope, receive=None, send=None)

de starlette.websockets import WebSocket


async def app(scope, receive, send):
    Websocket = WebSocket(scope=scope, receive=receive, send=send)
    await websocket.accept()
    await websocket.send_text('Hello, world!')
    await websocket.close()

WebSockets fournit une interface de mappage, de sorte que vous pouvez les utiliser comme vous le feriez avec scope.

Exemple:websocket['path'] renverra le chemin ASGI.

Uniform Resource Locator (également souvent appelé simplement adresse Web)

Les URL WebSocket sont accessibles en websocket.url.

Cette propriété est en fait une sous-classe de str et expose également tous les composants qui peuvent être analysés à partir de l'URL.

Exemples:websocket.url.path,websocket.url.port,websocket.url.scheme.

Le titre

Les en-têtes sont exposés comme immuables, insensibles à la casse multidictionnaire.

Par exemple: websocket.headers['sec-websocket-version']

Query paramètres

Les paramètres de requête sont exposés comme des dictionnaires multiples immuables.

Par exemple: websocket.query_params['search']

Paramètres de chemin

Les paramètres de chemin de routeur sont exposés comme une interface de dictionnaire.

Par exemple: websocket.path_params['username']

Accepter la connexion

• await websocket.accept(subprotocol=None, headers=None)

Envoyer des données

• await websocket.send_text(data)
• await websocket.send_bytes(data)
• await websocket.send_json(data)

Depuis la version 0.10.0, les messages JSON sont envoyés par défaut via une trame de données textuelle.websocket.send_json(data, mode="binary") est utilisé pour envoyer JSON via une trame de données binaire.

Recevoir les données

• await websocket.receive_text()
• await websocket.receive_bytes()
• await websocket.receive_json()

Le texte « May raise starlette.websockets.WebSocketDisconnect().» ne semble pas correspondre à l'expression sémantique conventionnelle, ce qui rend difficile une traduction chinoise précise. si vous avez plus d'informations contextuelles ou si vous confirmez l'exactitude du contenu du texte, je ferai de mon mieux pour vous fournir une traduction plus satisfaisante. actuellement, dans une traduction littérale, il ne peut être traduit que par: « peut-être améliorer starlette.websockets.WebSocketDisconnect().» mais cette traduction peut ne pas correspondre tout à fait au sens pratique.

Depuis la version 0.10.0, les messages JSON sont reçus par défaut via des trames de données textuelles.websocket.receive_json(data, mode="binary") est utilisé pour recevoir JSON via des trames de données binaires.

Iteration des données

• websocket.iter_text()
• websocket.iter_bytes()
• websocket.iter_json()

Similaire à receive_text,receive_bytes et receive_json, mais renvoie un itérateur asynchrone.

de starlette.websockets import WebSocket


async def app(scope, receive, send):
    Websocket = WebSocket(scope=scope, receive=receive, send=send)
    await websocket.accept()
    async for message in websocket.iter_text():
        await websocket.send_text(f"Message text was:{message}")
    await websocket.close()

Lorsque starlette.websockets.WebSocketDisconnect est promu, l'itérateur se termine.

Fermer la connexion.

• await websocket.close(code=1000, reason=None)

Envoyer et recevoir des messages

Si vous avez besoin d'envoyer ou de recevoir des messages ASGI bruts, alors vous devriez utiliser websocket.send() et websocket.receive() au lieu d'utiliser les appelables send et receive bruts.

• await websocket.send(message)
• await websocket.receive()

Envoyer une réponse de refus

Si vous appelez websocket.close() avant d'appeler websocket.accept(), le serveur enverra automatiquement une erreur HTTP 403 au client.

Si vous souhaitez envoyer une réponse d'erreur différente, vous pouvez utiliser la méthode websocket.send_denial_response(). cela enverra une réponse et fermera la connexion par la suite.

• await websocket.send_denial_response(response)

Cela nécessite que le serveur ASGI prenne en charge l'extension WebSocket Denial Response. si elle n'est pas prise en charge, une erreur RuntimeError sera levée.