orionis.http.adapters.websocket.asgi.ASGIWebSocketTransport

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

class documentation

class ASGIWebSocketTransport(IWebSocketTransport): (source)

Constructor: ASGIWebSocketTransport(scope, receive, send)

View In Hierarchy

Translate the ASGI WebSocket protocol into normalized operations.

Method	`__init__`	Retain callbacks and advertised WebSocket specification version.
Method	`_acceptHeaders`	Validate and encode handshake headers before sending any event.
Async Method	`_connectEvent`	Consume the server's initial handshake event once.
Async Method	`accept`	Accept using the server's supported handshake fields.
Async Method	`close`	Send a close event using fields supported by the advertised version.
Async Method	`receive`	Normalize one complete ASGI data or disconnect event.
Async Method	`reject`	Deny the handshake with an HTTP status when the extension exists.
Async Method	`sendBytes`	Send one binary message and await server backpressure.
Async Method	`sendText`	Send one text message and await server backpressure.
Class Variable	`__slots__`	Undocumented
Instance Variable	`_handshake_received`	Undocumented
Instance Variable	`_receive`	Undocumented
Instance Variable	`_scope`	Undocumented
Instance Variable	`_send`	Undocumented
Instance Variable	`_version`	Undocumented

Inherited from IWebSocketTransport:

Property supportsCloseDetails Declare whether custom close details can reach the wire.

def __init__(self, scope: dict, receive: Callable[[], Awaitable[dict]], send: Callable[[dict], Awaitable[None]]): (source) ¶

Retain callbacks and advertised WebSocket specification version.

Parameters
scope:`dict`	Original ASGI connection scope.
receive:`Callable`	Server event reader.
send:`Callable`	Server event writer.

Returns
`None`	Prepare the pending handshake.

def _acceptHeaders(self, headers: Mapping[str, str] | None) -> list[tuple[bytes, bytes]]: (source) ¶

Validate and encode handshake headers before sending any event.

Parameters
headers:`Mapping[str, str] \| None`	User supplied response headers.

Returns
`list[tuple[bytes, bytes]]`	Encoded lowercase header pairs.

Raises
`NotImplementedError`	If the server predates ASGI HTTP 2.1.
`ValueError`	If a header violates the handshake header rules.

async def _connectEvent(self): (source) ¶

Consume the server's initial handshake event once.

Returns
`None`	Validate that the peer is awaiting acceptance.

Raises
`WebSocketDisconnected`	If the peer disconnects before acceptance.
`RuntimeError`	If the server supplies an invalid initial event.

async def accept(self, *, subprotocol: str | None = None, headers: Mapping[str, str] | None = None): (source) ¶

overrides orionis.http.adapters.websocket.contracts.transport.IWebSocketTransport.accept

Accept using the server's supported handshake fields.

Parameters
subprotocol:`str \| None, optional`	Subprotocol offered by the client.
headers:`Mapping[str, str] \| None, optional`	Additional headers supported since ASGI HTTP 2.1.

Returns
`None`	Await server acceptance.

Raises
`ValueError`	If the selected subprotocol was not offered.
`WebSocketDisconnected`	If the peer disappears during acceptance.

async def close(self, code: int = 1000, reason: str = ''): (source) ¶

overrides orionis.http.adapters.websocket.contracts.transport.IWebSocketTransport.close

Send a close event using fields supported by the advertised version.

Parameters
code:`int, optional`	Valid WebSocket close code.
reason:`str, optional`	Reason sent on servers supporting ASGI HTTP 2.3 or later.

Returns
`None`	Await the close event delivery.

Raises
`WebSocketDisconnected`	If the peer closes before the event reaches the server.

async def receive(self) -> WebSocketMessage: (source) ¶

overrides orionis.http.adapters.websocket.contracts.transport.IWebSocketTransport.receive

Normalize one complete ASGI data or disconnect event.

Returns
`WebSocketMessage`	The payload or peer closure details.

Raises
`RuntimeError`	If the server supplies an invalid event or two payloads.
`WebSocketDisconnected`	If a transport read raises an operating system error.

async def reject(self, status_code: int = 403): (source) ¶

overrides orionis.http.adapters.websocket.contracts.transport.IWebSocketTransport.reject

Deny the handshake with an HTTP status when the extension exists.

Parameters
status_code:`int, optional`	Rejection status; servers without the extension always use 403.

Returns
`None`	Finish the denial exchange.

Raises
`WebSocketDisconnected`	If the peer closes while rejection is sent.

async def sendBytes(self, data: bytes): (source) ¶

overrides orionis.http.adapters.websocket.contracts.transport.IWebSocketTransport.sendBytes

Send one binary message and await server backpressure.

Parameters
data:`bytes`	Binary payload.

Returns
`None`	Finish the transport write.

Raises
`WebSocketDisconnected`	If the peer has closed the connection.

async def sendText(self, data: str): (source) ¶

overrides orionis.http.adapters.websocket.contracts.transport.IWebSocketTransport.sendText

Send one text message and await server backpressure.

Parameters
data:`str`	Text payload.

Returns
`None`	Finish the transport write.

Raises
`WebSocketDisconnected`	If the peer has closed the connection.

__slots__: tuple[str, ...] = (source) ¶

overrides orionis.http.adapters.websocket.contracts.transport.IWebSocketTransport.__slots__

Undocumented

_handshake_received: bool = (source) ¶

Undocumented

_receive = (source) ¶

Undocumented

_scope = (source) ¶

Undocumented

_send = (source) ¶

Undocumented

_version = (source) ¶

Undocumented