matteridge.websocket

Attributes

log

Exceptions

WebsocketRequestError

Common base class for all non-exit exceptions.

Classes

Module Contents

exception matteridge.websocket.WebsocketRequestError[source]

Bases: Exception

Common base class for all non-exit exceptions.

Initialize self. See help(type(self)) for accurate signature.

class matteridge.websocket.Websocket(url, token, client)[source]
Parameters:

client (aiohttp.ClientSession)

SSL_VERIFY = True[source]
HEART_BEAT = 30[source]
MIN_RETRY_TIME = 3[source]
MAX_RETRY_TIME = 300[source]
REQUEST_TIMEOUT = 30[source]
token[source]
url[source]
_connected = False[source]
_connecting[source]
_ws: aiohttp.ClientWebSocketResponse | None = None[source]
_futures: dict[int, asyncio.Future[dict]][source]
_seq_cursor = 1[source]
_aiohttp_client[source]
_attempts = 0[source]
__context[source]
async connect(event_handler)[source]

Connect to the websocket and authenticate it. When the authentication has finished, start the loop listening for messages, sending a ping to the server to keep the connection alive. :param event_handler: Every websocket event will be passed there. Takes one argument. :type event_handler: Function(message)

async _connect(event_handler)[source]
async _initial_connect_sequence(ws)[source]
Parameters:

ws (aiohttp.ClientWebSocketResponse)

async listen(ws, event_handler)[source]
Parameters:

ws (aiohttp.ClientWebSocketResponse)

async __request(action, data=None)[source]
Parameters:
  • action (str)

  • data (Optional[dict])

async user_typing(channel_id)[source]
async get_statuses()[source]
Returns:

A dict mapping user_ids to statuses

Return type:

dict[str, matteridge.events.StatusType]

async get_statuses_by_ids(ids)[source]
Returns:

A dict mapping user_ids to statuses

Parameters:

ids (list[str])

Return type:

dict[str, matteridge.events.StatusType]

matteridge.websocket.log[source]