Source code for PyFunceble.config.loader

The tool to check the availability or syntax of domain, IP or URL.


    ██████╗ ██╗   ██╗███████╗██╗   ██╗███╗   ██╗ ██████╗███████╗██████╗ ██╗     ███████╗
    ██╔══██╗╚██╗ ██╔╝██╔════╝██║   ██║████╗  ██║██╔════╝██╔════╝██╔══██╗██║     ██╔════╝
    ██████╔╝ ╚████╔╝ █████╗  ██║   ██║██╔██╗ ██║██║     █████╗  ██████╔╝██║     █████╗
    ██╔═══╝   ╚██╔╝  ██╔══╝  ██║   ██║██║╚██╗██║██║     ██╔══╝  ██╔══██╗██║     ██╔══╝
    ██║        ██║   ██║     ╚██████╔╝██║ ╚████║╚██████╗███████╗██████╔╝███████╗███████╗
    ╚═╝        ╚═╝   ╚═╝      ╚═════╝ ╚═╝  ╚═══╝ ╚═════╝╚══════╝╚═════╝ ╚══════╝╚══════╝

Provides the configuration loader.

    Nissar Chababy, @funilrys, contactTATAfunilrysTODTODcom

Special thanks:


Project link:

Project documentation:

Project homepage:


    Copyright 2017, 2018, 2019, 2020, 2022, 2023 Nissar Chababy

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.

import functools
import os
from typing import Any, Optional

    import importlib.resources as package_resources
except ImportError:  # pragma: no cover ## Retro compatibility
    import importlib_resources as package_resources

from box import Box
from yaml.error import MarkedYAMLError

from import ConfigComparison
from PyFunceble.downloader.iana import IANADownloader
from PyFunceble.downloader.public_suffix import PublicSuffixDownloader
from PyFunceble.downloader.user_agents import UserAgentsDownloader
from PyFunceble.helpers.dict import DictHelper
from PyFunceble.helpers.environment_variable import EnvironmentVariableHelper
from PyFunceble.helpers.file import FileHelper
from PyFunceble.helpers.merge import Merge

