matrixlib.client module

General Matrix client functionality

class matrixlib.client.Callbacks

Client callback functions

re_log_in: Callable[[Client], Awaitable[None]]

class matrixlib.client.Client(storage: Storage, callbacks: Callbacks = {}, base_client_url: str | None = None, well_known: dict[str, Any] | None = None, versions: dict[str, Any] | None = None)

Represents a client connection

Arguments:

storage:

The storage object to use

callbacks:

Functions that will be called to perform processing needed by the client

base_client_url:

The base URL to use to connect to the server, including the _matrix/client/ suffix. If omitted, the client must have previously logged in and stored the base URL in the storage

well_known:

The contents of the .well-known file obtained from server discovery.

versions:

The result of calling /_matrix/client/versions on the server, for example as returned by the discover function.

async authenticated(req_func, *args, **kwargs) → ClientResponse

Make an authenticated request to the homeserver.

Arguments:

req_func:

the function to call to make the request. e.g. to make an authenticated GET request, this would be self.http_session.get.

*args, **kwargs:

the arguments to pass to req_func.

async claim_otks(algorithm: str, devices: dict[str, list[str]], timeout: int | None = None) → Tuple[dict[str, dict[str, Tuple[str, Any]]], list[str]]

Claim one-time keys for the given devices.

Arguments:

algorithm:

the key algorithm for the one-time keys that we want to request

devices:

the devices that we want to request one-time keys for. This is given as a dict, mapping user IDs to a list of device IDs

timeout:

how long to wait, in milliseconds, for a response from remote servers

Returns a tuple consisting of:

• a dict mapping user IDs to device IDs to (key ID, one-time key) tuple. A requested device may be missing, indicating that no one-time key for that device could be obtained

• a list of remote servers that failed to respond in time

async close() → None

Free up resources used by the client.

property device_id: str | None

The device ID of the logged-in user, or None if not logged in

async log_in_with_password(identifier: str | dict[str, Any], password: str, identifier_type: str = 'user', **kwargs: dict[str, Any]) → dict[str, Any]

Log in to the server

Arguments:

identifier:

the identifier to use. The format depends on the identifier type, but in most cases will be a string.

password:

the password to use.

identifier_type:

the type of identifer to use. The default, 'user' indicates that the identifier is a username or user ID. Other recognized values are 'email' and 'phone'. Any other values will be passed unchanged to the login endpoint. For 'phone', identifier is either a dict with country (the two-letter country code that the phone number is from) and phone (the actual phone number) properties, or a the canonicalised phone number as a string. For unknown values, the identifier should be a dict.

async log_out() → None

Log out the current session and clear out the storage.

async login_methods() → list[str]

Get a list of the login methods supported by the server.

make_txn_id() → str

Generate a unique transaction ID

async re_log_in(auth_parameter: str | dict[str, Any], login_type: str = 'password') → None

Re-log in after getting soft logged out

Arguments:

auth_parameter:

The authentication parameter (e.g. password or token) to use.

login_type:

The method of logging in. The default value, 'password', indicates that you are logging in with a password. A value of 'token' indicates that you are logging in with a token. An unknown value will set that as the type in the request body, and auth_parameter must be a dict whose contents will be copied to the request body.

async send_event(room_id: str, event_type: str, event_content: dict[str, Any], retry_ms: int = 0, txn_id: str | None = None) → Tuple[str, str]

Send an event to a room

Arguments:

room_id:

the ID of the room to send to

event_type:

the type of event to send

event_content:

the content of the event

retry_ms:

how long to retry sending, in milliseconds

txn_id:

the transaction ID to use. If none is specified, one is generated.

async send_to_device(event_type: str, event_contents: dict[str, dict[str, dict]], retry_ms: int = 0, txn_id: str | None = None) → None

Send an events to devices

Arguments:

room_id:

the ID of the room to send to

event_type:

the type of event to send

event_contents:

a map from user ID to device ID to event content. If “*” is used as the device ID, the event is sent to all of the user’s devices.

