7.6. Common functions (`vantage6-common`)#

This page contains the API reference of the functions in the vantage-common package.

7.6.1. vantage6.common.configuration_manager#

class Configuration(*args, **kwargs)#

Base class to contain a single configuration.

property is_valid: bool#

Check if the configuration is valid.

Returns:: Whether or not the configuration is valid.
Return type:: bool

class ConfigurationManager(conf_class=<class 'vantage6.common.configuration_manager.Configuration'>, name=None)#

Class to maintain valid configuration settings.

Parameters:

conf_class (Configuration) – The class to use for the configuration.
name (str) – The name of the configuration.

classmethod from_file(path, conf_class=<class 'vantage6.common.configuration_manager.Configuration'>)#

Load a configuration from a file.

Parameters:

path (Path | str) – The path to the file to load the configuration from.
conf_class (Type[Configuration]) – The class to use for the configuration.

Returns:

The configuration manager with the configuration.

Return type:

ConfigurationManager

Raises:

AssertionError – If the name of the configuration could not be extracted from the file path.

get()#

Get a configuration from the configuration manager.

Returns:: The configuration.
Return type:: Configuration

property is_empty: bool#

Check if the configuration manager is empty.

Returns:: Whether or not the configuration manager is empty.
Return type:: bool

load(path)#

Load a configuration from a file.

Parameters:: path (Path | str) – The path to the file to load the configuration from.
Return type:: None

put(config)#

Add a configuration to the configuration manager.

Parameters:: config (dict) – The configuration to add.
Raises:: AssertionError – If the configuration is not valid.
Return type:: None

save(path)#

Save the configuration to a file.

Parameters:: path (Path | str) – The path to the file to save the configuration to.
Return type:: None

7.6.2. vantage6.common.context#

class AppContext(instance_type, instance_name, system_folders=False, config_file=None)#

Base class from which to create Node and Server context classes.

INST_CONFIG_MANAGER#: alias of ConfigurationManager

classmethod available_configurations(instance_type, system_folders)#

Returns a list of configuration managers and a list of paths to configuration files that could not be loaded.

Parameters:

instance_type (str) – ‘server’ or ‘node’
system_folders (bool) – Use system folders rather than user folders

Returns:

A list of configuration managers and a list of paths to configuration files that could not be loaded.

Return type:

list[ConfigurationManager], list[Path]

classmethod config_exists(instance_type, instance_name, system_folders=False)#

Check if a config file exists for the given instance type and name.

Parameters:

instance_type (str) – ‘server’ or ‘node’
instance_name (str) – Name of the configuration
system_folders (bool) – Use system folders rather than user folders

Returns:

True if the config file exists, False otherwise

Return type:

bool

property config_file: Path#

Return the path to the configuration file.

Returns:: Path to the configuration file
Return type:: Path

property config_file_name: str#

Return the name of the configuration file.

Returns:: Name of the configuration file
Return type:: str

static configure_logger(name, level)#

Set the logging level of a logger.

Parameters:

name (str) – Name of the logger to configure. If None, the root logger is configured.
level (str) – Logging level to set. Must be one of ‘debug’, ‘info’, ‘warning’, ‘error’, ‘critical’.

Returns:

The logger object and the logging level that was set.

Return type:

Tuple[Logger, int]

classmethod find_config_file(instance_type, instance_name, system_folders, config_file=None, verbose=True)#

Find a configuration file.

Parameters:

instance_type (str) – ‘server’ or ‘node’
instance_name (str) – Name of the configuration
system_folders (bool) – Use system folders rather than user folders
config_file (str | None) – Name of the configuration file. If None, the name of the configuration is used.
verbose (bool) – Print the directories that are searched for the configuration file.

Returns:

Path to the configuration file

Return type:

str

Raises:

Exception – If the configuration file is not found

classmethod from_external_config_file(path, instance_type, system_folders=False)#

Create a new AppContext instance from an external config file.

Parameters:

path (str) – Path to the config file
instance_type (str) – ‘server’ or ‘node’
system_folders (bool) – Use system folders rather than user folders

Returns:

A new AppContext instance

Return type:

AppContext

get_data_file(filename)#

