Handlers

Handlers in MatrixCtl are used to handle the communication between the server and the Addons or to load config files.

YAML

Read and parse the configuration file with this module.

class matrixctl.handlers.yaml.JinjaUndefined(hint=None, obj=missing, name=None, exc=<class 'jinja2.exceptions.UndefinedError'>)

Bases: Undefined

Use this class as undefined argument in a Jinja2 Template.

The class replaces every undefined template with an enpty string.

class matrixctl.handlers.yaml.YAML(paths=None, server=None)

Bases: object

Use the YAML class to read and parse the configuration file(s).

DEFAULT_PATHS: ClassVar[list[Path]] = [PosixPath('/etc/matrixctl/config'), PosixPath('/home/docs/.config/matrixctl/config')]

JINJA_PREDEFINED: ClassVar[dict[str, str | int]] = {'default_api_concurrent_limit': 4, 'default_ssh_port': 22, 'home': '/home/docs', 'user': 'docs'}

static apply_defaults(server)

Apply defaults to the configuration.

Return type:

ConfigServer

Parameters:

servermatrixctl.structures.ConfigServer

The configuration of a (home)server.

Returns:

servermatrixctl.structures.ConfigServer

The configuration of a (home)server with applied defaults.

get(*keys)

Get a value from a config entry safely.

Usage

Pass strings, describing the path in the self.__yaml dictionary. Let’s say, you are looking for the synapse path:

Return type:

Any

Parameters:

*keysstr

A tuple of strings describing the values you are looking for.

Returns:

answerany

The value of the entry you described.

Examples

from matrixctl.handlers.yaml import YAML

yaml: YAML = YAML()
port: int = yaml.get("server", "ssh", "port")
print(port)
# Output: 22

static get_paths_to_config()

Generate a tuple of path which may contain a configuration file. :rtype: tuple[Path, ...]

Note

This function preserves the order. The priority of the user configuration in XDG_CONFIG_HOME is higher than the global configuration in /etc/matrixctl/. The priority of the file extension yaml is greater than the priority of the file extension yml.

Warning

The paths returned by this function might not exist.

Returns:

config_pathstuple of pathlib.Path

A tuple of paths, which might contain a config file.

get_server_config(paths, server)

Read and concentrate the config in one dict.

The servers: ... will be removed form the dict. A new entry server will be created, which represents the selected server.

Return type:

Config

Parameters:

pathsIterable of pathlib.Path

The paths to the configfiles.

serverstr

The selected server. (Default: “default”)

Returns:

server_configmatrixctl.typehints.Config

The config for the selected server.

Notes

When all files were empty or don’t exist, an empty dict will be returned.

static read_from_file(yaml, path)

Read the config from a YAML file and render the Jinja2 tmplates. :rtype: Config

Note

• The Renderer does one pass. This means, you can only render templated strings but not the templated string of another templated string.

• If the file was empty or does not exist, an empty dict will be returned.

Parameters:

yamlruamel.yaml.Yaml

The yaml object.

pathPath

The path where the config file is located.

Returns:

full_configmatrixctl.typehints.Config

The full (with server name) config file as dict.

server: str

matrixctl.handlers.yaml.secrets_filter(tree, key)

Redact secrets when printing the configuration file.

Return type:

Any

Parameters:

treedict [str, str]

A patrial of tree from tree_printer. (Can only be this type) afterwards.

keystr

A dict key. (Can only be this type)

Returns:

None

matrixctl.handlers.yaml.tree_printer(tree, depth=0)

Print the configuration file recursively.

Return type:

None

Parameters:

treeany

Initial a matrixctl.typehints.Config and partials of it afterwards.

depthint

The depth of the table

Returns:

None

API

Get access to the API of your homeserver.

class matrixctl.handlers.api.RequestBuilder(token, domain, path, scheme='https', subdomain='matrix', api_path='_synapse/admin', api_version='v2', data=None, json=None, content=None, method='GET', params={}, headers={}, concurrent_limit=4, timeout=5.0, success_codes=(200, 201, 202, 203, 204, 205, 206, 207, 226))

