Handlers

Handlers in MatrixCtl are used to handle the communication between the server and the Commands 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'>)[source]

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)[source]

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/runner/.config/matrixctl/config')]

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

static apply_defaults(server)[source]

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)[source]

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()[source]

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)[source]

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)[source]

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)[source]

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)[source]

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', 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))[source]

Bases: object

Build the URL for an API request.

concurrent_limit: int

content: str | bytes | Iterable[bytes] | None

data: dict[str, Any] | None

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: dict[str, Any] | None

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)[source]

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)[source]

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

Return type:

None

Attributes:

input_queueasyncio.Queue: The input queue, which provides the RequestBuilder.
output_queueasyncio.Queue: The output queue, which gets the responses of there requests.

Returns:

None

See also

RequestBuilder: matrixctl.handlers.api.RequestBuilder

async matrixctl.handlers.api.exec_async_request(request_config)[source]

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

Return type:

Response | list[Response]

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.)

Returns:

responseslist of httpx.Response or httpx.Response: Depending on concurrent_limit an request_config.

See also

RequestBuilder: matrixctl.handlers.api.RequestBuilder

matrixctl.handlers.api.generate_worker_configs(request_config, next_token, limit)[source]

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

Return type:

Generator[RequestBuilder, None, None]

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.

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!

async matrixctl.handlers.api.group_async_results(input_size, output_queue)[source]

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

Return type:

list[Exception | 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.

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.

matrixctl.handlers.api.handle_sync_response_status_code(response, success_codes=None)[source]

Handle the response status code of a synchronous request.

Return type:

None

Attributes:

responsehttps.Response: The response of the synchronous request.
success_codestuple of int, optional: A tuple of success codes. For example: 200, 201, 202, 203. If the parameter is not set, the default success codes are used.

Returns:

None: The function either returns None or raises an exception.

matrixctl.handlers.api.preplan_request_strategy(limit, concurrent_limit, max_step_size=100)[source]

Use this functiona as helper for optimizing asynchronous requests.

Return type:

RequestStrategy

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.

Returns:

RequestStrategymatrixctl.handlers.api.RequestStrategy: A Named tuple with the RequestStrategy values.

matrixctl.handlers.api.request(request_config)[source]

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

Return type:

list[Response] | Response

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.)

Returns:

responsehttpx.Response: Returns the response

See also

RequestBuilder: matrixctl.handlers.api.RequestBuilder

matrixctl.handlers.api.streamed_download(request_config, download_path)[source]

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

Return type:

None

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.)

Returns:

responsehttpx.Response: Returns the response

See also

RequestBuilder: matrixctl.handlers.api.RequestBuilder

Ansible

Run a ansible playbook with this module.

matrixctl.handlers.ansible.ansible_run(playbook, tags=None, extra_vars=None)[source]

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)[source]

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)[source]

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()[source]

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)[source]

Bases: object

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

address: str

port: int

run_cmd(cmd)[source]

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)[source]

Bases: NamedTuple

Store the response of a SSH command as response.

stderr: str | None: Alias for field number 2

stdin: str | None: Alias for field number 0

stdout: str | None: Alias for field number 1

Table

Use this handler to generate and print tables.

matrixctl.handlers.table.cells_to_str(part, none)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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')[source]

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)[source]

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)[source]

Create an SSH tunnel.

Return type:

Iterator[int | None]

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