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: list[Path] = [PosixPath('/etc/matrixctl/config'), PosixPath('/home/runner/.config/matrixctl/config')]

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

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: Union[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, Union[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, sep=True, none='-')

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, enabled=True, port=22)

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