Add alternative config and mirror options classes

.pre-commit-config.yaml

@@ -31,7 +31,7 @@ repos:
     hooks:
       - id: mypy
         exclude: (docs/.*)
-        additional_dependencies: ["types-filelock", "types-freezegun", "types-pkg_resources"]
+        additional_dependencies: ["attrs", "types-filelock", "types-freezegun", "types-pkg_resources"]
 
   - repo: https://github.com/PyCQA/flake8
     rev: 7.0.0

src/bandersnatch/config/__init__.py

@@ -0,0 +1,4 @@
+# flake8: noqa
+from .core import BandersnatchConfig
+from .errors import ConfigurationError, InvalidValueError, MissingOptionError
+from .mirror_options import MirrorOptions

src/bandersnatch/config/attrs_utils.py

@@ -0,0 +1,72 @@
+from collections.abc import Callable
+from configparser import ConfigParser
+from typing import Any, TypeAlias, TypeVar
+
+import attrs
+from attrs import Attribute
+
+from . import errors
+
+AttrsConverter: TypeAlias = Callable[[Any], Any]
+AttrsFieldTransformer: TypeAlias = Callable[[type, list[Attribute]], list[Attribute]]
+
+_V = TypeVar("_V")
+AttrsValidator: TypeAlias = Callable[[Any, Attribute, _V], _V]
+
+
+def only_if_str(converter_fn: AttrsConverter) -> AttrsConverter:
+    """Wrap an attrs converter function so it is only applied to strings.
+
+    'converter' functions on attrs fields are applied to all values set to the field,
+    *including* the default value. This causes problems if the default value is already
+    the desired type but the converter only handles strings.
+
+    :param AttrsConverter converter_fn: any attrs converter
+    :return AttrsConverter: converter function that uses `converter_fn` if the passed
+        value is a string, and otherwise returns the value unmodified.
+    """
+
+    def _apply_if_str(value: Any) -> Any:
+        if isinstance(value, str):
+            return converter_fn(value)
+        else:
+            return value
+
+    return _apply_if_str
+
+
+def get_name_value_for_option(
+    config: ConfigParser, section_name: str, option: Attribute
+) -> tuple[str, object | None]:
+    option_name = config.optionxform(option.alias or option.name)
+
+    if option.default is attrs.NOTHING and not config.has_option(
+        section_name, option_name
+    ):
+        raise errors.MissingOptionError.for_option(section_name, option_name)
+
+    getter: Callable[..., Any]
+    if option.converter is not None:
+        getter = config.get
+    elif option.type == bool:
+        getter = config.getboolean
+    elif option.type == float:
+        getter = config.getfloat
+    elif option.type == int:
+        getter = config.getint
+    else:
+        getter = config.get
+
+    try:
+        option_value = getter(section_name, option_name, fallback=None)
+    except ValueError as conversion_error:
+        type_name = option.type.__name__ if option.type else "???"
+        message = f"can't convert option name '{option_name}' to expected type '{type_name}': {conversion_error!s}"
+        raise errors.InvalidValueError.for_option(
+            section_name, option_name, message
+        ) from conversion_error
+
+    return option_name, option_value
+
+
+validate_not_empty: AttrsValidator = attrs.validators.min_len(1)

src/bandersnatch/config/comparison_method.py

@@ -0,0 +1,30 @@
+"""Enumeration of supported file comparison strategies"""
+
+import sys
+
+if sys.version_info >= (3, 11):
+    from enum import StrEnum
+else:
+    from bandersnatch.utils import StrEnum
+
+
+class ComparisonMethod(StrEnum):
+    HASH = "hash"
+    STAT = "stat"
+
+
+class InvalidComparisonMethod(ValueError):
+    """We don't have a valid comparison method choice from configuration"""
+
+    pass
+
+
+def get_comparison_value(method: str) -> ComparisonMethod:
+    try:
+        return ComparisonMethod(method)
+    except ValueError:
+        valid_methods = sorted(v.value for v in ComparisonMethod)
+        raise InvalidComparisonMethod(
+            f"{method} is not a valid file comparison method. "
+            + f"Valid options are: {valid_methods}"
+        )

