Skip to content

WebSocket class

Importing the WebSocket from esmerald is as simple as:

from esmerald import WebSocket

esmerald.WebSocket

WebSocket(scope, receive, send)

Bases: HTTPConnection

PARAMETER DESCRIPTION
scope

TYPE: Scope

receive

TYPE: Receive

send

TYPE: Send

Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
22
23
24
25
26
27
28
def __init__(self, scope: Scope, receive: Receive, send: Send) -> None:
    super().__init__(scope)
    assert scope["type"] == "websocket"
    self._receive = receive
    self._send = send
    self.client_state = WebSocketState.CONNECTING
    self.application_state = WebSocketState.CONNECTING

scope instance-attribute

scope = scope

app property

app

url property

url

base_url property

base_url

headers property

headers

query_params property

query_params

path_params property

path_params

cookies property

cookies

client property

client

session property

session

auth property

auth

user property

user

state property

state

client_state instance-attribute

client_state = WebSocketState.CONNECTING

application_state instance-attribute

application_state = WebSocketState.CONNECTING

url_for

url_for(name, /, **path_params)
PARAMETER DESCRIPTION
name

TYPE: str

**path_params

TYPE: Any DEFAULT: {}

Source code in .venv/lib/python3.8/site-packages/starlette/requests.py
174
175
176
177
def url_for(self, name: str, /, **path_params: typing.Any) -> URL:
    router: Router = self.scope["router"]
    url_path = router.url_path_for(name, **path_params)
    return url_path.make_absolute_url(base_url=self.base_url)

receive async

receive()

Receive ASGI websocket messages, ensuring valid state transitions.

Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
async def receive(self) -> Message:
    """
    Receive ASGI websocket messages, ensuring valid state transitions.
    """
    if self.client_state == WebSocketState.CONNECTING:
        message = await self._receive()
        message_type = message["type"]
        if message_type != "websocket.connect":
            raise RuntimeError(
                'Expected ASGI message "websocket.connect", '
                f"but got {message_type!r}"
            )
        self.client_state = WebSocketState.CONNECTED
        return message
    elif self.client_state == WebSocketState.CONNECTED:
        message = await self._receive()
        message_type = message["type"]
        if message_type not in {"websocket.receive", "websocket.disconnect"}:
            raise RuntimeError(
                'Expected ASGI message "websocket.receive" or '
                f'"websocket.disconnect", but got {message_type!r}'
            )
        if message_type == "websocket.disconnect":
            self.client_state = WebSocketState.DISCONNECTED
        return message
    else:
        raise RuntimeError(
            'Cannot call "receive" once a disconnect message has been received.'
        )

send async

send(message)

Send ASGI websocket messages, ensuring valid state transitions.

PARAMETER DESCRIPTION
message

TYPE: Message

Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
async def send(self, message: Message) -> None:
    """
    Send ASGI websocket messages, ensuring valid state transitions.
    """
    if self.application_state == WebSocketState.CONNECTING:
        message_type = message["type"]
        if message_type not in {"websocket.accept", "websocket.close"}:
            raise RuntimeError(
                'Expected ASGI message "websocket.accept" or '
                f'"websocket.close", but got {message_type!r}'
            )
        if message_type == "websocket.close":
            self.application_state = WebSocketState.DISCONNECTED
        else:
            self.application_state = WebSocketState.CONNECTED
        await self._send(message)
    elif self.application_state == WebSocketState.CONNECTED:
        message_type = message["type"]
        if message_type not in {"websocket.send", "websocket.close"}:
            raise RuntimeError(
                'Expected ASGI message "websocket.send" or "websocket.close", '
                f"but got {message_type!r}"
            )
        if message_type == "websocket.close":
            self.application_state = WebSocketState.DISCONNECTED
        await self._send(message)
    else:
        raise RuntimeError('Cannot call "send" once a close message has been sent.')

accept async

accept(subprotocol=None, headers=None)
PARAMETER DESCRIPTION
subprotocol

TYPE: Optional[str] DEFAULT: None

headers

TYPE: Optional[Iterable[Tuple[bytes, bytes]]] DEFAULT: None

Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
async def accept(
    self,
    subprotocol: typing.Optional[str] = None,
    headers: typing.Optional[typing.Iterable[typing.Tuple[bytes, bytes]]] = None,
) -> None:
    headers = headers or []

    if self.client_state == WebSocketState.CONNECTING:
        # If we haven't yet seen the 'connect' message, then wait for it first.
        await self.receive()
    await self.send(
        {"type": "websocket.accept", "subprotocol": subprotocol, "headers": headers}
    )