Return the path to a data file.

Parameters:: filename (str) – Name of the data file
Returns:: Path to the data file
Return type:: str

static instance_folders(instance_type, instance_name, system_folders)#

Return OS and instance specific folders for storing logs, data and config files.

Parameters:

instance_type (str) – ‘server’ or ‘node’
instance_name (str) – Name of the configuration
system_folders (bool) – Use system folders rather than user folders

Returns:

Dictionary with Paths to the folders of the log, data and config files.

Return type:

dict

property log_file: Path#

Return the path to the log file.

Returns:: Path to the log file
Return type:: Path

log_file_name(type_)#

Return a path to a log file for a given log file type

Parameters:: type (str) – The type of log file to return.
Returns:: The path to the log file.
Return type:: Path
Raises:: AssertionError – If the configuration manager is not initialized.

set_folders(instance_type, instance_name, system_folders)#

Set the folders where the configuration, data and log files are stored.

Parameters:

instance_type (str) – ‘server’ or ‘node’
instance_name (str) – Name of the configuration
system_folders (bool) – Whether to use system folders rather than user folders

Return type:

None

setup_logging()#

Setup a basic logging mechanism.

Exits if the log file can’t be created.

Return type:: None

static type_data_folder(instance_type, system_folders)#

Return OS specific data folder.

Parameters:

instance_type (str) – ‘server’ or ‘node’
system_folders (bool) – Use system folders rather than user folders

Returns:

Path to the data folder

Return type:

Path

7.6.3. vantage6.common.encryption#

Encryption between organizations

Module to provide async encrpytion between organizations. All input and result fields should be encrypted when communicating to the central server.

All incomming messages (input/results) should be encrypted using the public key of this organization. This way we can decrypt them using our private key.

In the case we are sending messages (input/results) we need to encrypt it using the public key of the receiving organization. (retreiving these public keys is outside the scope of this module).

class CryptorBase(*args, **kwargs)#

Base class/interface for encryption implementations.

static bytes_to_str(data)#

Encode bytes as base64 encoded string.

Parameters:: data (bytes) – The data to encode.
Returns:: The base64 encoded string.
Return type:: str

decrypt_str_to_bytes(data)#

Decrypt base64 encoded string data.

Parameters:: data (str) – The data to decrypt.
Returns:: The decrypted data.
Return type:: bytes

encrypt_bytes_to_str(data, pubkey_base64)#

Encrypt bytes in data using a (base64 encoded) public key.

Note that the public key is ignored in this base class. If you want to encode your data with a public key, use the RSACryptor class.

Parameters:

data (bytes) – The data to encrypt.
pubkey_base64 (str) – The public key to use for encryption. This is ignored in this base class.

Returns:

The encrypted data encoded as base64 string.

Return type:

str

static str_to_bytes(data)#

Decode base64 encoded string to bytes.

Parameters:: data (str) – The base64 encoded string.
Returns:: The encoded string converted to bytes.
Return type:: bytes

class DummyCryptor(*args, **kwargs)#: Does absolutely nothing to encrypt the data.

class RSACryptor(private_key_file)#

Wrapper class for the cryptography package.

It loads the private key, and has an interface to encrypt en decrypt messages. If no private key is found, it can generate one, and store it at the default location. The encrpytion can be done via a public key from another organization, make sure the key is in the right data-type.

Communication between node and server requires serialization (and deserialization) of the encrypted messages (which are in bytes). The API can not communicate bytes, therefore a base64 conversion needs to be executed (and also a utf-8 encoding needs to be applied because of the way python implemented base64). The same goes for sending and receiving the public_key.

Parameters:: private_key_file (Path) – The path to the private key file.

static create_new_rsa_key(path)#

Creates a new RSA key for E2EE.

Parameters:: path (Path) – The path to the private key file.
Returns:: The newly created private key.
Return type:: RSAPrivateKey

static create_public_key_bytes(private_key)#

Create a public key from a private key.

Parameters:: private_key (RSAPrivateKey) – The private key to use.
Returns:: The public key as bytes.
Return type:: bytes

decrypt_str_to_bytes(data)#

Decrypt base64 encoded string data.

