orionis.http.adapters.websocket.rsgi.RSGIWebSocketTransport

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 RSGIWebSocketTransport(IWebSocketTransport): (source)

Constructor: RSGIWebSocketTransport(protocol)

View In Hierarchy

Adapt Granian's actual RSGI WebsocketProtocol and accepted transport.

Method	`__init__`	Retain the pending server protocol.
Method	`_acceptedTransport`	Require the transport returned by a completed handshake.
Async Method	`accept`	Accept through Granian's argument-free handshake method.
Async Method	`close`	Close through Granian's default connection closure API.
Async Method	`receive`	Normalize Granian's close, bytes and string message kinds.
Async Method	`reject`	Reject the handshake using the actual HTTP status argument.
Async Method	`sendBytes`	Await Granian's binary write without buffering.
Async Method	`sendText`	Await Granian's text write without buffering.
Class Variable	`__slots__`	Undocumented
Instance Variable	`_protocol`	Undocumented
Instance Variable	`_transport`	Undocumented
Property	`supportsCloseDetails`	Report the documented RSGI close-frame limitation.

def __init__(self, protocol: RSGIWebsocketProtocol): (source) ¶

Retain the pending server protocol.

Parameters
protocol:`WebsocketProtocol`	Granian's handshake and close interface.

Returns
`None`	Prepare a transport that will be bound during acceptance.

def _acceptedTransport(self) -> RSGIWebsocketTransport: (source) ¶

Require the transport returned by a completed handshake.

Returns
`RSGIWebsocketTransport`	Granian's active message transport.

Raises
`RuntimeError`	If called before acceptance completed.

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 through Granian's argument-free handshake method.

Parameters
subprotocol:`str \| None, optional`	Unsupported by this RSGI version.
headers:`Mapping[str, str] \| None, optional`	Unsupported by this RSGI version.

Returns
`None`	Store the returned accepted transport.

Raises
`NotImplementedError`	If unsupported handshake options are requested.
`WebSocketDisconnected`	If the server reports an already closed connection.

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

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

Close through Granian's default connection closure API.

Parameters
code:`int, optional`	Only normal closure is supported.
reason:`str, optional`	Only the empty reason is supported.

Returns
`None`	Invoke synchronous connection closure.

Raises
`NotImplementedError`	If a custom wire close code or reason is requested.

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

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

Normalize Granian's close, bytes and string message kinds.

Returns
`WebSocketMessage`	Message with unknown peer close details represented as code 1006.

Raises
`RuntimeError`	If the server supplies an invalid message kind or payload.
`WebSocketDisconnected`	If the server reports a transport disconnection.

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

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

Reject the handshake using the actual HTTP status argument.

Parameters
status_code:`int, optional`	HTTP rejection status.

Returns
`None`	Invoke Granian's synchronous close method.

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

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

Await Granian's binary write without buffering.

Parameters
data:`bytes`	Complete binary message.

Returns
`None`	Finish the server write.

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

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

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

Await Granian's text write without buffering.

Parameters
data:`str`	Complete text message.

Returns
`None`	Finish the server write.

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

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

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

Undocumented

_protocol = (source) ¶

Undocumented

_transport: RSGIWebsocketTransport | None = (source) ¶

Undocumented

@property

supportsCloseDetails: bool = (source) ¶

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

Report the documented RSGI close-frame limitation.

Returns
`bool`	False because `protocol.close(status)` controls HTTP rejection.