matridge.matrix

Classes

AuthenticationClient	An async IO matrix client.
Client	An async IO matrix client.

Functions

catch_all(coro)

Module Contents

matridge.matrix.catch_all(coro)

Parameters:

coro (Callable[[Client, nio.MatrixRoom, nio.Event], Awaitable[None]])

class matridge.matrix.AuthenticationClient(server, handle, jid, log=None)

Bases: nio.AsyncClient

An async IO matrix client.

Args:

homeserver (str): The URL of the homeserver which we want to connect

to.

user (str, optional): The user which will be used when we log in to the

homeserver.

device_id (str, optional): An unique identifier that distinguishes

this client instance. If not set the server will provide one after log in.

store_path (str, optional): The directory that should be used for state

storage.

config (AsyncClientConfig, optional): Configuration for the client. ssl (bool/ssl.SSLContext, optional): SSL validation mode. None for

default SSL check (ssl.create_default_context() is used), False for skip SSL certificate validation connection.

proxy (str, optional): The proxy that should be used for the HTTP

connection. Supports SOCKS4(a), SOCKS5, HTTP (tunneling) via an URL like e.g. ‘socks5://user:password@127.0.0.1:1080’.

Attributes:

synced (Event): An asyncio event that is fired every time the client

successfully syncs with the server. Note, this event will only be fired if the sync_forever() method is used.

A simple example can be found bellow.

Example:

>>> client = AsyncClient("https://example.org", "example")
>>> login_response = loop.run_until_complete(
>>>     client.login("hunter1")
>>> )
>>> asyncio.run(client.sync_forever(30000))

This example assumes a full sync on every run. If a sync token is provided for the since parameter of the sync_forever method full_state should be set to True as well.

Example:

>>> asyncio.run(
>>>     client.sync_forever(30000, since="token123",
>>>                         full_state=True)
>>> )

The client can also be configured to store and restore the sync token automatically. The full_state argument should be set to True in that case as well.

Example:

>>> config = ClientConfig(store_sync_tokens=True)
>>> client = AsyncClient("https://example.org", "example",
>>>                      store_path="/home/example",
>>>                      config=config)
>>> login_response = loop.run_until_complete(
>>>     client.login("hunter1")
>>> )
>>> asyncio.run(client.sync_forever(30000, full_state=True))

Parameters:

• server (str)

• handle (str)

• jid (slixmpp.JID)

• log (Optional[logging.Logger])

get_credentials(resp)

Parameters:

resp (nio.LoginResponse)

Return type:

matridge.util.Credentials

async fix_homeserver()

Uses https://$HOMESERVER/.well-known/matrix/client to fix the homeserver URL.

class matridge.matrix.Client(server, handle, session)

Bases: AuthenticationClient

An async IO matrix client.

Args:

homeserver (str): The URL of the homeserver which we want to connect

to.

user (str, optional): The user which will be used when we log in to the

homeserver.

device_id (str, optional): An unique identifier that distinguishes

this client instance. If not set the server will provide one after log in.

store_path (str, optional): The directory that should be used for state

storage.

config (AsyncClientConfig, optional): Configuration for the client. ssl (bool/ssl.SSLContext, optional): SSL validation mode. None for

default SSL check (ssl.create_default_context() is used), False for skip SSL certificate validation connection.

proxy (str, optional): The proxy that should be used for the HTTP

connection. Supports SOCKS4(a), SOCKS5, HTTP (tunneling) via an URL like e.g. ‘socks5://user:password@127.0.0.1:1080’.

Attributes:

synced (Event): An asyncio event that is fired every time the client

successfully syncs with the server. Note, this event will only be fired if the sync_forever() method is used.

A simple example can be found bellow.

Example:

>>> client = AsyncClient("https://example.org", "example")
>>> login_response = loop.run_until_complete(
>>>     client.login("hunter1")
>>> )
>>> asyncio.run(client.sync_forever(30000))

This example assumes a full sync on every run. If a sync token is provided for the since parameter of the sync_forever method full_state should be set to True as well.

Example:

>>> asyncio.run(
>>>     client.sync_forever(30000, since="token123",
>>>                         full_state=True)
>>> )

The client can also be configured to store and restore the sync token automatically. The full_state argument should be set to True in that case as well.

Example:

>>> config = ClientConfig(store_sync_tokens=True)
>>> client = AsyncClient("https://example.org", "example",
>>>                      store_path="/home/example",
>>>                      config=config)
>>> login_response = loop.run_until_complete(
>>>     client.login("hunter1")
>>> )
>>> asyncio.run(client.sync_forever(30000, full_state=True))

Parameters:

• server (str)

• handle (str)

• session (matridge.session.Session)

MIN_RETRY_TIME = 3

MAX_RETRY_TIME = 300

REQUEST_TIMEOUT = 60

CONSIDER_SUCCESSFUL = 10

__sync_task: asyncio.Task | None = None

session

reactions

load()

async login_token()

__add_event_handlers()

async __get_muc(room)

Parameters:

room (Union[nio.MatrixRoom, str])

Return type:

matridge.group.MUC

async __sync_forever()

async get_participant(room, event)

Parameters:

• room (nio.MatrixRoom)

• event (nio.Event)

Return type:

matridge.group.Participant

async listen()

stop_listen()

async fetch_history(room_id, limit)

Parameters:

• room_id (str)

• limit (int)

Return type:

AsyncIterator[nio.Event]

async on_event(room, event)

Parameters:

• room (nio.MatrixRoom)

• event (nio.Event)

async on_message(room, event)

Parameters:

• room (nio.MatrixRoom)

• event (nio.RoomMessage)

async on_presence(presence)

Parameters:

presence (nio.PresenceEvent)

async on_avatar(room, event)

Parameters:

• room (nio.MatrixRoom)

• event (nio.RoomAvatarEvent)

async on_topic(room, event)

Parameters:

• room (nio.MatrixRoom)

• event (nio.RoomTopicEvent)

async on_name(room, event)

Parameters:

• room (nio.MatrixRoom)

• event (nio.RoomNameEvent)

async on_sticker(room, event)

Parameters:

• room (nio.MatrixRoom)

• event (nio.StickerEvent)

async on_member(room, event)

Parameters:

• room (nio.MatrixRoom)

• event (nio.RoomMemberEvent)

async on_typing(room, event)

Parameters:

• room (nio.MatrixRoom)

• event (nio.TypingNoticeEvent)

async on_receipt(room, event)

Parameters:

• room (nio.MatrixRoom)

• event (nio.ReceiptEvent)

async on_reaction(room, event, **kw)

Parameters:

• room (nio.MatrixRoom)

• event (nio.ReactionEvent)

async on_redact(room, event)

Parameters:

• room (nio.MatrixRoom)

• event (nio.RedactionEvent)

async get_original_id(room_id, event_id)

Parameters:

• room_id (str)

• event_id (str)

Return type:

str

async get_event(room_id, event_id)

Parameters:

• room_id (str)

• event_id (str)

Return type:

Optional[nio.Event]