# SPDX-FileCopyrightText: 2020-present The Firebird Projects <www.firebirdsql.org>
#
# SPDX-License-Identifier: MIT
#
# PROGRAM/MODULE: firebird-base
# FILE: firebird/base/config.py
# DESCRIPTION: Classes for configuration definitions
# CREATED: 14.5.2020
#
# The contents of this file are subject to the MIT License
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# Copyright (c) 2019 Firebird Project (www.firebirdsql.org)
# All Rights Reserved.
#
# Contributor(s): Pavel Císař (original code)
# ______________________________________.
"""Firebird Base - Classes for configuration definitions
Complex applications (and some library modules like `logging`) could be often parametrized
via configuration. This module provides a framework for unified structured configuration
that supports:
* configuration options of various data type, including lists and other complex types
* validation
* direct manipulation of configuration values
* reading from (and writing into) configuration in `configparser` format
* exchanging configuration (for example between processes) using Google protobuf messages
* application directory scheme
"""
from __future__ import annotations
from typing import Generic, Type, Any, List, Dict, Union, Sequence, Callable, Optional, \
TypeVar, cast, get_type_hints
from abc import ABC, abstractmethod
import platform
from pathlib import Path
from uuid import UUID
from decimal import Decimal, DecimalException
from configparser import (ConfigParser, DEFAULTSECT, ExtendedInterpolation,
MAX_INTERPOLATION_DEPTH, InterpolationDepthError,
InterpolationSyntaxError, NoSectionError, NoOptionError,
InterpolationMissingOptionError)
from inspect import signature, Signature, Parameter
from enum import Enum, Flag
import os
from .config_pb2 import ConfigProto
from .types import Error, MIME, ZMQAddress, PyExpr, PyCode, PyCallable
from .strconv import get_convertor, convert_to_str, Convertor
PROTO_CONFIG = 'firebird.base.ConfigProto'
[docs]
def has_verticals(value: str) -> bool:
"Returns True if lines in multiline string contains leading '|' character."
return any(1 for line in value.split('\n') if line.startswith('|'))
[docs]
def has_leading_spaces(value: str) -> bool:
"Returns True if any line in multiline string starts with space(s)."
return any(1 for line in value.split('\n') if line.startswith(' '))
def unindent_verticals(value: str) -> str:
"""Removes leading '|' character from each line in multiline string."""
lines = []
indent = None
for line in value.split('\n'):
if line.startswith('|'):
if indent is None:
indent = (len(line[1:]) - len(line[1:].strip())) + 1
lines.append(line[indent:])
else:
lines.append(line)
return '\n'.join(lines)
def _eq(a: Any, b: Any) -> bool:
return str(a) == str(b)
[docs]
def create_config(_cls: Type[Config], name: str) -> Config: # pragma: no cover
"""Return newly created `Config` instance. Intended to be used with `functools.partial`.
.. deprecated:: 1.6
Will be removed in version 2.0
"""
return _cls(name)
# Next two functions are copied from stdlib enum module, as they were removed in Python 3.11
def _decompose(flag, value):
"""
Extract all members from the value.
"""
# _decompose is only called if the value is not named
not_covered = value
negative = value < 0
# issue29167: wrap accesses to _value2member_map_ in a list to avoid race
# conditions between iterating over it and having more pseudo-
# members added to it
if negative:
# only check for named flags
flags_to_check = [
(m, v)
for v, m in list(flag._value2member_map_.items())
if m.name is not None
]
else:
# check for named flags and powers-of-two flags
flags_to_check = [
(m, v)
for v, m in list(flag._value2member_map_.items())
if m.name is not None or _power_of_two(v)
]
members = []
for member, member_value in flags_to_check:
if member_value and member_value & value == member_value:
members.append(member)
not_covered &= ~member_value
if not members and value in flag._value2member_map_:
members.append(flag._value2member_map_[value])
members.sort(key=lambda m: m._value_, reverse=True)
if len(members) > 1 and members[0].value == value:
# we have the breakdown, don't need the value member itself
members.pop(0)
return members, not_covered
def _power_of_two(value):
if value < 1:
return False
return value == 2 ** (value.bit_length() - 1)
[docs]
class EnvExtendedInterpolation(ExtendedInterpolation):
""".. versionadded:: 1.8.0
Modified version of `configparser.ExtendedInterpolation` class that adds special
handling for "env" section that returns value of specified environment variable,
or empty string if such variable is not defined.
Example::
${env:path} is reference to PATH environment variable.
"""
def _interpolate_some(self, parser, option, accum, rest, section, map,
depth):
rawval = parser.get(section, option, raw=True, fallback=rest)
if depth > MAX_INTERPOLATION_DEPTH:
raise InterpolationDepthError(option, section, rawval)
while rest:
p = rest.find("$")
if p < 0:
accum.append(rest)
return
if p > 0:
accum.append(rest[:p])
rest = rest[p:]
# p is no longer used
c = rest[1:2]
if c == "$":
accum.append("$")
rest = rest[2:]
elif c == "{":
m = self._KEYCRE.match(rest)
if m is None:
raise InterpolationSyntaxError(option, section,
"bad interpolation variable reference %r" % rest)
path = m.group(1).split(':')
rest = rest[m.end():]
sect = section
opt = option
try:
if len(path) == 1:
opt = parser.optionxform(path[0])
v = map[opt]
elif len(path) == 2:
sect = path[0]
opt = parser.optionxform(path[1])
if sect == 'env':
v = os.getenv(opt.upper(), '')
else:
v = parser.get(sect, opt, raw=True)
else:
raise InterpolationSyntaxError(
option, section,
"More than one ':' found: %r" % (rest,))
except (KeyError, NoSectionError, NoOptionError):
raise InterpolationMissingOptionError(
option, section, rawval, ":".join(path)) from None
if "$" in v:
self._interpolate_some(parser, opt, accum, v, sect,
dict(parser.items(sect, raw=True)),
depth + 1)
else:
accum.append(v)
else:
raise InterpolationSyntaxError(
option, section,
"'$' must be followed by '$' or '{', "
"found: %r" % (rest,))
[docs]
class DirectoryScheme:
"""Class that provide paths to typically used application directories.
Default scheme uses HOME directory as root for other directories. The HOME is
determined as follows:
1. If environment variable `<app_name>_HOME` exists, its value is used as HOME directory.
2. HOME directory is set to current working directory.
Note:
All paths are set when the instance is created and can be changed later.
"""
[docs]
def __init__(self, name: str, version: str=None, force_home: bool=False):
"""
Arguments:
name: Appplication name.
version: Application version.
force_home: When True, general directories (i.e. all except user-specific and
TMP) would be always based on HOME directory.
"""
self.name: str = name
self.version: str = version
self.force_home: bool = force_home
_h = os.getenv(f'{self.name.upper()}_HOME')
self.__home: Path = Path(_h) if _h is not None else Path(os.getcwd())
home = self.home
self.dir_map: Dict[str, Path] = {'config': home / 'config',
'run_data': home / 'run_data',
'logs': home / 'logs',
'data': home / 'data',
'tmp': home / 'tmp',
'cache': home / 'cache',
'srv': home / 'srv',
'user_config': home / 'user_config',
'user_data': home / 'user_data',
'user_sync': home / 'user_sync',
'user_cache': home / 'user_cache',
}
[docs]
def has_home_env(self) -> bool:
"""Returns True if HOME directory is set by "<app_name>_HOME" environment variable.
"""
return os.getenv(f'{self.name.upper()}_HOME') is not None
@property
def home(self) -> Path:
"""HOME directory. Initial value is path set by `<app_name>_HOME` environment
variable, or to current working directory when variable is not defined.
Important:
When new value is assigned, the general directories (i.e. all except user-specific
and TMP) are redefined as subdirectories of new home path ONLY when HOME was
initially defined using `<app_name>_HOME` environment variable, or instance
was created with `force_home` = True.
However, all paths could be still changed individually to any value.
"""
return self.__home
@home.setter
def home(self, value: Union[Path, str]) -> None:
self.__home = value if isinstance(value, Path) else Path(value)
if self.has_home_env() or self.force_home:
self.dir_map.update({'config': self.__home / 'config',
'run_data': self.__home / 'run_data',
'logs': self.__home / 'logs',
'data': self.__home / 'data',
'cache': self.__home / 'cache',
'srv': self.__home / 'srv'})
@property
def config(self) -> Path:
"""Directory for host-specific system-wide configuration files.
"""
return self.dir_map['config']
@config.setter
def config(self, path: Path) -> None:
self.dir_map['config'] = path
@property
def run_data(self) -> Path:
"""Directory for run-time variable data that may not persist over boot.
"""
return self.dir_map['run_data']
@run_data.setter
def run_data(self, path: Path) -> None:
self.dir_map['run_data'] = path
@property
def logs(self) -> Path:
"""Directory for log files.
"""
return self.dir_map['logs']
@logs.setter
def logs(self, path: Path) -> None:
self.dir_map['logs'] = path
@property
def data(self) -> Path:
"""Directory for state information / persistent data modified by application as
it runs.
"""
return self.dir_map['data']
@data.setter
def data(self, path: Path) -> None:
self.dir_map['data'] = path
@property
def tmp(self) -> Path:
"""Directory for temporary files to be preserved between reboots.
"""
return self.dir_map['tmp']
@tmp.setter
def tmp(self, path: Path) -> None:
self.dir_map['tmp'] = path
@property
def cache(self) -> Path:
"""Directory for application cache data.
Such data are locally generated as a result of time-consuming I/O or calculation.
The application must be able to regenerate or restore the data. The cached files
can be deleted without loss of data.
"""
return self.dir_map['cache']
@cache.setter
def cache(self, path: Path) -> None:
self.dir_map['cache'] = path
@property
def srv(self) -> Path:
"""Directory for site-specific data served by this system, such as data and
scripts for web servers, data offered by FTP servers, and repositories for
version control systems etc.
"""
return self.dir_map['srv']
@srv.setter
def srv(self, path: Path) -> None:
self.dir_map['srv'] = path
@property
def user_config(self) -> Path:
"""Directory for user-specific configuration.
"""
return self.dir_map['user_config']
@user_config.setter
def user_config(self, path: Path) -> None:
self.dir_map['user_config'] = path
@property
def user_data(self) -> Path:
"""Directory for User local data.
"""
return self.dir_map['user_data']
@user_data.setter
def user_data(self, path: Path) -> None:
self.dir_map['user_data'] = path
@property
def user_sync(self) -> Path:
"""Directory for user data synced accross systems (roaming).
"""
return self.dir_map['user_sync']
@user_sync.setter
def user_sync(self, path: Path) -> None:
self.dir_map['user_sync'] = path
@property
def user_cache(self) -> Path:
"""Directory for user-specific application cache data.
"""
return self.dir_map['user_cache']
@user_cache.setter
def user_cache(self, path: Path) -> None:
self.dir_map['user_cache'] = path
[docs]
class WindowsDirectoryScheme(DirectoryScheme):
"""Directory scheme that conforms to Windows standards.
If HOME is defined using "<app_name>_HOME" environment variable, or `force_home` parameter
is True, only user-specific directories and TMP are set according to platform standars,
while general directories remain as defined by base `DirectoryScheme`.
"""
[docs]
def __init__(self, name: str, version: str=None, force_home: bool=False):
"""
Arguments:
name: Appplication name.
version: Application version.
force_home: When True, general directories (i.e. all except user-specific and
TMP) would be always based on HOME directory.
"""
super().__init__(name, version, force_home)
app_dir = Path(self.name)
if self.version is not None:
app_dir /= self.version
pd = Path(os.path.expandvars('%PROGRAMDATA%'))
lad = Path(os.path.expandvars('%LOCALAPPDATA%'))
ad = Path(os.path.expandvars('%APPDATA%'))
# Set general directories only when HOME is not forced by environment variable.
if not self.has_home_env() and not force_home:
self.dir_map.update({'config': pd / app_dir / 'config',
'run_data': pd / app_dir / 'run',
'logs': pd / app_dir / 'log',
'data': pd / app_dir / 'data',
'cache': pd / app_dir / 'cache',
'srv': pd / app_dir / 'srv',
})
# Always set user-specific directories and TMP
self.dir_map.update({'tmp': lad / app_dir / 'tmp',
'user_config': lad / app_dir / 'config',
'user_data': lad / app_dir / 'data',
'user_sync': ad / app_dir,
'user_cache': lad / app_dir / 'cache',
})
[docs]
class LinuxDirectoryScheme(DirectoryScheme):
"""Directory scheme that conforms to Linux standards.
If HOME is defined using "<app_name>_HOME" environment variable, or `force_home` parameter
is True, only user-specific directories and TMP are set according to platform standars,
while general directories remain as defined by base `DirectoryScheme`.
"""
[docs]
def __init__(self, name: str, version: str=None, force_home: bool=False):
"""
Arguments:
name: Appplication name.
version: Application version.
force_home: When True, general directories (i.e. all except user-specific and
TMP) would be always based on HOME directory.
"""
super().__init__(name, version, force_home)
app_dir = Path(self.name)
if self.version is not None:
app_dir /= self.version
# Set general directories only when HOME is not forced by environment variable.
if not self.has_home_env() and not force_home:
self.dir_map.update({'config': Path('/etc') / app_dir,
'run_data': Path('/run') / app_dir,
'logs': Path('/var/log') / app_dir,
'data': Path('/var/lib') / app_dir,
'cache': Path('/var/cache') / app_dir,
'srv': Path('/srv') / app_dir,
})
# Always set user-specific directories and TMP
self.dir_map.update({'tmp': Path('/var/tmp') / app_dir,
'user_config': Path('~/.config').expanduser() / app_dir,
'user_data': Path('~/.local/share').expanduser() / app_dir,
'user_sync': Path('~/.local/sync').expanduser() / app_dir,
'user_cache': Path('~/.cache').expanduser() / app_dir,
})
[docs]
class MacOSDirectoryScheme(DirectoryScheme):
"""Directory scheme that conforms to MacOS standards.
If HOME is defined using "<app_name>_HOME" environment variable, only user-specific
directories and TMP are set according to platform standars, while general directories
remain as defined by base `DirectoryScheme`.
"""
[docs]
def __init__(self, name: str, version: str=None, force_home: bool=False):
"""
Arguments:
name: Appplication name.
version: Application version.
"""
super().__init__(name, version, force_home)
app_dir = Path(self.name)
if self.version is not None:
app_dir /= self.version
pd = Path('/Library/Application Support')
lad = Path('~/Library/Application Support').expanduser()
# Set general directories only when HOME is not forced by environment variable.
if not self.has_home_env() and not force_home:
self.dir_map.update({'config': pd / app_dir / 'config',
'run_data': pd / app_dir / 'run',
'logs': pd / app_dir / 'log',
'data': pd / app_dir / 'data',
'cache': pd / app_dir / 'cache',
'srv': pd / app_dir / 'srv',
})
# Always set user-specific directories and TMP
self.dir_map.update({'tmp': Path(os.getenv('TMPDIR')) / app_dir,
'user_config': lad / app_dir / 'config',
'user_data': lad / app_dir / 'data',
'user_sync': lad / app_dir,
'user_cache': Path('~/Library/Caches').expanduser() / app_dir / 'cache',
})
[docs]
def get_directory_scheme(app_name: str, version: str=None, *, force_home: bool=False) -> DirectoryScheme:
"""Returns directory scheme for current platform.
Arguments:
app_name: Application name
version: Application version string
force_home: When True, the general directies are always set to subdirectories of
`DirectoryScheme.home` directory. When False, then home is used ONLY
when it's set by "<app_name>_HOME" environment variable.
"""
return {'Windows': WindowsDirectoryScheme,
'Linux':LinuxDirectoryScheme,
'Darwin': MacOSDirectoryScheme}.get(platform.system(), DirectoryScheme)(app_name, version, force_home)
T = TypeVar('T')
[docs]
class Option(Generic[T], ABC):
"""Generic abstract base class for configuration options.
"""
[docs]
def __init__(self, name: str, datatype: T, description: str, required: bool=False,
default: T=None):
"""
Arguments:
name: Option name.
datatype: Option datatype.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
"""
assert name and isinstance(name, str), "name required"
assert datatype and isinstance(datatype, type), "datatype required"
assert description and isinstance(description, str), "description required"
assert default is None or isinstance(default, datatype), "default has wrong data type"
#: Option name.
self.name: str = name
#: Option datatype.
self.datatype: T = datatype
#: Option description. Can span multiple lines.
self.description: str = description
#: True if option must have a value.
self.required: bool = required
#: Default option value.
self.default: T = default
if default is not None:
self.set_value(default)
def _check_value(self, value: T) -> None:
if value is None and self.required:
raise ValueError(f"Value is required for option '{self.name}'.")
if value is not None and not isinstance(value, self.datatype):
raise TypeError(f"Option '{self.name}' value must be a "
f"'{self.datatype.__name__}',"
f" not '{type(value).__name__}'")
def _get_value_description(self) -> str:
return f'{self.datatype.__name__}\n'
[docs]
def _get_config_lines(self, plain: bool=False) -> List[str]:
"""Returns list of strings containing text lines suitable for use in configuration
file processed with `~configparser.ConfigParser`.
Text lines with configuration start with comment marker `;` and end with newline.
Arguments:
plain: When True, it outputs only the option value. When False, it includes also
option description and other helpful information.
Note:
This function is intended for internal use. To get string describing current
configuration that is suitable for configuration files, use `get_config` method.
"""
lines = []
if not plain:
if self.required:
lines.append("; REQUIRED option.\n")
for line in self.description.strip().splitlines():
lines.append(f"; {line}\n")
first = True
for line in self._get_value_description().splitlines():
lines.append(f"; {'Type: ' if first else ''}{line}\n")
first = False
value = self.get_value()
nodef = ';' if value == self.default else ''
value = '<UNDEFINED>' if value is None else self.get_formatted()
if '\n' in value:
chunks = value.splitlines(keepends=True)
new_value = [chunks[0]]
new_value.extend(f'{nodef}{x}' for x in chunks[1:])
value = ''.join(new_value)
lines.append(f"{nodef}{self.name} = {value}\n")
return lines
[docs]
def load_config(self, config: ConfigParser, section: str) -> None:
"""Update option value from `~configparser.ConfigParser` instance.
Arguments:
config: ConfigParser instance.
section: Name of ConfigParser section that should be used to get new option
value.
Raises:
ValueError: When option value cannot be loadded.
KeyError: If section does not exists, and it's not `configparser.DEFAULTSECT`.
"""
if not config.has_section(section) and section != DEFAULTSECT:
raise KeyError(f"Configuration error: section '{section}' not found!")
if config.has_option(section, self.name):
self.set_as_str(config[section][self.name])
[docs]
def validate(self) -> None:
"""Validates option state.
Raises:
Error: When required option does not have a value.
"""
if self.required and self.get_value() is None:
raise Error(f"Missing value for required option '{self.name}'")
[docs]
def get_config(self, *, plain: bool=False) -> str:
"""Returns string containing text lines suitable for use in configuration file
processed with `~configparser.ConfigParser`.
Arguments:
plain: When True, it outputs only the option value. When False, it includes also
option description and other helpful information.
"""
return ''.join(self._get_config_lines(plain=plain))
[docs]
def has_value(self) -> bool:
"""Returns True if option value is not None.
"""
return self.get_value() is not None
[docs]
@abstractmethod
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
[docs]
@abstractmethod
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
[docs]
@abstractmethod
def get_as_str(self) -> str:
"""Returns value as string.
"""
[docs]
@abstractmethod
def get_value(self) -> T:
"""Returns current option value.
"""
[docs]
@abstractmethod
def set_value(self, value: T) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
[docs]
@abstractmethod
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
[docs]
@abstractmethod
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
[docs]
class Config:
"""Collection of configuration options.
Important:
Descendants must define individual options and sub configs as instance attributes.
"""
[docs]
def __init__(self, name: str, *, optional: bool=False, description: str=None):
"""
Arguments:
name: Name associated with Config (default section name).
optional: Whether config is optional (False) or mandatory (True) for
configuration file (see `.load_config()` for details).
description: Optional configuration description. Can span multiple lines.
"""
self._name: str = name
self._optional: bool = optional
self._description: str = description if description is not None else self.__doc__
def __setattr__(self, name, value):
for attr in vars(self).values():
if isinstance(attr, Option) and attr.name == name:
raise ValueError('Cannot assign values to option itself, use `option.value` instead')
super().__setattr__(name, value)
[docs]
def validate(self) -> None:
"""Checks whether:
- all required options have value other than None.
- all options are defined as config attribute with the same name as option name
Raises exception when any constraint required by configuration is violated.
"""
for option in self.options:
option.validate()
if not hasattr(self, option.name):
raise Error(f"Option '{option.name}' is not defined as "
f"attribute with the same name")
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears all owned options and options in owned sub-configs.
Arguments:
to_default: If True, sets the option values to defaults, else to None.
"""
for option in self.options:
option.clear(to_default=to_default)
for config in self.configs:
config.clear(to_default=to_default)
[docs]
def get_description(self) -> str:
"""Configuration description. Can span multiple lines.
Note: If description is not provided on instance creation, class doc string.
"""
return '' if self._description is None else self._description
[docs]
def get_config(self, *, plain: bool=False) -> str:
"""Returns string containing text lines suitable for use in configuration file
processed with `~configparser.ConfigParser`.
Arguments:
plain: When True, it outputs only the option values. When False, it includes also
option descriptions and other helpful information.
"""
lines = [f'[{self.name}]\n', ';\n']
if not plain:
for line in self.get_description().strip().splitlines():
lines.append(f"; {line}\n")
for option in self.options:
if not plain:
lines.append('\n')
lines.append(option.get_config(plain=plain))
for config in self.configs:
if not plain:
lines.append('\n')
lines.append(config.get_config(plain=plain))
return ''.join(lines)
[docs]
def load_config(self, config: ConfigParser, section: str=None) -> None:
"""Update configuration.
Arguments:
config: ConfigParser instance with configuration values.
section: Name of ConfigParser section that should be used to get new
configuration values. If not provided, uses `name`.
Raises:
ValueError: When any option value cannot be loadded.
KeyError: If section does not exists, and config is not `optional` or section is
not `configparser.DEFAULTSECT`.
"""
if section is None:
section = self.name
if not config.has_section(section):
if self._optional:
return
if section != DEFAULTSECT:
raise Error(f"Configuration error: section '{section}' not found!")
try:
for option in self.options:
option.load_config(config, section)
for subcfg in self.configs:
subcfg.load_config(config)
except Error:
raise
except Exception as exc: # pragma: no cover
raise Error(f"Configuration error: {exc}") from exc
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains option values and sub-configs.
"""
for option in self.options:
option.load_proto(proto)
for subcfg in self.configs:
if subcfg.name in proto.configs:
subcfg.load_proto(proto.configs[subcfg.name])
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option values and sub-configs should be stored.
"""
for option in self.options:
option.save_proto(proto)
for subcfg in self.configs:
subcfg.save_proto(proto.configs.get_or_create(subcfg.name))
@property
def name(self) -> str:
"""Name associated with Config (default section name).
"""
return self._name
@property
def optional(self) -> bool:
"""Whether config is optional (False) or mandatory (True) for configuration file
(see `.load_config()` for details).
"""
return self._optional
@property
def options(self) -> List[Option]:
"""List of options defined for this Config instance.
"""
return [v for v in vars(self).values() if isinstance(v, Option)]
@property
def configs(self) -> List[Config]:
"""List of sub-Configs defined for this Config instance. It includes all instance
attributes of `Config` type, and `Config` values of owned `ConfigOption` and
`ConfigListOption` instances.
"""
result = [v if isinstance(v, Config) else v.value
for v in vars(self).values() if isinstance(v, (Config, ConfigOption))]
for opt in (v for v in vars(self).values() if isinstance(v, ConfigListOption)):
result.extend(opt.value)
return result
# Options
[docs]
class StrOption(Option[str]):
"""Configuration option with string value.
.. versionadded:: 1.6.1
Support for verticals to preserve leading whitespace.
Important:
Multiline string values could contain significant leading whitespace, but
ConfigParser multiline string values have leading whitespace removed. To circumvent
this, the `StrOption` supports assignment of text values where lines start with `|`
character. This character is removed, along with any number of subsequent whitespace
characters that are between `|` and first non-whitespace character on first line
starting with `|`.
"""
[docs]
def __init__(self, name: str, description: str, *, required: bool=False, default: str=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
"""
self._value: str = None
super().__init__(name, str, description, required, default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
value = unindent_verticals(value)
self._value = value
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return self._value
[docs]
def get_value(self) -> str:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: str) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_value(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_string = self._value
value: str = property(get_value, set_value, doc="Current option value")
[docs]
class IntOption(Option[int]):
"""Configuration option with integer value.
"""
[docs]
def __init__(self, name: str, description: str, *, required: bool=False,
default: int=None, signed: bool=False):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
signed: When False, the option value cannot be negative.
"""
self._value: int = None
self.__signed: bool = signed
super().__init__(name, int, description, required, default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
new = int(value)
if not self.__signed and new < 0:
raise ValueError("Negative numbers not allowed")
self._value = new
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return str(self._value)
[docs]
def get_value(self) -> int:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: int) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
if value is not None and (not self.__signed and value < 0):
raise ValueError("Negative numbers not allowed")
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
oneof = opt.WhichOneof('kind')
if oneof not in ['as_sint32', 'as_sint64', 'as_uint32', 'as_uint64', 'as_string']:
raise TypeError(f"Wrong value type: {oneof[3:]}")
if oneof == 'as_string':
self.set_as_str(opt.as_string)
else:
self.set_value(getattr(opt, oneof))
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
opt = proto.options[self.name]
if self.__signed:
opt.as_sint64 = self._value
else:
opt.as_uint64 = self._value
value: int = property(get_value, set_value, doc="Current option value")
[docs]
class FloatOption(Option[float]):
"""Configuration option with float value.
"""
[docs]
def __init__(self, name: str, description: str, *, required: bool=False, default: float=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
"""
self._value: float = None
super().__init__(name, float, description, required, default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
self._value = float(value)
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return str(self._value)
[docs]
def get_value(self) -> float:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: float) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
oneof = opt.WhichOneof('kind')
if oneof not in ['as_float', 'as_double', 'as_string']:
raise TypeError(f"Wrong value type: {oneof[3:]}")
if oneof == 'as_string':
self.set_as_str(opt.as_string)
else:
self.set_value(getattr(opt, oneof))
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_double = self._value
value: float = property(get_value, set_value, doc="Current option value")
[docs]
class DecimalOption(Option[Decimal]):
"""Configuration option with decimal.Decimal value.
"""
[docs]
def __init__(self, name: str, description: str, *, required: bool=False, default: Decimal=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
"""
self._value: Decimal = None
super().__init__(name, Decimal, description, required, default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
try:
self._value = Decimal(value)
except DecimalException as exc:
raise ValueError(str(exc)) from exc
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return str(self._value)
[docs]
def get_value(self) -> Decimal:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: Decimal) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = value
[docs]
def load_proto(self, proto: ConfigProto):
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
oneof = opt.WhichOneof('kind')
if oneof not in ['as_sint32', 'as_sint64', 'as_uint32', 'as_uint64', 'as_string']:
raise TypeError(f"Wrong value type: {oneof[3:]}")
if oneof == 'as_string':
self.set_as_str(opt.as_string)
else:
self.set_value(Decimal(getattr(opt, oneof)))
[docs]
def save_proto(self, proto: ConfigProto):
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_string = str(self._value)
value: Decimal = property(get_value, set_value, doc="Current option value")
[docs]
class BoolOption(Option[bool]):
"""Configuration option with boolean value.
"""
[docs]
def __init__(self, name: str, description: str, *, required: bool=False, default: bool=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
"""
self._value: bool = None
self.from_str = get_convertor(bool).from_str
super().__init__(name, bool, description, required, default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
self._value = self.from_str(bool, value)
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return str(self._value)
[docs]
def get_value(self) -> bool:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: bool) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
oneof = opt.WhichOneof('kind')
if oneof not in ['as_bool', 'as_string']:
raise TypeError(f"Wrong value type: {oneof[3:]}")
if oneof == 'as_string':
self.set_as_str(opt.as_string)
else:
self.set_value(opt.as_bool)
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_bool = self._value
value: bool = property(get_value, set_value, doc="Current option value")
[docs]
class ZMQAddressOption(Option[ZMQAddress]):
"""Configuration option with `.ZMQAddress` value.
"""
[docs]
def __init__(self, name: str, description: str, *, required: bool=False,
default: ZMQAddress=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
"""
self._value: ZMQAddress = None
super().__init__(name, ZMQAddress, description, required, default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
self._value = ZMQAddress(value)
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return self._value
[docs]
def get_value(self) -> ZMQAddress:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: ZMQAddress) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_as_str(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_string = self._value
value: ZMQAddress = property(get_value, set_value, doc="Current option value")
[docs]
class EnumOption(Option[Enum]):
"""Configuration option with enum value.
"""
[docs]
def __init__(self, name: str, enum_class: Enum, description: str, *, required: bool=False,
default: Enum=None, allowed: List=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
allowed: List of allowed Enum members. When not defined, all members of enum type are
allowed.
"""
self._value: Enum = None
#: List of allowed enum values.
self.allowed: Sequence = enum_class if allowed is None else allowed
self._members: Dict = {i.name.lower(): i for i in self.allowed}
super().__init__(name, enum_class, description, required, default)
def _get_value_description(self) -> str:
return f"enum [{', '.join(x.name.lower() for x in self.allowed)}]\n"
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
name = value.lower()
if name in self._members:
self.set_value(self._members[name])
else:
raise ValueError(f"Illegal value '{value}' for enum type "
f"'{self.datatype.__name__}'")
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return self._value.name
[docs]
def get_value(self) -> Enum:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: Enum) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
if value is not None and value not in self.allowed:
raise ValueError(f"Value '{value}' not allowed")
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_as_str(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_string = self._value.name
value: Enum = property(get_value, set_value, doc="Current option value")
[docs]
class FlagOption(Option[Flag]):
"""Configuration option with flag value.
"""
[docs]
def __init__(self, name: str, flag_class: Flag, description: str, *, required: bool=False,
default: Flag=None, allowed: List=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
allowed: List of allowed Flag members. When not defined, all members of flag type are
allowed.
"""
self._value: Flag = None
#: List of allowed flag values.
self.allowed: Sequence = flag_class if allowed is None else allowed
self._members: Dict = {i.name.lower(): i for i in self.allowed}
super().__init__(name, flag_class, description, required, default)
def _get_value_description(self) -> str:
return f"flag [{', '.join(x.name.lower() for x in self.allowed)}]\n"
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
result = self.datatype(0)
for name in (x.strip().lower() for x in value.split('|' if '|' in value else ',')):
if name in self._members:
result |= self._members[name]
else:
raise ValueError(f"Illegal value '{name}' for flag option '{self.name}'")
self.set_value(result)
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
if self._value._name_ is not None:
return self._value.name
members, uncovered = _decompose(self.datatype, self._value)
if len(members) == 1 and members[0]._name_ is None:
return f'{members[0]._value_}'
return ' | '.join([str(m._name_ or m._value_) for m in members])
[docs]
def get_value(self) -> Flag:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: Flag) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
if value is not None:
members, uncovered = _decompose(self.datatype, value.value)
if uncovered or [i for i in members if i.name is None or i.name.lower() not in self._members]:
raise ValueError(f"Illegal value '{value!s}' for flag option '{self.name}'")
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
oneof = opt.WhichOneof('kind')
if oneof not in ['as_uint64', 'as_string']:
raise TypeError(f"Wrong value type: {oneof[3:]}")
if oneof == 'as_uint64':
self.set_value(self.datatype(opt.as_uint64))
else:
self.set_as_str(opt.as_string)
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_uint64 = self._value.value
value: Flag = property(get_value, set_value, doc="Current option value")
[docs]
class UUIDOption(Option[UUID]):
"""Configuration option with UUID value.
"""
[docs]
def __init__(self, name: str, description: str, *, required: bool=False, default: UUID=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
"""
self._value: UUID = None
super().__init__(name, UUID, description, required, default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
self._value = UUID(value)
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return 'None' if self._value is None else self._value.hex
[docs]
def get_value(self) -> UUID:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: UUID) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
oneof = opt.WhichOneof('kind')
if oneof not in ['as_bytes', 'as_string']:
raise TypeError(f"Wrong value type: {oneof[3:]}")
if oneof == 'as_bytes':
self.set_value(UUID(bytes=opt.as_bytes))
else:
self.set_value(UUID(opt.as_string))
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_bytes = self._value.bytes
value: UUID = property(get_value, set_value, doc="Current option value")
[docs]
class MIMEOption(Option[MIME]):
"""Configuration option with MIME type specification value.
"""
[docs]
def __init__(self, name: str, description: str, *, required: bool=False, default: MIME=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
"""
self._value: MIME = None
super().__init__(name, MIME, description, required, default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
self._value = MIME(value)
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return 'None' if self._value is None else self._value
[docs]
def get_value(self) -> MIME:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: MIME) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_as_str(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_string = self._value
value: MIME = property(get_value, set_value, doc="Current option value")
[docs]
class ListOption(Option[List]):
"""Configuration option with list of values.
Important:
When option is read from `ConfigParser`, empty values are ignored.
"""
[docs]
def __init__(self, name: str, item_type: Union[Type, Sequence[Type]], description: str,
*, required: bool=False, default: List=None, separator: str=None):
"""
Arguments:
name: Option name.
item_type: Datatype of list items. It could be a type or sequence of types.
If multiple types are provided, each value in config file must
have format: `type_name:value_as_str`.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
separator: String that separates list item values when options value is read
from `ConfigParser`. It's possible to use a line break as separator.
If separator is `None` [default] and the value contains line breaks,
it uses the line break as separator, otherwise it uses comma as
separator.
"""
self._value: List = None
#: Datatypes of list items. If there is more than one type, each value in
#: config file must have format: `type_name:value_as_str`.
self.item_types: Sequence[Type] = (item_type, ) if isinstance(item_type, type) else item_type
#: String that separates list item values when options value is read from
#: `ConfigParser`. Default separator is None. It's possible to use a line break as
#: separator. If separator is `None` and the value contains line breaks, it uses
#: the line break as separator, otherwise it uses comma as separator.
self.separator: Optional[str] = separator
self._convertor: Convertor = get_convertor(item_type) if isinstance(item_type, type) else None
super().__init__(name, list, description, required, default)
def _get_value_description(self) -> str:
return f"list [{', '.join(x.__name__ for x in self.item_types)}]\n"
def _check_value(self, value: List) -> None:
super()._check_value(value)
if value is not None:
i = 0
for item in value:
if item.__class__ not in self.item_types:
raise ValueError(f"List item[{i}] has wrong type")
i += 1
def _get_as_typed_str(self, value: Any) -> str:
result = convert_to_str(value)
if len(self.item_types) > 1:
result = f'{value.__class__.__name__}:{result}'
return result
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
new = []
if value.strip():
separator = ('\n' if '\n' in value else ',') if self.separator is None else self.separator
itype = self.item_types[0]
convertor = self._convertor
name_map = {}
if len(self.item_types) > 1:
name_map = {cls.__name__: cls for cls in self.item_types}
fullname_map = {f'{cls.__module__}.{cls.__name__}': cls for cls in self.item_types}
for item in (i for i in value.split(separator) if i.strip()):
if name_map:
itype_name, item = item.split(':', 1)
itype_name = itype_name.strip()
itype = fullname_map.get(itype_name) if '.' in itype_name else name_map.get(itype_name)
if itype is None:
raise ValueError(f"Item type '{itype_name}' not supported")
convertor = get_convertor(itype)
new.append(convertor.from_str(itype, item.strip()))
self._value = new
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
result = [convert_to_str(i) for i in self._value]
sep = self.separator
if sep is None:
sep = '\n' if sum(len(i) for i in result) > 80 else ','
return sep.join(result)
[docs]
def get_value(self) -> List:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: List) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = None if value is None else list(value)
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_as_str(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
result = [self._get_as_typed_str(i) for i in self._value]
sep = self.separator
if sep is None:
sep = '\n' if sum(len(i) for i in result) > 80 else ','
proto.options[self.name].as_string = sep.join(result)
value: List = property(get_value, set_value, doc="Current option value")
[docs]
class PyExprOption(Option[PyExpr]):
"""String configuration option with Python expression value.
"""
def __init__(self, name: str, description: str, *, required: bool=False, default: PyExpr=None):
self._value: PyExpr = None
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
"""
super().__init__(name, PyExpr, description, required, default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
self._value = PyExpr(value)
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return self._value
[docs]
def get_value(self) -> PyExpr:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: PyExpr) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_as_str(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_string = self._value
value: PyExpr = property(get_value, set_value, doc="Current option value")
[docs]
class PyCodeOption(Option[PyCode]):
"""String configuration option with Python code value.
Important:
Python code must be properly indented, but ConfigParser multiline string values have
leading whitespace removed. To circumvent this, the `PyCodeOption` supports assignment
of text values where lines start with `|` character. This character is removed, along
with any number of subsequent whitespace characters that are between `|` and first
non-whitespace character on first line starting with `|`.
"""
def __init__(self, name: str, description: str, *, required: bool=False, default: PyCode=None):
self._value: PyCode = None
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
"""
super().__init__(name, PyCode, description, required, default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
value = unindent_verticals(value)
self._value = PyCode(value)
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return self._value
[docs]
def get_value(self) -> PyCode:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: PyCode) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_as_str(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_string = self._value
value: PyCode = property(get_value, set_value, doc="Current option value")
[docs]
class PyCallableOption(Option[PyCallable]):
"""String configuration option with Python callable value.
Important:
Python code must be properly indented, but `ConfigParser` multiline string values have
leading whitespace removed. To circumvent this, the `PyCodeOption` supports assignment
of text values where lines start with `|` character. This character is removed, along
with any number of subsequent whitespace characters that are between `|` and first
non-whitespace character on first line starting with `|`.
"""
# pylint: disable=[W0621]
[docs]
def __init__(self, name: str, description: str, signature: Union[Signature, Callable], * ,
required: bool=False, default: PyCallable=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
signature: Callable signature or callable.
required: True if option must have a value.
default: Default option value.
"""
self._value: PyCallable = None
#: Callable signature.
if not isinstance(signature, Signature):
signature = Signature.from_callable(PyCallable(signature)._callable_)
self.signature: Signature = signature
super().__init__(name, PyCallable, description, required=required, default=default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
value = unindent_verticals(value)
self.set_value(PyCallable(value))
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return self._value
[docs]
def get_value(self) -> PyCallable:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: PyCallable) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the callable has wrong signature.
"""
self._check_value(value)
if value is not None:
val_sig = signature(value._callable_)
if not _eq(val_sig.return_annotation, self.signature.return_annotation):
raise ValueError("Wrong callable return type")
if len(val_sig.parameters) != len(self.signature.parameters):
raise ValueError("Wrong number of parameters")
for par in self.signature.parameters.values():
val_par: Parameter = val_sig.parameters[cast(Signature, par).name]
if not _eq(val_par.annotation, cast(Signature, par).annotation):
raise ValueError(f"Wrong type, parameter '{val_par.name}'")
if not _eq(val_par.default, cast(Signature, par).default):
raise ValueError(f"Wrong default, parameter '{val_par.name}'")
if not _eq(val_par.kind, cast(Signature, par).kind):
raise ValueError(f"Wrong parameter kind, parameter '{val_par.name}'")
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_as_str(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_string = self._value
value: PyCallable = property(get_value, set_value, doc="Current option value")
[docs]
class ConfigOption(Option[str]):
"""Configuration option with `Config` value.
Important:
This option is intended for sub-configs that should have *configurable* name (i.e. the
section name that holds sub-config values). To create sub-configs with fixed section
names, simply assign them to instance attributes of `Config` instance that owns them
(preferably in constructor).
While the `value` attribute for this option is an instance of any class inherited from
`Config`, in other ways it behaves like `StrOption` that loads/saves only name of its
`Config` value (i.e. the section name). The actual I/O for sub-config's options is
delegated to `Config` instance that owns this option.
The "empty" value for this option is not `None` (because the `Config` instance always
exists), but an empty string for `Config.name` attribute.
"""
[docs]
def __init__(self, name: str, description: str, config: Config, *, required: bool=False,
default: str=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
config: Option's value.
required: True if option must have a value.
default: Default `Config.name` value.
"""
assert isinstance(config, Config)
self._value: Config = config
super().__init__(name, str, description, required, default)
def _get_value_description(self) -> str:
return "configuration section name\n"
[docs]
def validate(self) -> None:
"""Validates option state.
Raises:
Error: When required option does not have a value.
"""
if self.required and self.get_value().name == '':
raise Error(f"Missing value for required option '{self.name}'")
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Note:
This method calls `~Config.clear(to_default)`.
Arguments:
to_default: If True, sets the `Config.name` to default value, else to empty string.
"""
self._value.clear(to_default=to_default)
self._value.name = self.default if to_default else ''
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New `Config.name` value.
Important:
Because the actual value is a `Config` instance, the string must contain the
`Config.name` value (which is the section name used to store `Config` options).
Beware that multiple Config instances with the same (section) name may cause
collision when configuration is written to protobuf message or configuration file.
"""
self._value.name = value
[docs]
def get_as_str(self) -> str:
"""Return value as string.
Important:
Because the actual value is a `Config` instance, the returned string is the section
name used to store `Config` options.
"""
return self._value.name
[docs]
def get_value(self) -> Config:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: str) -> None:
"""Set new option value.
This option type does not support direct assignment of `Config` value. Because this method
is also used to assign default value (which is a `Config.name`), it accepts None or string
argument that is interpreted as new Config name. `None` value is translated to empty string.
Arguments:
value: New `Config` name.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When None or empty string is passed and option value is required.
"""
if value is None:
value = ''
if value == '' and self.required:
raise ValueError(f"Value is required for option '{self.name}'.")
self._value.name = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_as_str(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_string = self._value.name
value: Config = property(get_value, set_value, doc="Current option value")
[docs]
class ConfigListOption(Option[List]):
"""Configuration option with list of `Config` values.
Important:
This option is intended for configurable set of sub-configs of fixed type.
While the `value` attribute for this option is a list of instances of single class
inherited from `Config`, in other ways it behaves like `ListOption` with `str` items
that loads/saves only names of its `Config` items (i.e. the section names). The actual
I/O for sub-config options is delegated to `Config` instance that owns this option.
Important:
When option is read from `ConfigParser`, empty values are ignored.
"""
[docs]
def __init__(self, name: str, description: str, item_type: Type[Config], *,
required: bool=False, separator: str=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
item_type: Datatype of list items. Must be subclass of `Config`.
required: True if option must have a value.
separator: String that separates values when options value is read from `ConfigParser`.
It's possible to use a line break as separator.
If separator is `None` [default] and the value contains line breaks, it uses
the line break as separator, otherwise it uses comma as separator.
"""
assert issubclass(item_type, Config)
self._value: List = []
#: Datatype of list items.
self.item_type: Type[Config] = item_type
#: String that separates values when options value is read from `ConfigParser`.
#: Default separator is None. It's possible to use a line break as separator.
#: If separator is `None` and the value contains line breaks, it uses the line
#: break as separator, otherwise it uses comma as separator.
self.separator: Optional[str] = separator
super().__init__(name, list, description, required, [])
def _get_value_description(self) -> str:
return "list of configuration section names\n"
def _check_value(self, value: List) -> None:
super()._check_value(value)
if value is not None:
i = 0
for item in value:
if item.__class__ is not self.item_type:
raise ValueError(f"List item[{i}] has wrong type")
i += 1
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value.clear()
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
new = []
if value.strip():
separator = ('\n' if '\n' in value else ',') if self.separator is None else self.separator
for item in (i for i in value.split(separator) if i.strip()):
new.append(self.item_type(item.strip()))
self._value = new
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
result = [i.name for i in self._value]
sep = self.separator
if sep is None:
sep = '\n' if sum(len(i) for i in result) > 80 else ','
return sep.join(result)
[docs]
def get_value(self) -> List:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: List) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
if value is None:
self.clear()
else:
self._value = list(value)
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_as_str(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
result = [i.name for i in self._value]
sep = self.separator
if sep is None:
sep = '\n' if sum(len(i) for i in result) > 80 else ','
proto.options[self.name].as_string = sep.join(result)
value: List = property(get_value, set_value, doc="Current option value")
[docs]
class DataclassOption(Option[Any]):
"""Configuration option with a dataclass value.
The `ConfigParser` format for this option is a list of values, where each list items
defines value for dataclass field in `field_name:value_as_str` format. The configuration
must contain values for all fields for the dataclass that does not have default value.
Important:
This option uses type annotation for dataclass to determine the actual data type for
conversion from string. It means that:
1. If type annotation contains "typing" types, it's necessary to specify "real" types
for all dataclass fields using the `fields` argument.
2. All used data types must have string convertors registered in `strconv` module.
Important:
When option is read from `ConfigParser`, empty values are ignored.
"""
[docs]
def __init__(self, name: str, dataclass: Type, description: str, *, required: bool=False,
default: Any=None, separator: str=None, fields: Dict[str, Type]=None):
"""
Arguments:
name: Option name.
dataclass: Dataclass type.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
separator: String that separates dataclass field values when options value is read
from `ConfigParser`. It's possible to use a line break as separator.
If separator is `None` [default] and the value contains line breaks, it
uses the line break as separator, otherwise it uses comma as separator.
fields: Dictionary that maps dataclass field names to data types.
"""
assert hasattr(dataclass, '__dataclass_fields__')
self._fields: Dict[str, Type] = get_type_hints(dataclass) if fields is None else fields
if __debug__:
for ftype in self._fields.values():
assert get_convertor(ftype) is not None
self._value: Any = None
#: Dataclass type.
self.dataclass: Type = dataclass
#: String that separates dataclass field values when options value is read from
#: `ConfigParser`. Default separator is None. It's possible to use a line break
#: as separator. If separator is `None` and the value contains line breaks, it
#: uses the line break as separator, otherwise it uses comma as separator.
self.separator: Optional[str] = separator
super().__init__(name, dataclass, description, required, default)
def _get_value_description(self) -> str:
return "list of values, where each list item defines value for a dataclass field.\n" \
"Item format: field_name:value_as_str\n"
def _get_str_fields(self) -> List[str]:
result = []
if self._value is not None:
for fname in self._fields:
result.append(f'{fname}:{convert_to_str(getattr(self._value, fname))}')
return result
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
new = {}
if value.strip():
separator = ('\n' if '\n' in value else ',') if self.separator is None else self.separator
for item in (i for i in value.split(separator) if i.strip()):
try:
field_name, field_value = item.split(':', 1)
except Exception as exc:
raise ValueError(f"Illegal value '{value}' for option '{self.name}'") from exc
field_name = field_name.strip()
ftype = self._fields.get(field_name)
if ftype is None:
raise ValueError(f"Unknown data field '{field_name}' for option '{self.name}'")
convertor = get_convertor(ftype)
new[field_name] = convertor.from_str(ftype, field_value.strip())
try:
new_val = self.dataclass(**new)
except Exception as exc:
raise ValueError(f"Illegal value '{value}' for option '{self.name}'") from exc
self._value = new_val
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
result = self._get_str_fields()
sep = self.separator
if sep is None:
sep = '\n' if sum(len(i) for i in result) > 80 else ','
return sep.join(result)
[docs]
def get_value(self) -> Any:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: Any) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_as_str(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
result = self._get_str_fields()
sep = self.separator
if sep is None:
sep = '\n' if sum(len(i) for i in result) > 80 else ','
proto.options[self.name].as_string = sep.join(result)
value: Any = property(get_value, set_value, doc="Current option value")
[docs]
class PathOption(Option[str]):
"""Configuration option with `pathlib.Path` value.
"""
[docs]
def __init__(self, name: str, description: str, *, required: bool=False, default: Path=None):
"""
Arguments:
name: Option name.
description: Option description. Can span multiple lines.
required: True if option must have a value.
default: Default option value.
"""
self._value: Path = None
super().__init__(name, Path, description, required, default)
[docs]
def clear(self, *, to_default: bool=True) -> None:
"""Clears the option value.
Arguments:
to_default: If True, sets the option value to default value, else to None.
"""
self._value = self.default if to_default else None
[docs]
def set_as_str(self, value: str) -> None:
"""Set new option value from string.
Arguments:
value: New option value.
Raises:
ValueError: When the argument is not a valid option value.
"""
self._value = Path(value)
[docs]
def get_as_str(self) -> str:
"""Returns value as string.
"""
return str(self._value)
[docs]
def get_value(self) -> Path:
"""Returns current option value.
"""
return self._value
[docs]
def set_value(self, value: Path) -> None:
"""Set new option value.
Arguments:
value: New option value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
self._check_value(value)
self._value = value
[docs]
def load_proto(self, proto: ConfigProto) -> None:
"""Deserialize value from `.ConfigProto` message.
Arguments:
proto: Protobuf message that may contains options value.
Raises:
TypeError: When the new value is of the wrong type.
ValueError: When the argument is not a valid option value.
"""
if self.name in proto.options:
opt = proto.options[self.name]
if opt.HasField('as_string'):
self.set_as_str(opt.as_string)
else:
raise TypeError(f"Wrong value type: {opt.WhichOneof('kind')[3:]}")
[docs]
def save_proto(self, proto: ConfigProto) -> None:
"""Serialize value into `.ConfigProto` message.
Arguments:
proto: Protobuf message where option value should be stored.
"""
if self._value is not None:
proto.options[self.name].as_string = self.get_as_str()
value: Path = property(get_value, set_value, doc="Current option value")