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

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

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

A configuration file contains at top level an application and/or environments key. The environments key can contain up to four keys: dev, test, acc, prod. e.g.:

application:

…

environments:

dev:: …
test:: …
acc:: …
prod:: …

Note that this structure is the same for the node and server.

Parameters:

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

property available_environments: list[str]#

Get a list of available environments.

Returns:: A list of available environments.
Return type:: list[str]

property environments: dict#

Get all environments.

Returns:: A dictionary containing all environments.
Return type:: dict

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(env)#

Get a configuration from the configuration manager.

Parameters:: env (str) – The environment to get the configuration from.
Returns:: The configuration.
Return type:: Configuration
Raises:: AssertionError – If the environment is not valid.

property has_application: bool#

Check if the configuration manager has an application configuration.

Returns:: Whether or not the configuration manager has an application configuration.
Return type:: bool

property has_environments: bool#

Check if the configuration manager has any environment configurations.

Returns:: Whether or not the configuration manager has any environment configurations.
Return type:: bool

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(env, config)#

Add a configuration to the configuration manager.

Parameters:

env (str) – The environment to add the configuration to.
config (dict) – The configuration to add.

Raises:

AssertionError – If the environment 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.5.2. vantage6.common.context#

class AppContext(instance_type, instance_name, environment='application', 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, environment='application', 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
environment (str) – Environment within the config file to use. Can be any of ‘application’, ‘dev’, ‘test’, ‘acc’ or ‘prod’.
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]

property environment: str#

Return the environment.

Returns:: Environment
Return type:: str

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, environment='application', 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’
environment (str) – Environment within the config file to use. Can be any of ‘application’, ‘dev’, ‘test’, ‘acc’ or ‘prod’.
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.5.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.5.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_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

Notes

The databases configuration can be in two formats. The new format allows for the specification of the database type. The structure of the new format is as follows:

1. Old format: {

“database_label”: “database_uri”, …

}

2. New format: [

{
“label”: “database_label”, “uri”: “database_uri”, “db_type”: “database_type”

}

]

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.5.5. vantage6.common.docker.addons#

7.5.6. vantage6.common.docker.network_manager#

7.5.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.5.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.5.9. vantage6.common.exceptions#

exception AuthenticationException#: Exception to indicate authentication has failed

7.5. Common functions (vantage6-common)#

7.5.1. vantage6.common.configuration_manager#

7.5.2. vantage6.common.context#

7.5.3. vantage6.common.encryption#

7.5.4. vantage6.common#

7.5.5. vantage6.common.docker.addons#

7.5.6. vantage6.common.docker.network_manager#

7.5.7. vantage6.common.task_status#

7.5.8. vantage6.common.colors#

7.5.9. vantage6.common.exceptions#

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