Перейти к содержанию

WebSocket class

Warning

The current page still doesn't have a translation for this language.

But you can help translating it: Contributing.

Importing the WebSocket from esmerald is as simple as:

from esmerald import WebSocket

esmerald.WebSocket 

WebSocket(scope, receive, send)

Bases: WebsocketMixin

PARAMETER DESCRIPTION
scope

TYPE: Scope

receive

TYPE: Receive

send

TYPE: Send

Source code in lilya/websockets.py 
20
21
22
23
24
25
26
27
def __init__(self, scope: Scope, receive: Receive, send: Send) -> None:
    super().__init__(scope)
    assert scope["type"] == ScopeType.WEBSOCKET
    self._receive = receive
    self._send = send
    self._accepted = False
    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

state property 

state

query_params property 

query_params

path_params property 

path_params

cookies property 

cookies

client property 

client

server property 

server

auth property 

auth

user property 

user

session property 

session

is_server_push property 

is_server_push

is_server_pull property 

is_server_pull

client_state instance-attribute 

client_state = CONNECTING

application_state instance-attribute 

application_state = CONNECTING

set_session 

set_session(value)

Sets the value of a request session by passing a dictionary.

PARAMETER DESCRIPTION
value

TYPE: Any

Source code in lilya/_internal/_connection.py 
177
178
179
180
181
def set_session(self, value: Any) -> None:
    """
    Sets the value of a request session by passing a dictionary.
    """
    self.scope["session"] = value

clear_session 

clear_session()

Clears the scope session.

Source code in lilya/_internal/_connection.py 
183
184
185
186
187
def clear_session(self) -> None:
    """
    Clears the scope session.
    """
    self.scope["session"] = None

is_secure 

is_secure()

Check if the connection is secure (HTTPS).

RETURNS DESCRIPTION
bool

True if the connection is secure (HTTPS), False otherwise.

TYPE: bool

Source code in lilya/_internal/_connection.py 
189
190
191
192
193
194
195
196
def is_secure(self) -> bool:
    """
    Check if the connection is secure (HTTPS).

    Returns:
        bool: True if the connection is secure (HTTPS), False otherwise.
    """
    return self.url.is_secure

path_for 

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

TYPE: str

**path_params

TYPE: Any DEFAULT: {}

Source code in lilya/_internal/_connection.py 
198
199
200
201
def path_for(self, name: str, /, **path_params: Any) -> URL:
    router: Router = self.scope["router"]
    url_path = router.path_for(name, **path_params)
    return url_path.make_absolute_url(base_url=self.base_url)

raise_for_disconnect 

raise_for_disconnect(message)
PARAMETER DESCRIPTION
message

TYPE: Message

Source code in lilya/websockets.py 
29
30
31
def raise_for_disconnect(self, message: Message) -> None:
    if message["type"] == Event.WEBSOCKET_DISCONNECT:
        raise WebSocketDisconnect(message["code"], message.get("reason"))

raise_for_connection_state async 

raise_for_connection_state()
Source code in lilya/websockets.py 
33
34
35
async def raise_for_connection_state(self) -> None:
    if self.application_state != WebSocketState.CONNECTED:
        raise WebSocketRuntimeError('WebSocket is not connected. Need to call "accept" first.')

receive async 

receive()

Receive ASGI websocket messages, ensuring valid state transitions.

Source code in lilya/websockets.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
async def receive(self) -> Message:
    """
    Receive ASGI websocket messages, ensuring valid state transitions.
    """
    if self.client_state not in {WebSocketState.CONNECTING, WebSocketState.CONNECTED}:
        raise WebSocketRuntimeError(
            'Cannot call "receive" once a disconnect message has been received.'
        )

    message = await self._receive()
    message_type = message["type"]

    if self.client_state == WebSocketState.CONNECTING:
        if message_type != Event.WEBSOCKET_CONNECT:
            raise WebSocketRuntimeError(
                'Expected ASGI message "websocket.connect", ' f"but got {message_type!r}"
            )
        self.client_state = WebSocketState.CONNECTED
    elif self.client_state == WebSocketState.CONNECTED:
        if message_type not in {Event.WEBSOCKET_RECEIVE, Event.WEBSOCKET_DISCONNECT}:
            raise WebSocketRuntimeError(
                "Expected ASGI message websocket.receive or "
                f'"websocket.disconnect", but got {message_type!r}'
            )

        self.client_state = (
            WebSocketState.DISCONNECTED
            if message_type == Event.WEBSOCKET_DISCONNECT
            else self.client_state
        )
    return message

send async 

send(message)

Send ASGI websocket messages, ensuring valid state transitions.

PARAMETER DESCRIPTION
message

TYPE: Message