Parameters:: data (str) – The data to decrypt.
Returns:: The decrypted data.
Return type:: bytes

encrypt_bytes_to_str(data, pubkey_base64s)#

Encrypt bytes in data using a (base64 encoded) public key.

Parameters:

data (bytes) – The data to encrypt.
pubkey_base64s (str) – The public key to use for encryption.

Returns:

The encrypted data encoded as base64 string.

Return type:

str

property public_key_bytes: bytes#

Returns the public key bytes from the organization.

Returns:: The public key as bytes.
Return type:: bytes

property public_key_str: str#

Returns a JSON safe public key, used for the API.

Returns:: The public key as base64 encoded string.
Return type:: str

verify_public_key(pubkey_base64)#

Verifies the public key.

Compare a public key with the generated public key from the private key that is stored in this instance. This is usefull for verifying that the public key stored on the server is derived from the currently used private key.

Parameters:: pubkey_base64 (str) – The public key to verify as returned from the server.
Returns:: True if the public key is valid, False otherwise.
Return type:: bool

7.6.4. vantage6.common#

class ClickLogger#

“Logs output to the click interface.

static debug(msg)#

Print a debug message to the click interface.

Parameters:: msg (str) – The message to print.
Return type:: None

static error(msg)#

Print an error message to the click interface.

Parameters:: msg (str) – The message to print.
Return type:: None

static info(msg)#

Print an info message to the click interface.

Parameters:: msg (str) – The message to print.
Return type:: None

static warn(msg)#

Print a warning message to the click interface.

Parameters:: msg (str) – The message to print.
Return type:: None

class Singleton#: Singleton metaclass. It allows us to create just a single instance of a class to which it is the metaclass.

class WhoAmI(type_: str, id_: int, name: str, organization_name: str, organization_id: int)#

Data-class to store Authenticatable information in.

Variables:

type (str) – The type of the authenticatable (user or node).
id (int) – The id of the authenticatable.
name (str) – The name of the authenticatable.
organization_name (str) – The name of the organization of the authenticatable.
organization_id (int) – The id of the organization of the authenticatable.

id_: int#: Alias for field number 1

name: str#: Alias for field number 2

organization_id: int#: Alias for field number 4

organization_name: str#: Alias for field number 3

type_: str#: Alias for field number 0

base64s_to_bytes(bytes_string)#

Convert base64 encoded string to bytes.

Parameters:: bytes_string (str) – The base64 encoded string.
Returns:: The encoded string converted to bytes.
Return type:: bytes

bytes_to_base64s(bytes_)#

Convert bytes into base64 encoded string.

Parameters:: bytes (bytes) – The bytes to convert.
Returns:: The base64 encoded string.
Return type:: str

check_config_writeable(system_folders=False)#

Check if the user has write permissions to create the configuration file.

Parameters:: system_folders (bool) – Whether to check the system folders or the user folders.
Returns:: Whether the user has write permissions to create the configuration file or not.
Return type:: bool

debug(msg)#

Print a debug message to the CLI.

Parameters:: msg (str) – The message to print.
Return type:: None

echo(msg, level='info')#

Print a message to the CLI.

Parameters:

msg (str) – The message to print.
level (str) – The level of the message. Can be one of: “error”, “warn”, “info”, “debug”.

Return type:

None

error(msg)#

Print an error message to the CLI.

Parameters:: msg (str) – The message to print.
Return type:: None

generate_apikey()#

Creates random api_key using uuid.

Returns:: api_key
Return type:: str

get_config_path(dirs, system_folders=False)#

Get the path to the configuration directory.

Parameters:

dirs (appdirs.AppDirs) – The appdirs object.
system_folders (bool) – Whether to get path to the system folders or the user folders.

Returns:

The path to the configuration directory.

Return type:

str

get_database_config(databases, label)#

Get database configuration from config file

Parameters:

databases (list[dict]) – List of database configurations
label (str) – Label of database configuration to retrieve

Returns:

Database configuration, or None if not found

Return type:

Dict | None

info(msg)#

Print an info message to the CLI.

Parameters:: msg (str) – The message to print.
Return type:: None

is_ip_address(ip)#

Test if input IP address is a valid IP address