src/bandersnatch/config/core.py

@@ -0,0 +1,42 @@
+from collections.abc import Mapping
+from configparser import ConfigParser, ExtendedInterpolation
+from pathlib import Path
+from typing import Protocol, TypeVar, cast
+
+from typing_extensions import Self
+
+
+class ConfigModel(Protocol):
+
+    @classmethod
+    def from_config_parser(cls, source: ConfigParser) -> Self: ...
+
+
+_C = TypeVar("_C", bound=ConfigModel)
+
+
+class BandersnatchConfig(ConfigParser):
+
+    def __init__(self, defaults: Mapping[str, str] | None = None) -> None:
+        super().__init__(
+            defaults=defaults,
+            delimiters=("=",),
+            strict=True,
+            interpolation=ExtendedInterpolation(),
+        )
+
+        self._validate_config_models: dict[str, ConfigModel] = {}
+
+    # This allows writing option names in the config file with either '_' or '-' as word separators
+    def optionxform(self, optionstr: str) -> str:
+        return optionstr.lower().replace("-", "_")
+
+    def read_path(self, file_path: Path) -> None:
+        with file_path.open() as cfg_file:
+            self.read_file(cfg_file)
+
+    def get_validated(self, model: type[_C]) -> _C:
+        name = model.__qualname__
+        if name not in self._validate_config_models:
+            self._validate_config_models[name] = model.from_config_parser(self)
+        return cast(_C, self._validate_config_models[name])

src/bandersnatch/config/errors.py

@@ -0,0 +1,22 @@
+from typing_extensions import Self
+
+
+class ConfigurationError(Exception):
+
+    @classmethod
+    def for_section(cls, section: str, message: str) -> Self:
+        return cls(f"Configuration error in [{section}] section: {message}")
+
+
+class MissingOptionError(ConfigurationError):
+
+    @classmethod
+    def for_option(cls, section: str, option: str) -> Self:
+        return cls.for_section(section, f"missing required option '{option}'")
+
+
+class InvalidValueError(ConfigurationError):
+
+    @classmethod
+    def for_option(cls, section: str, option: str, info: str) -> Self:
+        return cls.for_section(section, f"invalid value for option '{option}': {info}")

src/bandersnatch/config/mirror_options.py