receive_text async

receive_text()
Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
107
108
109
110
111
112
113
114
async def receive_text(self) -> str:
    if self.application_state != WebSocketState.CONNECTED:
        raise RuntimeError(
            'WebSocket is not connected. Need to call "accept" first.'
        )
    message = await self.receive()
    self._raise_on_disconnect(message)
    return typing.cast(str, message["text"])

receive_bytes async

receive_bytes()
Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
116
117
118
119
120
121
122
123
async def receive_bytes(self) -> bytes:
    if self.application_state != WebSocketState.CONNECTED:
        raise RuntimeError(
            'WebSocket is not connected. Need to call "accept" first.'
        )
    message = await self.receive()
    self._raise_on_disconnect(message)
    return typing.cast(bytes, message["bytes"])

receive_json async

receive_json(mode='text')
PARAMETER DESCRIPTION
mode

TYPE: str DEFAULT: 'text'

Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
async def receive_json(self, mode: str = "text") -> typing.Any:
    if mode not in {"text", "binary"}:
        raise RuntimeError('The "mode" argument should be "text" or "binary".')
    if self.application_state != WebSocketState.CONNECTED:
        raise RuntimeError(
            'WebSocket is not connected. Need to call "accept" first.'
        )
    message = await self.receive()
    self._raise_on_disconnect(message)

    if mode == "text":
        text = message["text"]
    else:
        text = message["bytes"].decode("utf-8")
    return json.loads(text)

iter_text async

iter_text()
Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
141
142
143
144
145
146
async def iter_text(self) -> typing.AsyncIterator[str]:
    try:
        while True:
            yield await self.receive_text()
    except WebSocketDisconnect:
        pass

iter_bytes async

iter_bytes()
Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
148
149
150
151
152
153
async def iter_bytes(self) -> typing.AsyncIterator[bytes]:
    try:
        while True:
            yield await self.receive_bytes()
    except WebSocketDisconnect:
        pass

iter_json async

iter_json()
Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
155
156
157
158
159
160
async def iter_json(self) -> typing.AsyncIterator[typing.Any]:
    try:
        while True:
            yield await self.receive_json()
    except WebSocketDisconnect:
        pass

send_text async

send_text(data)
PARAMETER DESCRIPTION
data

TYPE: str

Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
162
163
async def send_text(self, data: str) -> None:
    await self.send({"type": "websocket.send", "text": data})

send_bytes async

send_bytes(data)
PARAMETER DESCRIPTION
data

TYPE: bytes

Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
165
166
async def send_bytes(self, data: bytes) -> None:
    await self.send({"type": "websocket.send", "bytes": data})

send_json async

send_json(data, mode='text')
PARAMETER DESCRIPTION
data

TYPE: Any

mode

TYPE: str DEFAULT: 'text'

Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
168
169
170
171
172
173
174
175
async def send_json(self, data: typing.Any, mode: str = "text") -> None:
    if mode not in {"text", "binary"}:
        raise RuntimeError('The "mode" argument should be "text" or "binary".')
    text = json.dumps(data, separators=(",", ":"), ensure_ascii=False)
    if mode == "text":
        await self.send({"type": "websocket.send", "text": text})
    else:
        await self.send({"type": "websocket.send", "bytes": text.encode("utf-8")})

close async

close(code=1000, reason=None)
PARAMETER DESCRIPTION
code

TYPE: int DEFAULT: 1000

reason

TYPE: Optional[str] DEFAULT: None

Source code in .venv/lib/python3.8/site-packages/starlette/websockets.py
177
178
179
180
181
182
async def close(
    self, code: int = 1000, reason: typing.Optional[str] = None
) -> None:
    await self.send(
        {"type": "websocket.close", "code": code, "reason": reason or ""}
    )

esmerald.WebSocketDisconnect

WebSocketDisconnect(code=1000, reason=None)

Bases: WebSocketDisconnect

Esmerald WebSocketDisconnect

PARAMETER DESCRIPTION
code

TYPE: int DEFAULT: 1000

reason

TYPE: Optional[str] DEFAULT: None

Source code in esmerald/websockets.py
13
14
def __init__(self, code: int = 1000, reason: Optional[str] = None) -> None:
    super().__init__(code, reason)  # pragma: no cover

code instance-attribute

code = code

reason instance-attribute

reason = reason or ''

esmerald.WebSocketState

Bases: Enum

CONNECTING class-attribute instance-attribute

CONNECTING = 0

CONNECTED class-attribute instance-attribute

CONNECTED = 1

DISCONNECTED class-attribute instance-attribute

DISCONNECTED = 2