Bases: object

Build the URL for an API request.

api_path: str

api_version: str

concurrent_limit: int

content: UnionType[str, bytes, Iterable[bytes], None]

data: Optional[dict[str, Any]]

domain: str

headers: dict[str, str]

property headers_with_auth: dict[str, str]

Get the headers with bearer token.

Parameters:

None

Returns:

headersdict [str, str]

Headers with auth. token.

json: Optional[dict[str, Any]]

method: str

params: dict[str, str | int]

path: str

scheme: str

subdomain: str

success_codes: tuple[int, ...]

timeout: float

token: str

class matrixctl.handlers.api.RequestStrategy(limit: int, step_size: int, concurrent_limit: int, offset: int, iterations: int)

Bases: NamedTuple

Use this NamedTuple as request strategy data.

This NamedTuple is only used in this module.

concurrent_limit: int

Alias for field number 2

iterations: int

Alias for field number 4

limit: int

Alias for field number 0

offset: int

Alias for field number 3

step_size: int

Alias for field number 1

async matrixctl.handlers.api.async_worker(input_queue, output_queue)

Use this coro as worker to make (a)synchronous request.

Return type:

None

Returns:

None

See also

RequestBuilder

matrixctl.handlers.api.RequestBuilder

Attributes:

input_queueasyncio.Queue

The input queue, which provides the RequestBuilder.

output_queueasyncio.Queue

The output queue, which gets the responses of there requests.

async matrixctl.handlers.api.exec_async_request(request_config)

Use this coro to generate and run workers and group the responses.

Return type:

Response | list[Response]

Returns:

responseslist of httpx.Response or httpx.Response

Depending on concurrent_limit an request_config.

See also

RequestBuilder

matrixctl.handlers.api.RequestBuilder

Attributes:

request_configRequestBuilder or Generator [RequestBuilder, None, None]

An instance of an RequestBuilder or a list of RequestBuilder. If the function gets a RequestBuilder, the request will be synchronous. If it gets a Generator, the request will be asynchronous.

concurrent_limitint

The maximum of concurrent workers. (This information must be pulled from the config.)

matrixctl.handlers.api.generate_worker_configs(request_config, next_token, limit)

Create workers for async requests (minus the already done sync request).

Return type:

Generator[RequestBuilder, None, None]

Yields:

request_configmatrixctl.handlers.api.RequestBuilder

Yields a fully configured RequestsBuilder for every request that has to be done to get all entries.

Notes

Warning Call-By-Reference like behavior! The param limit and the concurrent_limit in request_config will get changed in this function. Make sure to only use them after using this function!

Attributes:

request_configmatrixctl.handlers.api.RequestBuilder

An instance of an RequestBuilder from which was used for an initial synchronous request to get the first part of the data and the other two arguments from the response.

next_tokenint

The value, which defines from where to start in the next request. You get this value from the response of an initial synchronous request.

totalint

The value which defines how many entries there are. You get this value from the response of an initial synchronous request.

async matrixctl.handlers.api.group_async_results(input_size, output_queue)

Use this coro to group the requests afterwards in a single list.

Return type:

list[Exception | Response]

Returns:

responseslist of httpx.Response or httpx.Response

Depending on concurrent, it is a httpx.Response if concurrent is true, otherwise it is a list of httpx.Response.

Attributes:

input_sizeint

The number of items in the queue.

output_queueasyncio.Queue

The output queue, which holds the responses of there requests.

concurrentbool

When True, make requests concurrently. When False, make requests synchronously.

matrixctl.handlers.api.preplan_request_strategy(limit, concurrent_limit, max_step_size=100)

Use this functiona as helper for optimizing asynchronous requests.

Return type:

RequestStrategy

Returns:

RequestStrategymatrixctl.handlers.api.RequestStrategy

