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

Bases: jinja2.runtime.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: list[Path] = [PosixPath('/etc/matrixctl/config'), PosixPath('/home/docs/.config/matrixctl/config')]
JINJA_PREDEFINED: dict[str, str | int] = {'default_api_concurrent_limit': 4, 'default_ssh_port': 22, 'home': '/home/docs', 'user': 'docs'}
static apply_defaults(server)[source]

Apply defaults to the configuration.

Parameters
servermatrixctl.structures.ConfigServer

The configuration of a (home)server.

Returns
servermatrixctl.structures.ConfigServer

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

:rtype:pyConfigServer
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:

Parameters
*keysstr

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

Returns
answerany

The value of the entry you described.

:rtype:pyAny

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.

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.

:rtype:pytuple[Path, ...]
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.

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.

:rtype:pyConfig

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.

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.

:rtype:pyConfig
server: str
matrixctl.handlers.yaml.secrets_filter(tree, key)[source]

Redact secrets when printing the configuration file.

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
:rtype:pyAny
matrixctl.handlers.yaml.tree_printer(tree, depth=0)[source]

Print the configuration file recursively.

Parameters
treeany

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

depthint

The depth of the table

Returns
None
:rtype:pyNone

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

Bases: object

Build the URL for an API request.

api_path: str
api_version: str
concurrent_limit: int
content: Optional[Union[str, bytes, collections.abc.Iterable[bytes]]]
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.

:rtype:pydict[str, str]
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)[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.

Returns
None
:rtype:pyNone

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: collections.abc.Generator[matrixctl.handlers.api.RequestBuilder, None, None]) list[httpx.Response][source]
async matrixctl.handlers.api.exec_async_request(request_config: matrixctl.handlers.api.RequestBuilder) httpx.Response

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

Returns
responseslist of httpx.Response or httpx.Response

Depending on concurrent_limit an request_config.

rtype

httpx.Response | list[httpx.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.)

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

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

Yields
request_configmatrixctl.handlers.api.RequestBuilder

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

:rtype:pyGenerator[RequestBuilder, None, None]

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

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

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.

rtype

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

Use this functiona as helper for optimizing asyncronous requests.

Returns
RequestStrategymatrixctl.handlers.api.RequestStrategy

A Named tuple with the RequestStrategy values.

rtype

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 usualy 100, but that value might be different. Check out the API documentation. We usualy take the default one.

matrixctl.handlers.api.request(request_config: collections.abc.Generator[matrixctl.handlers.api.RequestBuilder, None, None]) list[httpx.Response][source]
matrixctl.handlers.api.request(request_config: matrixctl.handlers.api.RequestBuilder) httpx.Response

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

Returns
responsehttpx.Response

Returns the response

rtype

list[httpx.Response] | httpx.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)[source]

Run an ansible playbook.

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
rtype

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

:rtype:pydatetime
log(since=None)[source]

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

Parameters
sincedatetime.datetime, optional, default=None

The datetime the last commit was puled.

Returns
None
rtype

None ..

pull()[source]

Git pull the latest commits from GH.

Parameters
None
Returns
None
:rtype:pyNone

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, tty=False)[source]

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

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.

:rtype:pySSHResponse
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.

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.
:rtype:pylist[list[str]]
matrixctl.handlers.table.find_newlines(data)[source]

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

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.

:rtype:pydict[int, int]

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.

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 srting per column. (n is the number of columns)

Returns
row_stringstr

A formatted string, which represents a table row.

:rtype:pystr
matrixctl.handlers.table.get_colum_length(data, headers)[source]

Transpose rows and find longest line.

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_tupleNone

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

rtype

tuple[int, …] ..

matrixctl.handlers.table.handle_newlines(part, newlines)[source]

Update and insert new lines.

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
part, inhibit_septuple [list of list of str, set of int]

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

:rtype:pytuple[list[list[str]], set[int]]
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 dertermine in how many rows the row gets expanded, to get one row per newline - 1.

Parameters
rowlist of str

Data or headers of the table.

Returns
max_newlinesint

The highest number of newlines ber row.

:rtype:pyint
matrixctl.handlers.table.table(table_data, table_headers=None, sep=True, none='-')[source]

Create a table from data and a optional headers.

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 ther should be a separator between every row of data.

nonestr, default = “-”

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

Yields
tableGenerator [str, None, None]
The table (row for row).
rtype

Generator[str, None, None] ..

matrixctl.handlers.table.transpose_newlines_to_rows(splitted, occurences)[source]

Transpose newlines in new rows.

Parameters
splittedlist of list of str

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

occurencesint

The maximal number of newlines across the row.

Yields
rowlist[str]

A row for each occurence.

:rtype:pyGenerator[list[str], None, None]