Skip to content

WebSockets

Starlette contains an WebSocket class that acts similar to HTTP requests but allows data to be sent and received over WebSockets.

WebSocket

Signed by: WebSocket(scope, receive=None, send=None)

from 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 provide a mapping interface, so you can use them just as you would scope.

For example:websocket['path'] will return the ASGI path.

Uniform Resource Locator (also commonly referred to simply as URL)

WebSocket URL is accessed as websocket.url.

This property is actually a subclass of str and exposes all components that can be parsed from the URL.

For example:websocket.url.path,websocket.url.port,websocket.url.scheme.

headline

The header is exposed as an immutable, case-insensitive multidictionary.

For example: websocket.headers['sec-websocket-version']

query parameter

Query parameters are exposed as immutable multiple dictionaries.

For example: websocket.query_params['search']

path parameters

Router path parameters are exposed as dictionary interfaces.

For example: websocket.path_params['username']

accept connections

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

TXD

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

As of version 0.10.0, JSON messages are sent in text data frames by default. Use websocket.send_json(data, mode="binary") to send JSON in binary data frames.

RXD

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

The text "May raise starlette.websockets.WebSocketDisconnect() ."seems not to conform to the conventional semantic expression, it is difficult to accurately translate into Chinese. If you have more context information or confirm the accuracy of the text content, I will try my best to provide you with a more satisfactory translation starlette.websockets.WebSocketDisconnect() But this translation may not be quite in line with the actual meaning.

As of version 0.10.0, JSON messages are received in text data frames by default. Use websocket.receive_json(data, mode="binary") to receive JSON in binary data frames.

iterative data

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

Similar to receive_text,receive_bytes, and receive_json, but returns an asynchronous iterator.

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

When starlette.websockets.WebSocketDisconnect is promoted, the iterator exits.

close the connection

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

send and receive messages

If you need to send or receive raw ASGI messages, you should use websocket.send() and websocket.receive() instead of the raw send and receive callables. This ensures that the WebSocket state is updated correctly.

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

Send a rejection response

If you call websocket.close() before calling websocket.accept(), the server automatically sends an HTTP 403 error to the client.

If you want to send a different error response, you can use the websocket.send_denial_response() method. This will send a response and then close the connection.

  • await websocket.send_denial_response(response)

This requires the ASGI server to support the WebSocket reject response extension. If it is not supported, an RuntimeError error will be raised.