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:
  • server (str) –

  • handle (str) –

  • jid (slixmpp.JID) –

  • log (Optional[logging.Logger]) –

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:
  • room_id (str) –

  • limit (int) –

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_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]