matridge.matrix

Module Contents

Classes

Credentials

dict() -> new empty dictionary

AuthenticationClient

An async IO matrix client.

Client

An async IO matrix client.

Functions

catch_all(coro)

matridge.matrix.catch_all(coro)[source]
Parameters:

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

class matridge.matrix.Credentials[source]

Bases: TypedDict

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object’s

(key, value) pairs

dict(iterable) -> new dictionary initialized as if via:

d = {} for k, v in iterable:

d[k] = v

dict(**kwargs) -> new dictionary initialized with the name=value pairs

in the keyword argument list. For example: dict(one=1, two=2)

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

homeserver: str[source]
user_id: str[source]
device_id: str[source]
access_token: str[source]
class matridge.matrix.AuthenticationClient(server, handle, jid, log=None)[source]

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:
save(resp)[source]
Parameters:

resp (nio.LoginResponse)

load()[source]
destroy()[source]
async login_token()[source]
async fix_homeserver()[source]

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

class matridge.matrix.Client(server, handle, session)[source]

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:
MIN_RETRY_TIME = 3[source]
MAX_RETRY_TIME = 300[source]
REQUEST_TIMEOUT = 60[source]
CONSIDER_SUCCESSFUL = 10[source]
__add_event_handlers()[source]
async __get_muc(room)[source]
Parameters:

room (Union[nio.MatrixRoom, str])

Return type:

matridge.group.MUC

async __sync_forever()[source]
async get_participant(room, event)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.Event)

Return type:

matridge.group.Participant

async listen()[source]
stop_listen()[source]
async fetch_history(room_id, limit)[source]
Parameters:
async on_event(room, event)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.Event)

async on_message(room, event)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.RoomMessage)

async on_presence(presence)[source]
Parameters:

presence (nio.PresenceEvent)

async on_avatar(room, event)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.RoomAvatarEvent)

async on_topic(room, event)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.RoomTopicEvent)

async on_name(room, event)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.RoomNameEvent)

async on_sticker(room, event)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.StickerEvent)

async on_member(room, event)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.RoomMemberEvent)

async on_typing(room, event)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.TypingNoticeEvent)

async on_receipt(room, event)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.ReceiptEvent)

async on_reaction(room, event, **kw)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.ReactionEvent)

async on_redact(room, event)[source]
Parameters:

  • room (nio.MatrixRoom)

  • event (nio.RedactionEvent)

async get_original_id(room_id, event_id)[source]
Parameters:

  • room_id (str)

  • event_id (str)

Return type:

str

async get_event(room_id, event_id)[source]
Parameters:

  • room_id (str)

  • event_id (str)

Return type:

Optional[nio.Event]