[docs]class ConfigLoader: """ Provides the interface which loads and updates the configuration (if needed). :param merge_upstream: Authorizes the merging of the upstream configuration. .. note:: If value is set to :py:class:`None` (default), we fallback to the :code:`PYFUNCEBLE_AUTO_CONFIGURATION` environment variable. """ path_to_config: Optional[str] = None path_to_default_config: Optional[str] = None path_to_overwrite_config: Optional[str] = None _custom_config: dict = {} _merge_upstream: bool = False file_helper: FileHelper = FileHelper() dict_helper: DictHelper = DictHelper() def __init__(self, merge_upstream: Optional[bool] = None) -> None: with package_resources.path( "",, ) as file_path: self.path_to_default_config = str(file_path) self.path_to_config = os.path.join(,, ) self.path_to_overwrite_config = os.path.join(,, ) if merge_upstream is not None: self.merge_upstream = merge_upstream elif EnvironmentVariableHelper("PYFUNCEBLE_AUTO_CONFIGURATION").exists(): self.merge_upstream = True def __del__(self) -> None: self.destroy()
[docs] def reload_config(func): # pylint: disable=no-self-argument """ Reload the configuration (if it was already loaded) after launching the decorated method. """ @functools.wraps(func) def wrapper(self, *args, **kwargs): result = func(self, *args, **kwargs) # pylint: disable=not-callable if self.is_already_loaded(): self.start() return result return wrapper
[docs] @staticmethod def conditional_switch(config: dict) -> dict: """ Given the configuration that we are going to load, switches some of setting. :param config: The configuration we are going to load. """ # pylint: disable=too-many-boolean-expressions if ( "cli_testing" in config and "ci" in config["cli_testing"] and "active" in config["cli_testing"]["ci"] and "autocontinue" in config["cli_testing"] and bool(config["cli_testing"]["ci"]["active"]) and not bool(config["cli_testing"]["autocontinue"]) ): # Conditional autocontinue. # If we are under continuous integration, the autocontinue should be # activated. config["cli_testing"]["autocontinue"] = True if ( "lookup" in config and "timeout" in config["lookup"] and config["lookup"]["timeout"] and config["lookup"]["timeout"] < 0 ): # If timeout is set to a negative digit, switch to the default one. config["lookup"]["timeout"] = 5 return config
[docs] @staticmethod def is_already_loaded() -> bool: """ Checks if the configuration was already loaded. """ return bool(
@property def custom_config(self) -> dict: """ Provides the current state of the :code:`_custom_config` attribute. """ return self._custom_config @custom_config.setter @reload_config def custom_config(self, value: dict) -> None: """ Sets the custom configuration to set after loading. Side Effect: Directly inject into the configuration variables if it was already loaded. :raise TypeError: When :code:`value` is not a :py:class:`dict`. """ if not isinstance(value, dict): raise TypeError(f"<value> should be {dict}, {type(value)} given.") if not self._custom_config: self._custom_config = value else: self._custom_config.update(value)
[docs] def set_custom_config(self, value: dict) -> "ConfigLoader": """ Sets the custom configuration to set after loading. Side Effect: Directly inject into the configuration variables if it was already loaded. """ self.custom_config = value return self
@property def merge_upstream(self) -> bool: """ Provides the current state of the :code:`_merge_upstream` attribute. """ return self._merge_upstream @merge_upstream.setter def merge_upstream(self, value: bool) -> None: """ Updates the value of :code:`_merge_upstream` attribute. :raise TypeError: When :code:`value` is not a :py:class:`bool`. """ if not isinstance(value, bool): raise TypeError(f"<value> should be {bool}, {type(value)} given.") self._merge_upstream = value
[docs] def set_merge_upstream(self, value: bool) -> "ConfigLoader": """ Updates the value of :code:`_merge_upstream` attribute. """ self.merge_upstream = value return self
[docs] def config_file_exist( self, ) -> bool: # pragma: no cover ## Existance checker already tested. """ Checks if the config file exists. """ return FileHelper(self.path_to_config).exists()
[docs] def default_config_file_exist( self, ) -> bool: # pragma: no cover ## Existance checker already tested. """ Checks if the default configuration file exists. """ return self.file_helper.set_path(self.path_to_default_config).exists()
[docs] def install_missing_infrastructure_files( self, ) -> "ConfigLoader": # pragma: no cover ## Copy method already tested """ Installs the missing files (when needed). .. note:: Installed if missing: - The configuration file. - The directory structure file. """ if not self.is_already_loaded(): if not self.file_helper.set_path(self.path_to_config).exists(): self.file_helper.set_path(self.path_to_default_config).copy( self.path_to_config ) return self
[docs] @classmethod def download_dynamic_infrastructure_files( cls, ) -> "ConfigLoader": """ Downloads all the dynamicly (generated) infrastructure files. .. note:: Downloaded if missing: - The IANA dump file. - The Public Suffix dump file. """ ## pragma: no cover ## Underlying download methods already tested. if not cls.is_already_loaded(): IANADownloader().start() PublicSuffixDownloader().start() UserAgentsDownloader().start()
[docs] def get_config_file_content(self) -> dict: """ Provides the content of the configuration file or the one already loaded. """ def is_3_x_version(config: dict) -> bool: """ Checks if the given configuration is an old one. :param config: The config to work with. """ return config and "days_between_inactive_db_clean" in config if not self.is_already_loaded(): self.install_missing_infrastructure_files() self.download_dynamic_infrastructure_files() try: config = self.dict_helper.from_yaml_file(self.path_to_config) except MarkedYAMLError: self.file_helper.set_path(self.path_to_default_config).copy( self.path_to_config ) config = self.dict_helper.from_yaml_file(self.path_to_config) config_comparer = ConfigComparison( local_config=config, upstream_config=self.dict_helper.from_yaml_file( self.path_to_default_config ), ) if ( not config or not isinstance(config, dict) or self.merge_upstream or is_3_x_version(config) or not config_comparer.is_local_identical() ): # pragma: no cover ## Testing the underlying comparison method is sufficent config = config_comparer.get_merged() self.dict_helper.set_subject(config).to_yaml_file(self.path_to_config) if self.file_helper.set_path(self.path_to_overwrite_config).exists(): overwrite_data = self.dict_helper.from_yaml_file( self.path_to_overwrite_config ) if isinstance(overwrite_data, dict): config = Merge( self.dict_helper.from_yaml_file(self.path_to_overwrite_config) ).into(config) else: # pragma: no cover ## Just make it visible to end-user. self.file_helper.write("") return config
[docs] def get_configured_value(self, entry: str) -> Any: """ Provides the currently configured value. :param entry: An entry to check. multilevel should be separated with a point. :raise RuntimeError: When the configuration is not loaded yet. :raise ValueError: When the given :code:`entry` is not found. """ if not self.is_already_loaded(): raise RuntimeError("Configuration not loaded, yet.") if entry not in raise ValueError(f"<entry> ({entry!r}) not in loaded configuration.") return[entry]
[docs] def start(self) -> "ConfigLoader": """ Starts the loading processIs. """ config = self.get_config_file_content() if self.custom_config: config = Merge(self.custom_config).into(config) config = self.conditional_switch(config) = Box( config, ) = DictHelper( ).flatten() = Box( config["http_codes"], ) if "collection" in config and config["collection"]: = Box(config["collection"]) = Box(config["links"]) if "proxy" in config and config["proxy"]: = Box(config["proxy"]) return self
[docs] def destroy(self) -> "ConfigLoader": """ Destroys everything loaded. """ try: = Box( {}, ) = {} = Box({}) = Box({}) = Box({}) = Box({}) except (AttributeError, TypeError): # pragma: no cover ## Safety. pass # This is not a mistake. self._custom_config = {} return self