retry_ms:

how long to retry sending, in milliseconds

start_sync()

Start the sync loop

stop_sync()

Stop the sync loop

url(endpoint: str) → str

Returns the full URL of a given endpoint

property user_id: str | None

The user ID of the logged-in user, or None if not logged in

class matrixlib.client.ClientClosed

A message indicating that the client is closed

class matrixlib.client.DeviceChanges(changed: list[str], left: list[str])

A message indicating that user’s devices have changed

Create new instance of DeviceChanges(changed, left)

changed: list[str]

Users whose devices have changed

left: list[str]

Users who no longer share a room

class matrixlib.client.LeftRoom(room_id: str)

A message indicating that the user has left a room

Create new instance of LeftRoom(room_id,)

room_id: str

The ID of the room that the user left

class matrixlib.client.OneTimeKeysCount(otk_count: dict[str, int], has_to_device: bool)

A message indicating the one-time keys count from the sync

Create new instance of OneTimeKeysCount(otk_count, has_to_device)

has_to_device: bool

Whether any to-device messages were in the sync

otk_count: dict[str, int]

Dict mapping algorithm name to one-time keys count

class matrixlib.client.RoomStateUpdates(room_id: str, events: list[matrixlib.events.StateEvent])

A message giving state updates for a room from sync

Create new instance of RoomStateUpdates(room_id, events)

static create_from_sync(room_id: str, state_updates: dict) → RoomStateUpdates

events: list[matrixlib.events.StateEvent]

The state events

room_id: str

The room ID

class matrixlib.client.RoomTimelineUpdates(room_id: str, events: list[matrixlib.events.RoomEvent], limited: bool, prev_batch: str | None)

A message giving timeline for a room from sync

Create new instance of RoomTimelineUpdates(room_id, events, limited, prev_batch)

static create_from_sync(room_id: str, timeline_updates: dict) → RoomTimelineUpdates

events: list[matrixlib.events.RoomEvent]

The state events

limited: bool

Whether the timeline update was limited

prev_batch: str | None

The token to retrieve previous messages

room_id: str

The room ID

class matrixlib.client.Storage(*args, **kwargs)

Protocol for classes that persist data for the client

clear() → None

get(key: str, default: Any = None) → Any

class matrixlib.client.SyncFailed(delay: int)

A message indicating that a sync call has failed

Create new instance of SyncFailed(delay,)

delay: int

The amount of time that the loop will wait before retrying

class matrixlib.client.SyncResumed

A message indicating that the sync loop has resumed after a previous failure

class matrixlib.client.ToDeviceEvents(events: list[matrixlib.events.Event])

A message indicating that to-device events have been received

Create new instance of ToDeviceEvents(events,)

events: list[matrixlib.events.Event]

The to-device events

async matrixlib.client.check_response(resp: ClientResponse) → tuple[int, dict[str, Any]]

Checks whether an HTTP response is an error.

If successful, returns the HTTP code and the response body. Raises an exception on error.

async matrixlib.client.discover(server_name: str) → tuple[str, dict, dict] | None

Perform discovery on the given server name.

On success, returns a tuple consisting of homeserver’s base client-server URL (the homeserver’s base URL with _matrix/client appended), the full contents of the JSON object returned from the .well-known file, and the full contents of the server’s /_matrix/client/versions response. On failure, returns None.

async matrixlib.client.retry(limit: int, req_func, *args, **kwargs) → ClientResponse

Retry a request until a time limit is reached.

The request will be retried if a non-Matrix response is received, or if the request was rate limited.

Note that this will not apply a timeout to the actual request: if the server responds slowly, then it may exceed the retry time limit. To limit the time taken by the request, a timeout argument should be passed to the request function.

Arguments:

limit:

the time limit, in milliseconds

req_func:

the function to call to make the request. e.g. to make a GET request, this could be session.get, where session is an aiohttp.ClientSession.

*args, **kwargs:

the arguments to pass to req_func.