Source code for karton.core.config

import configparser
import os
import re
import warnings
from typing import Any, Dict, List, Optional, cast, overload


[docs]class Config(object): """ Simple config loader. Loads configuration from paths specified below (in provided order): - ``/etc/karton/karton.ini`` (global) - ``~/.config/karton/karton.ini`` (user local) - ``./karton.ini`` (subsystem local) - ``<path>`` optional, additional path provided in arguments It is also possible to pass configuration via environment variables. Any variable named KARTON_FOO_BAR is equivalent to setting 'bar' variable in section 'foo' (note the lowercase names). Environment variables have higher precedence than those loaded from files. :param path: Path to additional configuration file :param check_sections: Check if sections ``redis`` and ``s3`` are defined in the configuration """ SEARCH_PATHS = [ "/etc/karton/karton.ini", os.path.expanduser("~/.config/karton/karton.ini"), "./karton.ini", ] def __init__( self, path: Optional[str] = None, check_sections: Optional[bool] = True ) -> None: self._config: Dict[str, Dict[str, Any]] = {} if path is not None: if not os.path.isfile(path): raise IOError("Configuration file not found in " + path) self.SEARCH_PATHS = self.SEARCH_PATHS + [path] self._load_from_file(self.SEARCH_PATHS) self._load_from_env() if check_sections: if self.has_section("minio") and not self.has_section("s3"): self._map_minio_to_s3() if not self.has_section("s3"): raise RuntimeError("Missing S3 configuration") if not self.has_section("redis"): raise RuntimeError("Missing Redis configuration") def _map_minio_to_s3(self): """ Configuration backwards compatibility. Before 5.x.x [minio] section was used. """ warnings.warn( "[minio] section in configuration is deprecated, replace it with [s3]" ) self._config["s3"] = dict(self._config["minio"]) if not ( self._config["s3"]["address"].startswith("http://") or self._config["s3"]["address"].startswith("https://") ): if self.getboolean("minio", "secure", True): self._config["s3"]["address"] = ( "https://" + self._config["s3"]["address"] ) else: self._config["s3"]["address"] = ( "http://" + self._config["s3"]["address"] )
[docs] def set(self, section_name: str, option_name: str, value: Any) -> None: """ Sets value in configuration """ if section_name not in self._config: self._config[section_name] = {} self._config[section_name][option_name] = value
[docs] def get( self, section_name: str, option_name: str, fallback: Optional[Any] = None ) -> Any: """ Gets value from configuration or returns ``fallback`` (None by default) if value was not set. """ if not self.has_option(section_name, option_name): return fallback return self._config[section_name][option_name]
[docs] def has_section(self, section_name: str) -> bool: """ Checks if configuration section exists """ return section_name in self._config
[docs] def has_option(self, section_name: str, option_name: str) -> bool: """ Checks if configuration value is set """ if section_name not in self._config: return False if option_name not in self._config[section_name]: return False return True
@overload def getint(self, section_name: str, option_name: str, fallback: int) -> int: ... @overload def getint(self, section_name: str, option_name: str) -> Optional[int]: ...
[docs] def getint( self, section_name: str, option_name: str, fallback: Optional[int] = None ) -> Optional[int]: """ Gets value from configuration or returns ``fallback`` (None by default) if value was not set. Value is coerced to int type. """ value = self.get(section_name, option_name, fallback) if value is None: return None return int(value)
@overload def getboolean(self, section_name: str, option_name: str, fallback: bool) -> bool: ... @overload def getboolean(self, section_name: str, option_name: str) -> Optional[bool]: ...
[docs] def getboolean( self, section_name: str, option_name: str, fallback: Optional[bool] = None ) -> Optional[bool]: """ Gets value from configuration or returns ``fallback`` (None by default) if value was not set. Value is coerced to bool type. .. seealso:: https://docs.python.org/3/library/configparser.html#configparser.ConfigParser.getboolean """ value = self.get(section_name, option_name, fallback) if value is None: return None if type(value) is bool: return value if type(value) is str and value.lower() in ["1", "yes", "true", "on"]: return True elif type(value) is str and value.lower() in ["0", "no", "false", "off"]: return False else: raise ValueError(f"{section_name}.{option_name} is not a correct boolean")
[docs] def append_to_list(self, section_name: str, option_name: str, value: Any) -> None: """ Appends value to a list in configuration """ if section_name not in self._config: self._config[section_name] = {} if option_name not in self._config[section_name]: self._config[section_name][option_name] = [] elif not isinstance(self._config[section_name][option_name], list): raise TypeError( f"{section_name}.{option_name} is " f"{type(self._config[section_name][option_name])} while " f"list was expected" ) self._config[section_name][option_name].append(value)
[docs] def load_from_dict(self, data: Dict[str, Dict[str, Any]]) -> None: """ Updates configuration values from dictionary compatible with ``ConfigParser.read_dict``. Accepts value in native type, so you don't need to convert them to string. None values are treated like missing value and are not added. .. code-block:: json { "section-name": { "option-name": "value" } } """ for section_name, section in data.items(): for option_name, value in section.items(): if value is None: continue self.set(section_name, option_name, value)
def _load_from_file(self, paths: List[str]) -> None: """ Function used for loading configuration items from karton.ini files :meta private: """ config_file = configparser.ConfigParser() config_file.read(paths) self.load_from_dict(cast(Dict[str, Dict[str, Any]], config_file)) def _load_from_env(self) -> None: """ Function used for loading configuration items from the environment variables :meta private: """ for name, value in os.environ.items(): # Load env variables named KARTON_[section]_[key] # to match ConfigParser structure result = re.fullmatch(r"KARTON_([A-Z0-9]+)_([A-Z0-9_]+)", name) if not result: continue section, key = result.groups() section = section.lower() key = key.lower() self.set(section, key, value) def __getitem__(self, section) -> Dict[str, Any]: """Gets a section named `section` from the config""" return self._config[section]