Source code in lilya/websockets.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
async def send(self, message: Message) -> None:
    """
    Send ASGI websocket messages, ensuring valid state transitions.
    """
    if self.application_state not in {WebSocketState.CONNECTING, WebSocketState.CONNECTED}:
        raise WebSocketRuntimeError('Cannot call "send" once a close message has been sent.')

    message_type = message["type"]

    if self.application_state == WebSocketState.CONNECTING:
        if message_type not in {Event.WEBSOCKET_ACCEPT, Event.WEBSOCKET_CLOSE}:
            raise WebSocketRuntimeError(
                'Expected ASGI message "websocket.accept" or '
                f'"websocket.close", but got {message_type!r}'
            )
        self.application_state = (
            WebSocketState.DISCONNECTED
            if message_type == Event.WEBSOCKET_CLOSE
            else WebSocketState.CONNECTED
        )
    elif self.application_state == WebSocketState.CONNECTED:
        if message_type not in {Event.WEBSOCKET_SEND, Event.WEBSOCKET_CLOSE}:
            raise WebSocketRuntimeError(
                'Expected ASGI message "websocket.send" or "websocket.close", '
                f"but got {message_type!r}"
            )

        self.application_state = (
            WebSocketState.DISCONNECTED
            if message_type == Event.WEBSOCKET_CLOSE
            else self.application_state
        )
    await self._send(message)

accept async 

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

TYPE: str | None DEFAULT: None

headers

TYPE: Iterable[tuple[bytes, bytes]] | None DEFAULT: None

Source code in lilya/websockets.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def accept(
    self,
    subprotocol: str | None = None,
    headers: Iterable[tuple[bytes, bytes]] | None = None,
) -> None:
    headers = headers or []

    await self._connect()
    message = {
        "type": "websocket.accept",
        "headers": headers,
        "subprotocol": subprotocol,
    }
    await self.send(message)

receive_text async 

receive_text()
Source code in lilya/websockets.py 
126
127
128
129
130
131
async def receive_text(self) -> str:
    await self.raise_for_connection_state()

    message = await self.receive()
    self.raise_for_disconnect(message)
    return cast(str, message["text"])

receive_bytes async 

receive_bytes()
Source code in lilya/websockets.py 
133
134
135
136
137
138
async def receive_bytes(self) -> bytes:
    await self.raise_for_connection_state()

    message = await self.receive()
    self.raise_for_disconnect(message)
    return cast(bytes, message["bytes"])

receive_json async 

receive_json(mode='text')
PARAMETER DESCRIPTION
mode

TYPE: str DEFAULT: 'text'

Source code in lilya/websockets.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
async def receive_json(self, mode: str = "text") -> Any:
    if mode not in {MessageMode.TEXT, MessageMode.BINARY}:
        raise WebSocketRuntimeError('The "mode" argument should be "text" or "binary".')

    await self.raise_for_connection_state()

    message = await self.receive()
    self.raise_for_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 lilya/websockets.py 
155
156
157
158
159
160
async def iter_text(self) -> AsyncIterator[str]:
    try:
        while True:
            yield await self.receive_text()
    except WebSocketDisconnect:
        ...  # pragma: no cover

iter_bytes async 

iter_bytes()
Source code in lilya/websockets.py 
162
163
164
165
166
167
async def iter_bytes(self) -> AsyncIterator[bytes]:
    try:
        while True:
            yield await self.receive_bytes()
    except WebSocketDisconnect:
        ...  # pragma: no cover

iter_json async 

iter_json()
Source code in lilya/websockets.py 
169
170
171
172
173
174
async def iter_json(self) -> AsyncIterator[Any]:
    try:
        while True:
            yield await self.receive_json()
    except WebSocketDisconnect:
        ...  # pragma: no cover

send_text async 

send_text(data)
PARAMETER DESCRIPTION
data

TYPE: str

Source code in lilya/websockets.py 
176
177
async def send_text(self, data: str) -> None:
    await self.send({"type": Event.WEBSOCKET_SEND, "text": data})

send_bytes async 

send_bytes(data)
PARAMETER DESCRIPTION
data

TYPE: bytes

Source code in lilya/websockets.py 
179
180
async def send_bytes(self, data: bytes) -> None:
    await self.send({"type": Event.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 lilya/websockets.py 
182
183
184
185
186
187
188
189
190
191
192
async def send_json(self, data: Any, mode: str = "text") -> None:
    if mode not in {MessageMode.TEXT, MessageMode.BINARY}:
        raise WebSocketRuntimeError('The "mode" argument should be "text" or "binary".')

    text = json.dumps(data, separators=(",", ":"), ensure_ascii=False)

    (
        await self.send({"type": Event.WEBSOCKET_SEND, "text": text})
        if mode == MessageMode.TEXT
        else await self.send({"type": Event.WEBSOCKET_SEND, "bytes": text.encode("utf-8")})
    )

close async 

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

TYPE: int DEFAULT: 1000

reason

TYPE: str | None DEFAULT: None

Source code in lilya/websockets.py 
194
195
async def close(self, code: int = 1000, reason: str | None = None) -> None:
    await self.send({"type": Event.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 
15
16
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: IntEnum

CONNECTING class-attribute instance-attribute 

CONNECTING = 0

CONNECTED class-attribute instance-attribute 

CONNECTED = 1

DISCONNECTED class-attribute instance-attribute 

DISCONNECTED = 2