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
- server
matrixctl.structures.ConfigServer
The configuration of a (home)server.
- server
- Returns
- server
matrixctl.structures.ConfigServer
The configuration of a (home)server with applied defaults.
- :rtype:
py
ConfigServer
- server
- 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
- *keys
str
A tuple of strings describing the values you are looking for.
- *keys
- Returns
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 extensionyaml
is greater than the priority of the file extensionyml
.Warning
The paths returned by this function might not exist.
- Returns
- config_paths
tuple
of
pathlib.Path
A tuple of paths, which might contain a config file.
- :rtype:
py
tuple
[Path
,...
]
- config_paths
- get_server_config(paths, server)[source]
Read and concentrate the config in one dict.
The
servers: ...
will be removed form the dict. A new entryserver
will be created, which represents the selected server.- Parameters
- paths
Iterable
of
pathlib.Path
The paths to the configfiles.
- server
str
The selected server. (Default: “default”)
- paths
- Returns
- server_config
matrixctl.typehints.Config
The config for the selected server.
- :rtype:
py
Config
- server_config
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
- yaml
ruamel.yaml.Yaml
The yaml object.
- path
Path
The path where the config file is located.
- yaml
- Returns
- full_config
matrixctl.typehints.Config
The full (with server name) config file as dict.
- :rtype:
py
Config
- full_config
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.
- content: Optional[Union[str, bytes, collections.abc.Iterable[bytes]]]
- 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.
- async matrixctl.handlers.api.async_worker(input_queue, output_queue)[source]
Use this coro as worker to make (a)synchronous request.
See also
RequestBuilder
matrixctl.handlers.api.RequestBuilder
- Attributes
- input_queue
asyncio.Queue
The input queue, which provides the
RequestBuilder
.- output_queue
asyncio.Queue
The output queue, which gets the responses of there requests.
- input_queue
- 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
- responses
list
of
httpx.Response
orhttpx.Response
Depending on
concurrent_limit
anrequest_config
.
- rtype
httpx.Response
|list
[httpx.Response
] ..
- responses
See also
RequestBuilder
matrixctl.handlers.api.RequestBuilder
- Attributes
- request_config
RequestBuilder
orGenerator
[RequestBuilder
,None
,None
] An instance of an
RequestBuilder
or a list ofRequestBuilder
. If the function gets aRequestBuilder
, the request will be synchronous. If it gets a Generator, the request will be asynchronous.- concurrent_limit
int
The maximum of concurrent workers. (This information must be pulled from the config.)
- request_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_config
matrixctl.handlers.api.RequestBuilder
Yields a fully configured
RequestsBuilder
for every request that has to be done to get all entries.- :rtype:
py
Generator
[RequestBuilder
,None
,None
]
- request_config
Notes
Warning
Call-By-Reference
like behavior! The paramlimit
and theconcurrent_limit
inrequest_config
will get changed in this function. Make sure to only use them after using this function!- Attributes
- request_config
matrixctl.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_token
int
The value, which defines from where to start in the next request. You get this value from the response of an initial synchronous request.
- total
int
The value which defines how many entries there are. You get this value from the response of an initial synchronous request.
- request_config
- 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
- responses
list
of
httpx.Response
orhttpx.Response
Depending on
concurrent
, it is ahttpx.Response
ifconcurrent
is true, otherwise it is alist
ofhttpx.Response
.
- responses
- Attributes
- input_size
int
The number of items in the queue.
- output_queue
asyncio.Queue
The output queue, which holds the responses of there requests.
- concurrentbool
When
True
, make requests concurrently. WhenFalse
, make requests synchronously.
- input_size
- matrixctl.handlers.api.preplan_request_strategy(limit, concurrent_limit, max_step_size=100)[source]
Use this functiona as helper for optimizing asyncronous requests.
- Returns
- RequestStrategy
matrixctl.handlers.api.RequestStrategy
A Named tuple with the RequestStrategy values.
- rtype
- RequestStrategy
- Attributes
- 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
- response
httpx.Response
Returns the response
- rtype
list
[httpx.Response
] |httpx.Response
..
- response
See also
RequestBuilder
matrixctl.handlers.api.RequestBuilder
- Attributes
- request_config
RequestBuilder
orGenerator
[RequestBuilder
,None
,None
] An instance of an
RequestBuilder
or a list ofRequestBuilder
. If the function gets aRequestBuilder
, the request will be synchronous. If it gets a Generator, the request will be asynchronous.- concurrent_limit
int
The maximum of concurrent workers. (This information must be pulled from the config.)
- request_config
Ansible
Run a ansible playbook with this module.
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
- datetime
datetime.datetime
The datetime object.
- :rtype:
py
datetime
- datetime
- log(since=None)[source]
Print a table of date, user and commit message since the last pull.
- Parameters
- since
datetime.datetime
,optional
, default=None The datetime the last commit was puled.
- since
- Returns
-
- rtype
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.
- run_cmd(cmd, tty=False)[source]
Run a command on the host machine and receive a response.
- Parameters
- Returns
- response
matrixctl.handlers.ssh.SSHResponse
Receive
stdin
,stdout
andstderr
as response.- :rtype:
py
SSHResponse
- response
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
- part
collections.abc.Sequence
of
collections.abc.Sequence
of
str
Data or header, in which every cell will be to casted to to strings.
- none
str
A string, which is used to replace
None
with the specific string.
- part
- Returns
- matrixctl.handlers.table.find_newlines(data)[source]
Find newlines and return a dict with positions (key) and occurences.
- Parameters
- Returns
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
.
- matrixctl.handlers.table.get_colum_length(data, headers)[source]
Transpose rows and find longest line.
- matrixctl.handlers.table.handle_newlines(part, newlines)[source]
Update and insert new lines.
- Parameters
- Returns
- part, inhibit_sep
tuple
[list
of
list
of
str
,set
of
int
] The
part
contains the supplemented and updated rows. Theinhibit_sep
set
contains the line numbers where a separator yhould be inhibited because the lines handled by this function are belonging together.- :rtype:
py
tuple
[list
[list
[str
]],set
[int
]]
- part, inhibit_sep
- 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.
- matrixctl.handlers.table.table(table_data, table_headers=None, sep=True, none='-')[source]
Create a table from data and a optional headers.
- Parameters
- table_data
collections.abc.Sequence
of
collections.abc.Sequence
of
str
Data.
- table_headers
collections.abc.Sequence
of
str
,Optional
Headers.
- sepbool,
default
=True
True
, when ther should be a separator between every row of data.- none
str
,default
= “-” A string, which is used to replace
None
with the specific string.
- table_data
- Yields