Parameters:: ip (str) – IP address to validate
Returns:: bool
Return type:: whether or not IP address is valid

logger_name(special__name__)#

Return the name of the logger.

Parameters:: special__name__ (str) – The __name__ variable of a module.
Returns:: The name of the logger.
Return type:: str

warning(msg)#

Print a warning message to the CLI.

Parameters:: msg (str) – The message to print.
Return type:: None

7.6.5. vantage6.common.docker.addons#

class ContainerKillListener#

Listen for signals that the docker container should be shut down

exit_gracefully(*args)#

Set kill_now to True. This will trigger the container to stop

Return type:: None

check_docker_running()#

Check if docker engine is running. If not, exit the program.

Return type:: None

delete_network(network, kill_containers=True)#

Delete network and optionally its containers

Parameters:

network (Network) – Network to delete
kill_containers (bool) – Whether to kill the containers in the network (otherwise they are merely disconnected)

Return type:

None

get_container(docker_client, **filters)#

Return container if it exists after searching using kwargs

Parameters:

docker_client (DockerClient) – Python docker client
**filters – These are arguments that will be passed to the client.container.list() function. They should yield 0 or 1 containers as result (e.g. name=’something’)

Returns:

Container if it exists, else None

Return type:

Container or None

get_network(docker_client, **filters)#

Return network if it exists after searching using kwargs

Parameters:

docker_client (DockerClient) – Python docker client
**filters – These are arguments that will be passed to the client.network.list() function. They should yield 0 or 1 networks as result (e.g. name=’something’)

Returns:

Container if it exists, else None

Return type:

Container or None

get_networks_of_container(container)#

Get list of networks the container is in

Parameters:: container (Container) – The container in which we are interested
Returns:: Describes container’s networks and their properties
Return type:: dict

get_num_nonempty_networks(container)#

Get number of networks the container is in where it is not the only one

Parameters:: container (Container) – The container in which we are interested
Returns:: Number of networks in which the container resides in which there are also other containers
Return type:: int

get_server_config_name(container_name, scope)#

Get the configuration name of a server from its docker container name

Docker container name of the server is formatted as f”{APPNAME}-{self.name}-{self.scope}-server”. This will return {self.name}

Parameters:

container_name (str) – Name of the docker container in which the server is running
scope (str) – Scope of the server (e.g. ‘system’ or ‘user’)

Returns:

A server’s configuration name

Return type:

str

inspect_local_image_timestamp(docker_client, image, log=<class 'vantage6.common.ClickLogger'>)#

Obtain creation timestamp object from local image.

Parameters:

docker_client (DockerClient) – Docker client
image (str) – Image name
log (logging.Logger | ClickLogger) – Logger

Returns:

Timestamp containing the creation date and time of the local image and the image digest. If the image does not exist, None is returned.

Return type:

tuple[datetime, str] | None

inspect_remote_image_timestamp(docker_client, image, log=<class 'vantage6.common.ClickLogger'>)#

Obtain creation timestamp object from remote image.

Parameters:

docker_client (DockerClient) – Docker client
image (str) – Image name
log (logging.Logger | ClickLogger) – Logger

Returns:

Timestamp containing the creation date of the image and its digest, or None if the remote image could not be found.

Return type:

tuple[datetime, str] | None

pull_if_newer(docker_client, image, log=<class 'vantage6.common.ClickLogger'>)#

Docker pull only if the remote image is newer.

Parameters:

docker_client (DockerClient) – A Docker client instance
image (str) – Image to be pulled
log (logger.Logger or ClickLogger) – Logger class

Raises:

docker.errors.APIError – If the image cannot be pulled

Return type:

None

registry_basic_auth_header(docker_client, registry)#

Obtain credentials for registry

This is a wrapper around docker-py to obtain the credentials used to access a registry. Normally communication to the registry goes through the Docker deamon API (locally), therefore we have to take extra steps in order to communicate with the (Harbor) registry directly.

Note that this has only been tested for the harbor registries.

Parameters:

docker_client (DockerClient) – Docker client
registry (str) – registry name (e.g. harbor2.vantage6.ai)

Returns:

Containing a basic authorization header

Return type:

