"""
Module containing classes to access the bandersnatch configuration file
"""
import configparser
import importlib.resources
import logging
from pathlib import Path
from typing import Any, NamedTuple
from .config.diff_file_reference import eval_config_reference, has_config_reference
from .simple import SimpleDigest, SimpleFormat, get_digest_value, get_format_value
logger = logging.getLogger("bandersnatch")
[docs]
class SetConfigValues(NamedTuple):
json_save: bool
root_uri: str
diff_file_path: str
diff_append_epoch: bool
digest_name: str
storage_backend_name: str
cleanup: bool
release_files_save: bool
compare_method: str
download_mirror: str
download_mirror_no_fallback: bool
simple_format: SimpleFormat
[docs]
class Singleton(type): # pragma: no cover
_instances: dict["Singleton", type] = {}
def __call__(cls, *args: Any, **kwargs: Any) -> type:
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
[docs]
class BandersnatchConfig(metaclass=Singleton):
# Ensure we only show the deprecations once
SHOWN_DEPRECATIONS = False
def __init__(self, config_file: str | None = None) -> None:
"""
Bandersnatch configuration class singleton
This class is a singleton that parses the configuration once at the
start time.
Parameters
==========
config_file: str, optional
Path to the configuration file to use
"""
self.found_deprecations: list[str] = []
self.default_config_file = str(
importlib.resources.files("bandersnatch") / "default.conf"
)
self.config_file = config_file
self.load_configuration()
# Keeping for future deprecations ... Commenting to save function call etc.
# self.check_for_deprecations()
[docs]
def check_for_deprecations(self) -> None:
if self.SHOWN_DEPRECATIONS:
return
self.SHOWN_DEPRECATIONS = True
[docs]
def load_configuration(self) -> None:
"""
Read the configuration from a configuration file
"""
config_file = self.default_config_file
if self.config_file:
config_file = self.config_file
self.config = configparser.ConfigParser(delimiters="=")
# mypy is unhappy with us assigning to a method - (monkeypatching?)
self.config.optionxform = lambda option: option # type: ignore
self.config.read(config_file)
[docs]
def validate_config_values( # noqa: C901
config: configparser.ConfigParser,
) -> SetConfigValues:
try:
json_save = config.getboolean("mirror", "json")
except configparser.NoOptionError:
logger.error(
"Please update your config to include a json "
+ "boolean in the [mirror] section. Setting to False"
)
json_save = False
try:
root_uri = config.get("mirror", "root_uri")
except configparser.NoOptionError:
root_uri = ""
try:
diff_file_path = config.get("mirror", "diff-file")
except configparser.NoOptionError:
diff_file_path = ""
if diff_file_path and has_config_reference(diff_file_path):
try:
diff_file_path = eval_config_reference(config, diff_file_path)
except ValueError as err:
logger.error(
"Invalid section reference in `diff-file` key: %s. Saving diff files in base mirror directory.",
str(err),
)
mirror_dir = config.get("mirror", "directory")
diff_file_path = (Path(mirror_dir) / "mirrored-files").as_posix()
try:
diff_append_epoch = config.getboolean("mirror", "diff-append-epoch")
except configparser.NoOptionError:
diff_append_epoch = False
try:
logger.debug("Checking config for storage backend...")
storage_backend_name = config.get("mirror", "storage-backend")
logger.debug("Found storage backend in config!")
except configparser.NoOptionError:
storage_backend_name = "filesystem"
logger.debug(
"Failed to find storage backend in config, falling back to default!"
)
logger.info(f"Selected storage backend: {storage_backend_name}")
try:
digest_name = get_digest_value(config.get("mirror", "digest_name"))
except configparser.NoOptionError:
digest_name = SimpleDigest.SHA256
logger.debug(f"Using digest {digest_name} by default ...")
except ValueError as e:
logger.error(
f"Supplied digest_name {config.get('mirror', 'digest_name')} is "
+ "not supported! Please update the digest_name in the [mirror] "
+ "section of your config to a supported digest value."
)
raise e
try:
cleanup = config.getboolean("mirror", "cleanup")
except configparser.NoOptionError:
logger.debug(
"bandersnatch is not cleaning up non PEP 503 normalized Simple "
+ "API directories"
)
cleanup = False
release_files_save = config.getboolean("mirror", "release-files", fallback=True)
if not release_files_save and not root_uri:
root_uri = "https://files.pythonhosted.org"
logger.error(
"Please update your config to include a root_uri in the [mirror] "
+ "section when disabling release file sync. Setting to "
+ root_uri
)
try:
logger.debug("Checking config for compare method...")
compare_method = config.get("mirror", "compare-method")
logger.debug("Found compare method in config!")
except configparser.NoOptionError:
compare_method = "hash"
logger.debug(
"Failed to find compare method in config, falling back to default!"
)
if compare_method not in ("hash", "stat"):
raise ValueError(
f"Supplied compare_method {compare_method} is not supported! Please "
+ "update compare_method to one of ('hash', 'stat') in the [mirror] "
+ "section."
)
logger.info(f"Selected compare method: {compare_method}")
try:
logger.debug("Checking config for alternative download mirror...")
download_mirror = config.get("mirror", "download-mirror")
logger.info(f"Selected alternative download mirror {download_mirror}")
except configparser.NoOptionError:
download_mirror = ""
logger.debug("No alternative download mirror found in config.")
if download_mirror:
try:
logger.debug(
"Checking config for only download from alternative download"
+ "mirror..."
)
download_mirror_no_fallback = config.getboolean(
"mirror", "download-mirror-no-fallback"
)
if download_mirror_no_fallback:
logger.info("Setting to download from mirror without fallback")
else:
logger.debug("Setting to fallback to original if download mirror fails")
except configparser.NoOptionError:
download_mirror_no_fallback = False
logger.debug("No download mirror fallback setting found in config.")
else:
download_mirror_no_fallback = False
logger.debug(
"Skip checking download-mirror-no-fallback because dependent option"
+ "is not set in config."
)
try:
simple_format = get_format_value(config.get("mirror", "simple-format"))
except configparser.NoOptionError:
logger.debug("Storing all Simple Formats by default ...")
simple_format = SimpleFormat.ALL
return SetConfigValues(
json_save,
root_uri,
diff_file_path,
diff_append_epoch,
digest_name,
storage_backend_name,
cleanup,
release_files_save,
compare_method,
download_mirror,
download_mirror_no_fallback,
simple_format,
)