@@ -0,0 +1,156 @@
+from configparser import ConfigParser, NoOptionError, NoSectionError
+from logging import getLogger
+from pathlib import PurePath
+from typing import Any
+
+import attrs
+
+from bandersnatch.simple import (
+    SimpleDigest,
+    SimpleFormat,
+    get_digest_value,
+    get_format_value,
+)
+
+from .attrs_utils import get_name_value_for_option, only_if_str, validate_not_empty
+from .comparison_method import ComparisonMethod, get_comparison_value
+from .diff_file_reference import eval_legacy_config_ref, has_legacy_config_ref
+from .errors import ConfigurationError, InvalidValueError
+
+logger = getLogger("bandersnatch")
+
+_default_master_url = "https://pypi.org"
+_default_root_uri = "https://files.pythonhosted.org"
+
+
+@attrs.define(kw_only=True)
+class MirrorOptions:
+    """Class with attributes for all the options that may appear in the
+    '[mirror]' section of a config file.
+    """
+
+    directory: PurePath = attrs.field(converter=PurePath)
+
+    storage_backend_name: str = attrs.field(
+        default="filesystem",
+        alias="storage_backend",
+        validator=validate_not_empty,
+    )
+
+    master_url: str = attrs.field(default=_default_master_url, alias="master")
+    proxy_url: str | None = attrs.field(default=None, alias="proxy")
+
+    download_mirror_url: str | None = attrs.field(default=None, alias="download_mirror")
+    download_mirror_no_fallback: bool = False
+
+    save_release_files: bool = attrs.field(default=True, alias="release_files")
+    save_json: bool = attrs.field(default=False, alias="json")
+
+    # type-ignores on converters for the following enums b/c MyPy's plugin for attrs
+    # doesn't handle using arbitrary functions as converters
+    simple_format: SimpleFormat = attrs.field(
+        default=SimpleFormat.ALL,
+        converter=only_if_str(get_format_value),  # type: ignore
+    )
+
+    compare_method: ComparisonMethod = attrs.field(
+        default=ComparisonMethod.HASH,
+        converter=only_if_str(get_comparison_value),  # type: ignore
+    )
+
+    digest_name: SimpleDigest = attrs.field(
+        default=SimpleDigest.SHA256,
+        converter=only_if_str(get_digest_value),  # type: ignore
+    )
+
+    # this gets a non-empty default value in post-init if save_release_files is False
+    root_uri: str = ""
+
+    hash_index: bool = False
+
+    keep_index_versions: int = attrs.field(default=0, validator=attrs.validators.ge(0))
+
+    diff_file: PurePath | None = attrs.field(
+        default=None, converter=attrs.converters.optional(PurePath)
+    )
+    diff_append_epoch: bool = False
+
+    stop_on_error: bool = False
+    timeout: float = attrs.field(default=10.0, validator=attrs.validators.gt(0))
+    global_timeout: float = attrs.field(
+        default=1800.0, validator=attrs.validators.gt(0)
+    )
+
+    workers: int = attrs.field(
+        default=3, validator=[attrs.validators.gt(0), attrs.validators.le(10)]
+    )
+
+    verifiers: int = attrs.field(
+        default=3, validator=[attrs.validators.gt(0), attrs.validators.le(10)]
+    )
+
+    log_config: PurePath | None = attrs.field(
+        default=None, converter=attrs.converters.optional(PurePath)
+    )
+
+    cleanup: bool = attrs.field(default=False, metadata={"deprecated": True})
+
+    # Called after the attrs class is constructed; doing cross-field validation here
+    def __attrs_post_init__(self) -> None:
+        # set default for root_uri if release-files is disabled
+        if not self.save_release_files and not self.root_uri:
+            logger.warning(
+                (
+                    "Inconsistent config: 'root_uri' should be set when "
+                    "'release-files' is disabled. Please set 'root-uri' in the "
+                    "[mirror] section of your config file. Using default value '%s'"
+                ),
+                _default_root_uri,
+            )
+            self.root_uri = _default_root_uri
+
+    @classmethod
+    def from_config_parser(cls, source: ConfigParser) -> "MirrorOptions":
+        if "mirror" not in source:
+            raise ConfigurationError("Config file missing required section '[mirror]'")
+
+        model_kwargs: dict[str, Any] = {}
+
+        for option in attrs.fields(cls):
+            option_name, option_value = get_name_value_for_option(
+                source, "mirror", option
+            )
+
+            if option_name == "diff_file" and isinstance(option_value, str):
+                option_value = _check_legacy_reference(source, option_value)
+
+            if option_value is not None:
+                model_kwargs[option_name] = option_value
+
+        try:
+            instance = cls(**model_kwargs)
+        except ValueError as err:
+            raise InvalidValueError.for_section("mirror", str(err)) from err
+        except TypeError as err:
+            raise ConfigurationError.for_section("mirror", str(err)) from err
+
+        return instance
+
+
+def _check_legacy_reference(config: ConfigParser, value: str) -> str | None:
+    if not has_legacy_config_ref(value):
+        return value
+
+    logger.warning(
+        "Found section reference using '{{ }}' in 'diff-file' path. "
+        "Use ConfigParser's built-in extended interpolation instead, "
+        "for example '${mirror:directory}/new-files'"
+    )
+    try:
+        return eval_legacy_config_ref(config, value)
+    except (ValueError, NoSectionError, NoOptionError) as ref_err:
+        # NOTE: raise here would be a breaking change; previous impl. logged and
+        # fell back to a default. Create exception anyway for consistent error messages.
+        exc = InvalidValueError.for_option("mirror", "diff-file", str(ref_err))
+        logger.error(str(exc))
+        return None