dict

remove_container(container, kill=False)#

Removes a docker container

Parameters:

container (Container) – The container that should be removed
kill (bool) – Whether or not container should be killed before it is removed

Return type:

None

remove_container_if_exists(docker_client, **filters)#

Kill and remove a docker container if it exists

Parameters:

docker_client (DockerClient) – A Docker client
**filters – These are arguments that will be passed to the client.container.list() function. They should yield 0 or 1 containers as result (e.g. name=’something’)

Return type:

None

running_in_docker()#

Check if this code is executed within a Docker container.

Returns:: True if the code is executed within a Docker container, False otherwise
Return type:: bool

7.6.6. vantage6.common.docker.network_manager#

class NetworkManager(network_name)#

Handle a Docker network

connect(container_name, aliases=None, ipv4=None)#

Connect a container to the network.

Parameters:

container_name (str) – Name of the container that should be connected to the network
aliases (list[str]) – A list of aliases for the container in the network
ipv4 (str | None) – An IP address to assign to the container in the network

Return type:

None

contains(container)#

Whether or not this network contains a certain container

Parameters:: container (Container) – container to look for in network
Returns:: Whether or not container is in the network
Return type:: bool

create_network(is_internal=True)#

Creates an internal (docker) network

Used by algorithm containers to communicate with the node API.

Parameters:: is_internal (bool) – True if network should only be able to communicate internally
Return type:: None

delete(kill_containers=True)#

Delete network

Parameters:: kill_containers (bool) – If true, kill and remove any containers in the network
Return type:: None

disconnect(container_name)#

Disconnect a container from the network.

Parameters:: container – Name of the container to disconnect
Return type:: None

get_container_ip(container_name)#

Get IP address of a container in the network

Parameters:: container_name (str) – Name of the container whose IP address is sought
Returns:: IP address of the container in the network
Return type:: str

remove_subnet_mask(ip)#

Remove the subnet mask of an ip address, e.g. 172.1.0.0/16 -> 172.1.0.0

Parameters:: ip (str) – IP subnet, potentially including a mask
Returns:: IP subnet address without the subnet mask
Return type:: str

7.6.7. vantage6.common.task_status#

class TaskStatus(value)#: Enum to represent the status of a task

has_task_failed(status)#

Check if task has failed to run to completion

Parameters:: status (TaskStatus | str) – The status of the task
Returns:: True if task has failed, False otherwise
Return type:: bool

has_task_finished(status)#

Check if task has finished or crashed

Parameters:: status (TaskStatus | str) – The status of the task
Returns:: True if task has finished or failed, False otherwise
Return type:: bool

7.6.8. vantage6.common.colors#

ColorStreamHandler#: alias of _AnsiColorStreamHandler

class _AnsiColorStreamHandler(stream=None)#

Handler for logging colors to a stream, for example sys.stderr or sys.stdout.

This handler is used for non-Windows systems.

classmethod _get_color(level)#

Define which color to print for each log level.

Parameters:: level (int) – The log level.
Returns:: The color to print.
Return type:: str

format(record)#

Format the log record.

Parameters:: record (logging.LogRecord) – The log record to format.
Returns:: The formatted log record.
Return type:: str

class _WinColorStreamHandler(stream=None)#

Color stream handler for Windows systems.

classmethod _get_color(level)#

Define which color to print for each log level.

Parameters:: level (int) – The log level.
Returns:: The color to print.
Return type:: str

emit(record)#

Write a log record to the stream.

Parameters:: record (logging.LogRecord) – The log record to write.
Return type:: None

7.6.9. vantage6.common.exceptions#

exception AuthenticationException#: Exception to indicate authentication has failed

7.6. Common functions (vantage6-common)#

7.6.1. vantage6.common.configuration_manager#

7.6.2. vantage6.common.context#

7.6.3. vantage6.common.encryption#

7.6.4. vantage6.common#

7.6.5. vantage6.common.docker.addons#

7.6.6. vantage6.common.docker.network_manager#

7.6.7. vantage6.common.task_status#

7.6.8. vantage6.common.colors#

7.6.9. vantage6.common.exceptions#

7.6. Common functions (`vantage6-common`)#