A Named tuple with the RequestStrategy values.

Attributes:

limitint

A user entered limit or total.

concurrent_limit: int

The concurrent limit from the config file.

max_step_sizeint, default=100

The maximal step size, which is a soft limit. It is usually 100, but that value might be different. Check out the API documentation. We usually take the default one.

matrixctl.handlers.api.request(request_config)

Make a (a)synchronous request to the synapse API and receive a response.

Return type:

list[Response] | Response

Returns:

responsehttpx.Response

Returns the response

See also

RequestBuilder

matrixctl.handlers.api.RequestBuilder

Attributes:

request_configRequestBuilder or Generator [RequestBuilder, None, None]

An instance of an RequestBuilder or a list of RequestBuilder. If the function gets a RequestBuilder, the request will be synchronous. If it gets a Generator, the request will be asynchronous.

concurrent_limitint

The maximum of concurrent workers. (This information must be pulled from the config.)

Ansible

Run a ansible playbook with this module.

matrixctl.handlers.ansible.ansible_run(playbook, tags=None, extra_vars=None)

Run an ansible playbook.

Return type:

None

Parameters:

playbookpathlib.Path

The path to the ansible Playbook

tagsstr, optional

The tags to use

extra_varsdict [str, str], optional

The extra_vars to use.

Returns:

None

Git (VCS)

Update and manage the synapse playbook repository with this module.

class matrixctl.handlers.vcs.VCS(path)

Bases: object

Update and manage a repository.

property datetime_last_pulled_commit: datetime

Get the datetime the commit was pulled last from git.

This is used to determine which messages will be produced in the table.

Parameters:

None

Returns:

datetimedatetime.datetime

The datetime object.

log(since=None)

Print a table of date, user and commit message since the last pull.

Return type:

None

Parameters:

sincedatetime.datetime, optional, default=None

The datetime the last commit was puled.

Returns:

None

pull()

Git pull the latest commits from GH.

Return type:

None

Parameters:

None

Returns:

None

SSH

Run and evaluate commands on the host machine of your synapse server.

class matrixctl.handlers.ssh.SSH(address, user=None, port=22)

Bases: object

Run and evaluate commands on the host machine of your synapse server.

address: str

port: int

run_cmd(cmd)

Run a command on the host machine and receive a response.

Return type:

SSHResponse

Parameters:

cmdstr

The command to run.

ttybool

Request a pseudo-terminal from the server (default: False)

Returns:

responsematrixctl.handlers.ssh.SSHResponse

Receive stdin, stdout and stderr as response.

user: str

class matrixctl.handlers.ssh.SSHResponse(stdin: str | None, stdout: str | None, stderr: str | None)

Bases: NamedTuple

Store the response of a SSH command as response.

stderr: Optional[str]

Alias for field number 2

stdin: Optional[str]

Alias for field number 0

stdout: Optional[str]

Alias for field number 1

Table

Use this handler to generate and print tables.

matrixctl.handlers.table.cells_to_str(part, none)

Convert all cells to strings and format None values.

Return type:

list[list[str]]

Parameters:

partcollections.abc.Sequence of collections.abc.Sequence of str

Data or header, in which every cell will be to casted to to strings.

nonestr

A string, which is used to replace None with the specific string.

Returns:

partlist of list of str

The part, where every cell is of type str.

matrixctl.handlers.table.find_newlines(data)

Find newlines and return a dict with positions (key) and occurrences.

Return type:

dict[int, int]

Parameters:

datalist of str

Data or headers of the table.

Returns:

posdict [int, int]

A dictionary {r: n}, where n are newlines in row r.

Notes

The function only adds an entry to the dict, if there is at least one newline in a row.

matrixctl.handlers.table.format_table_row(line, max_column_len)

Format a table row into a str.

Return type:

str

Parameters:

linelist of str

A data or headers row, which will be formatted to a string.

max_column_lentuple of int

A n-tuple which describes the longest string per column. (n is the number of columns)

Returns:

row_stringstr

A formatted string, which represents a table row.

matrixctl.handlers.table.get_colum_length(data, headers)

Transpose rows and find longest line.

Return type:

tuple[int, ...]

Parameters:

datalist of list of str

The data part of the table.

headersNone or list of list of str

The headers part of the table.

Returns:

column_length_tupletuple of int

A n-tuple which describes the longest string per column. (n is the number of columns)

matrixctl.handlers.table.handle_newlines(part, newlines)

Update and insert new lines.

Return type:

tuple[list[list[str]], set[int]]

Parameters:

partlist of list of str

Data or headers of the table.

newlinesdict [int, int]

A dictionary {r: n}, where n are newlines in row r.

Returns:

partlist of list of str

The part contains the supplemented and updated rows.

inhibit_sepset of int

The inhibit_sep set contains the line numbers where a separator yhould be inhibited because the lines handled by this function are belonging together.

matrixctl.handlers.table.newlines_in_row(row)

Get the highest number of newlines per row.

The highest number of newlines for a row is used to determine in how many rows the row gets expanded, to get one row per newline - 1.

Return type:

int

Parameters:

rowlist of str

Data or headers of the table.

Returns:

max_newlinesint

The highest number of newlines ber row.

matrixctl.handlers.table.table(table_data, table_headers=None, none='-', *, sep=True)

Create a table from data and a optional headers.

Return type:

Generator[str, None, None]

Parameters:

table_datacollections.abc.Sequence of collections.abc.Sequence of str

Data.

table_headerscollections.abc.Sequence of str, Optional

Headers.

sepbool, default = True

True, when there should be a separator between every row of data.

nonestr, default = “-”

A string, which is used to replace None with the specific string.

Yields:

tablestr

The table (row for row).

matrixctl.handlers.table.transpose_newlines_to_rows(split, occurrences)

Transpose newlines in new rows.

Return type:

Generator[list[str], None, None]

Parameters:

splitlist of list of str

A list of substring-lists, split from one row, which contains newline characters. The substing-lists are containing strings, which have been split into substings.

occurrencesint

The maximal number of newlines across the row.

Yields:

rowlist of str

A row for each occurrence.

Database

Talk to the the database.

class matrixctl.handlers.db.DBConnectionBuilder(host: str, database: str, username: str, password: str, port: int = 5432, timeout: int = 10, scheme: str = 'postgresql')

Bases: NamedTuple

Build the URL for an API request.

database: str

Alias for field number 1

host: str

Alias for field number 0

password: str

Alias for field number 3

port: int

Alias for field number 4

scheme: str

Alias for field number 6

timeout: int

Alias for field number 5

username: str

Alias for field number 2

matrixctl.handlers.db.db_connect(yaml)

Connect to a PostgreSQL database.

Return type:

Iterator[Connection]

Parameters:

yamlmatrixctl.handlers.yaml.YAML

The configuration file handler.

Yields:

connpsycopg.Connection

A new Connection instance.

matrixctl.handlers.db.ssh_tunnel(host, username, remote_port, port=22, *, enabled=True)

Create an SSH tunnel.

Return type:

Iterator[Optional[int]]

Parameters:

hoststr

The remote host e.g. 127.0.0.1 or host.domain.tld.

usernamestr

The username of the user.

remote_portint

The port of the application, which should be tunneled.

enabledbool, default: True

True if the tunnel should be enabled or False if not.

portint, default: 22

The ssh port

private_keyPath or str, optioal

The path to the private key (Currently Disabled)

Yields:

tunint

The remote port

NoneNone

Yields none, when the tunnel is disabled (enabled = False).

Notes

The tunnel will only be created, when it is enabled. If the tunnel is disabled (enabled = False), the function will yield None instead of the local bind port.

Examples

with ssh_tunnel("127.0.0.1", myuser, 5432) as remote_port:
    print(f"The local bind port is: {local_bind_port}")
# The local bind port is: 8765