Adapt Config to be a dataclass (#437)

* switch to dataclass implementation

* adapt rest of code

* fix tests

* fix init parameters and add tests for hierarchy

* remove redundant pytint comment

* Update sentinelhub/config.py

Co-authored-by: Matic Lubej <[email protected]>

* adjust precedence

---------

Co-authored-by: Matic Lubej <[email protected]>

Author: zigaLuksic

Diff for: sentinelhub/commands.py

@@ -31,7 +31,7 @@ def main_help() -> None:
 
 def _config_options(func: FC) -> FC:
     """A helper function which joins click.option functions of each parameter from config.json"""
-    for param in SHConfig().get_params()[-1::-1]:
+    for param in list(SHConfig().to_dict())[-1::-1]:
         func = click.option(f"--{param}", param, help=f"Set new values to configuration parameter `{param}`")(func)
     return func
 
@@ -66,12 +66,9 @@ def config(show: bool, profile: str, **params: Any) -> None:
 
     sh_config.save(profile=profile)
 
-    for param in sh_config.get_params():
-        value = getattr(sh_config, param)
+    for param, value in sh_config.to_dict(mask_credentials=False).items():
         if value != getattr(old_config, param):
-            if isinstance(value, str):
-                value = f"'{value}'"
-            click.echo(f"The value of parameter `{param}` was updated to {value}")
+            click.echo(f"The value of parameter `{param}` was updated to {repr(value)}")
 
     if show:
         unmasked_str_repr = json.dumps(sh_config.to_dict(mask_credentials=False), indent=2)

Diff for: sentinelhub/config.py

@@ -5,10 +5,10 @@
 
 import copy
 import json
-import numbers
 import os
+from dataclasses import asdict, dataclass
 from pathlib import Path
-from typing import Dict, Tuple, Union
+from typing import Any, Dict, Optional, Union
 
 import tomli
 import tomli_w
@@ -19,7 +19,37 @@
 SH_CLIENT_SECRET_ENV_VAR = "SH_CLIENT_SECRET"
 
 
-class SHConfig:  # pylint: disable=too-many-instance-attributes
+@dataclass(repr=False)
+class _SHConfig:
+    instance_id: str = ""
+    sh_client_id: str = ""
+    sh_client_secret: str = ""
+    sh_base_url: str = "https://services.sentinel-hub.com"
+    sh_auth_base_url: str = "https://services.sentinel-hub.com"
+    geopedia_wms_url: str = "https://service.geopedia.world"
+    geopedia_rest_url: str = "https://www.geopedia.world/rest"
+    aws_access_key_id: str = ""
+    aws_secret_access_key: str = ""
+    aws_session_token: str = ""
+    aws_metadata_url: str = "https://roda.sentinel-hub.com"
+    aws_s3_l1c_bucket: str = "sentinel-s2-l1c"
+    aws_s3_l2a_bucket: str = "sentinel-s2-l2a"
+    opensearch_url: str = "http://opensearch.sentinel-hub.com/resto/api/collections/Sentinel2"
+    max_wfs_records_per_query: int = 100
+    max_opensearch_records_per_query: int = 500  # pylint: disable=invalid-name
+    max_download_attempts: int = 4
+    download_sleep_time: float = 5.0
+    download_timeout_seconds: float = 120.0
+    number_of_download_processes: int = 1
+
+    def __post_init__(self) -> None:
+        if self.max_wfs_records_per_query > 100:
+            raise ValueError("Value of config parameter `max_wfs_records_per_query` must be at most 100")
+        if self.max_opensearch_records_per_query > 500:
+            raise ValueError("Value of config parameter `max_opensearch_records_per_query` must be at most 500")
+
+
+class SHConfig(_SHConfig):
     """A sentinelhub-py package configuration class.
 
     The class reads the configurable settings from ``config.toml`` file on initialization:
@@ -51,14 +81,7 @@ class SHConfig:  # pylint: disable=too-many-instance-attributes
         - `download_timeout_seconds`: Maximum number of seconds before download attempt is canceled.
         - `number_of_download_processes`: Number of download processes, used to calculate rate-limit sleep time.
 
-    For manual modification of `config.toml` you can see the expected location of the file via
-    `SHConfig.get_config_location()`.
-
-    Usage in the code:
-
-        * ``SHConfig().sh_base_url``
-        * ``SHConfig().instance_id``
-
+    The location of `config.toml` for manual modification can be found with `SHConfig.get_config_location()`.
     """
 
     CREDENTIALS = (
@@ -69,132 +92,64 @@ class SHConfig:  # pylint: disable=too-many-instance-attributes
         "aws_secret_access_key",
         "aws_session_token",
     )
-    OTHER_PARAMS = (
-        "sh_base_url",
-        "sh_auth_base_url",
-        "geopedia_wms_url",
-        "geopedia_rest_url",
-        "aws_metadata_url",
-        "aws_s3_l1c_bucket",
-        "aws_s3_l2a_bucket",
-        "opensearch_url",
-        "max_wfs_records_per_query",
-        "max_opensearch_records_per_query",
-        "max_download_attempts",
-        "download_sleep_time",
-        "download_timeout_seconds",
-        "number_of_download_processes",
-    )
 
-    def __init__(self, profile: str = DEFAULT_PROFILE, *, use_defaults: bool = False):
+    def __init__(self, profile: Optional[str] = None, *, use_defaults: bool = False, **kwargs: Any):
         """
-        :param profile: Specifies which profile to load form the configuration file. The environment variable
-            SH_USER_PROFILE has precedence.
+        :param profile: Specifies which profile to load from the configuration file. Has precedence over the environment
+            variable `SH_USER_PROFILE`.
         :param use_defaults: Does not load the configuration file, returns config object with defaults only.
+        :param kwargs: Any fields of `SHConfig` to be updated. Overrides settings from `config.toml` and environment.
         """
-
-        self.instance_id: str = ""
-        self.sh_client_id: str = ""
-        self.sh_client_secret: str = ""
-        self.sh_base_url: str = "https://services.sentinel-hub.com"
-        self.sh_auth_base_url: str = "https://services.sentinel-hub.com"
-        self.geopedia_wms_url: str = "https://service.geopedia.world"
-        self.geopedia_rest_url: str = "https://www.geopedia.world/rest"
-        self.aws_access_key_id: str = ""
-        self.aws_secret_access_key: str = ""
-        self.aws_session_token: str = ""
-        self.aws_metadata_url: str = "https://roda.sentinel-hub.com"
-        self.aws_s3_l1c_bucket: str = "sentinel-s2-l1c"
-        self.aws_s3_l2a_bucket: str = "sentinel-s2-l2a"
-        self.opensearch_url: str = "http://opensearch.sentinel-hub.com/resto/api/collections/Sentinel2"
-        self.max_wfs_records_per_query: int = 100
-        self.max_opensearch_records_per_query: int = 500  # pylint: disable=invalid-name
-        self.max_download_attempts: int = 4
-        self.download_sleep_time: float = 5.0
-        self.download_timeout_seconds: float = 120.0
-        self.number_of_download_processes: int = 1
-
-        profile = os.environ.get(SH_PROFILE_ENV_VAR, default=profile)
+        if profile is None:
+            profile = os.environ.get(SH_PROFILE_ENV_VAR, default=DEFAULT_PROFILE)
 
         if not use_defaults:
+            env_kwargs = {
+                "sh_client_id": os.environ.get(SH_CLIENT_ID_ENV_VAR),
+                "sh_client_secret": os.environ.get(SH_CLIENT_SECRET_ENV_VAR),
+            }
+            env_kwargs = {k: v for k, v in env_kwargs.items() if v is not None}
+
             # load from config.toml
-            loaded_instance = SHConfig.load(profile=profile)  # user parameters validated in here already
-            for param in SHConfig.get_params():
-                setattr(self, param, getattr(loaded_instance, param))
-
-            # check env
-            self.sh_client_id = os.environ.get(SH_CLIENT_ID_ENV_VAR, default=self.sh_client_id)
-            self.sh_client_secret = os.environ.get(SH_CLIENT_SECRET_ENV_VAR, default=self.sh_client_secret)
-
-    def _validate_values(self) -> None:
-        """Ensures that the values are aligned with expectations."""
-        default = SHConfig(use_defaults=True)
-
-        for param in self.get_params():
-            value = getattr(self, param)
-            default_value = getattr(default, param)
-            param_type = type(default_value)
-
-            if isinstance(value, str) and value.startswith("http"):
-                value = value.rstrip("/")
-            if (param_type is float) and isinstance(value, numbers.Number):
-                continue
-            if not isinstance(value, param_type):
-                raise ValueError(f"Value of parameter `{param}` must be of type {param_type.__name__}")
-        if self.max_wfs_records_per_query > 100:
-            raise ValueError("Value of config parameter `max_wfs_records_per_query` must be at most 100")
-        if self.max_opensearch_records_per_query > 500:
-            raise ValueError("Value of config parameter `max_opensearch_records_per_query` must be at most 500")
+            loaded_kwargs = SHConfig.load(profile=profile).to_dict(mask_credentials=False)
+
+            kwargs = {**loaded_kwargs, **env_kwargs, **kwargs}  # precedence: init params > env > loaded
+
+        super().__init__(**kwargs)
 
     def __str__(self) -> str:
-        """Content of SHConfig in json schema. Credentials are masked for safety."""
+        """Content of `SHConfig` in json schema. Credentials are masked for safety."""
         return json.dumps(self.to_dict(mask_credentials=True), indent=2)
 
     def __repr__(self) -> str:
-        """Representation of SHConfig parameters. Credentials are masked for safety."""
+        """Representation of `SHConfig`. Credentials are masked for safety."""
         config_dict = self.to_dict(mask_credentials=True)
         content = ",\n  ".join(f"{key}={repr(value)}" for key, value in config_dict.items())
         return f"{self.__class__.__name__}(\n  {content},\n)"
 
-    def __eq__(self, other: object) -> bool:
-        """Two instances of `SHConfig` are equal if all values of their parameters are equal."""
-        if not isinstance(other, SHConfig):
-            return False
-        return all(getattr(self, param) == getattr(other, param) for param in self.get_params())
-
     @classmethod
     def load(cls, profile: str = DEFAULT_PROFILE) -> SHConfig:
         """Loads configuration parameters from the config file at `SHConfig.get_config_location()`.
 
         :param profile: Which profile to load from the configuration file.
         """
-        config = cls(use_defaults=True)
-
         filename = cls.get_config_location()
         if not os.path.exists(filename):
-            config.save(profile)  # store default configuration to standard location
+            cls(use_defaults=True).save(profile)  # store default configuration to standard location
 
         with open(filename, "rb") as cfg_file:
             configurations_dict = tomli.load(cfg_file)
 
         if profile not in configurations_dict:
             raise KeyError(f"Profile {profile} not found in configuration file.")
 
-        config_fields = cls.get_params()
-        for param, value in configurations_dict[profile].items():
-            if param in config_fields:
-                setattr(config, param, value)
-
-        config._validate_values()
-        return config
+        return cls(use_defaults=True, **configurations_dict[profile])
 
     def save(self, profile: str = DEFAULT_PROFILE) -> None:
         """Saves configuration parameters to the config file at `SHConfig.get_config_location()`.
 
         :param profile: Under which profile to save the configuration.
         """
-        self._validate_values()
-
         file_path = Path(self.get_config_location())
         file_path.parent.mkdir(parents=True, exist_ok=True)
 
@@ -218,18 +173,13 @@ def copy(self) -> SHConfig:
         """Makes a copy of an instance of `SHConfig`"""
         return copy.copy(self)
 
-    @classmethod
-    def get_params(cls) -> Tuple[str, ...]:
-        """Returns a list of parameter names."""
-        return cls.CREDENTIALS + cls.OTHER_PARAMS
-
     def to_dict(self, mask_credentials: bool = True) -> Dict[str, Union[str, float]]:
         """Get a dictionary representation of the `SHConfig` class.
 
         :param hide_credentials: Wether to mask fields containing credentials.
         :return: A dictionary with configuration parameters
         """
-        config_params = {param: getattr(self, param) for param in self.get_params()}
+        config_params = asdict(self)
 
         if mask_credentials:
             for param in self.CREDENTIALS:

Diff for: tests/api/test_byoc.py

@@ -17,7 +17,7 @@
 @pytest.fixture(name="config")
 def config_fixture() -> SHConfig:
     config = SHConfig()
-    for param in config.get_params():
+    for param in config.to_dict():
         env_variable = param.upper()
         if os.environ.get(env_variable):
             setattr(config, param, os.environ.get(env_variable))

Diff for: tests/conftest.py

@@ -20,7 +20,7 @@
 
 def pytest_configure(config: Config) -> None:
     shconfig = SHConfig()
-    for param in shconfig.get_params():
+    for param in shconfig.to_dict():
         env_variable = param.upper()
         if os.environ.get(env_variable):
             setattr(shconfig, param, os.environ.get(env_variable))

Diff for: tests/test_config.py

@@ -71,10 +71,6 @@ def test_save(restore_config_file: None) -> None:
     config = SHConfig()
     old_value = config.download_timeout_seconds
 
-    config.download_timeout_seconds = "abcd"  # type: ignore[assignment]
-    with pytest.raises(ValueError):
-        config.save()
-
     new_value = 150.5
     config.download_timeout_seconds = new_value
 
@@ -102,10 +98,22 @@ def test_environment_variables(restore_config_file: None, monkeypatch) -> None:
     assert config.sh_client_secret == "bees-are-very-friendly"
 
 
+@pytest.mark.dependency(depends=["test_user_config_is_masked"])
+def test_initialization_with_params(monkeypatch) -> None:
+    loaded_config = SHConfig()
+    monkeypatch.setenv(SH_CLIENT_ID_ENV_VAR, "beekeeper")
+
+    config = SHConfig(sh_client_id="me", instance_id="beep", geopedia_wms_url="123")
+    assert config.instance_id == "beep"
+    assert config.geopedia_wms_url != loaded_config.geopedia_wms_url, "Should override settings from config.toml"
+    assert config.sh_client_id == "me", "Should override environment variable."
+
+
 @pytest.mark.dependency(depends=["test_user_config_is_masked"])
 def test_profiles(restore_config_file: None) -> None:
     config = SHConfig()
     config.instance_id = "beepbeep"
+    config.sh_client_id = "beepbeep"  # also some tests with a credentials field
     config.save(profile="beep")
 
     config.instance_id = "boopboop"
@@ -116,10 +124,10 @@ def test_profiles(restore_config_file: None) -> None:
     assert SHConfig.load(profile="boop").instance_id == "boopboop"
 
     # save an existing profile
-    beep_config.instance_id = "bap"
-    assert SHConfig(profile="beep").instance_id == "beepbeep"
+    beep_config.sh_client_id = "bap"
+    assert SHConfig(profile="beep").sh_client_id == "beepbeep"
     beep_config.save(profile="beep")
-    assert SHConfig(profile="beep").instance_id == "bap"
+    assert SHConfig(profile="beep").sh_client_id == "bap"
 
 
 @pytest.mark.dependency(depends=["test_user_config_is_masked"])
@@ -134,7 +142,7 @@ def test_profiles_from_env(restore_config_file: None, monkeypatch) -> None:
 
     monkeypatch.setenv(SH_PROFILE_ENV_VAR, "beekeeper")
     assert SHConfig().instance_id == "bee", "Environment profile is not used."
-    assert SHConfig(profile=DEFAULT_PROFILE).instance_id == "bee", "Environment should override explicit profile."
+    assert SHConfig(profile=DEFAULT_PROFILE).instance_id == "", "Explicit profile overrides environment."
 
 
 def test_loading_unknown_profile_fails() -> None:
@@ -175,8 +183,9 @@ def test_config_repr() -> None:
     assert config.instance_id not in config_repr, "Credentials are not masked properly."
     assert "*" * 16 + "a" * 4 in config_repr, "Credentials are not masked properly."
 
-    for param in SHConfig.OTHER_PARAMS:
-        assert f"{param}={repr(getattr(config, param))}" in config_repr
+    for param in config.to_dict():
+        if param not in SHConfig.CREDENTIALS:
+            assert f"{param}={repr(getattr(config, param))}" in config_repr
 
 
 @pytest.mark.dependency(depends=["test_user_config_is_masked"])
@@ -188,7 +197,6 @@ def test_transformation_to_dict(hide_credentials: bool) -> None:
 
     config_dict = config.to_dict(hide_credentials)
     assert isinstance(config_dict, dict)
-    assert tuple(config_dict) == config.get_params()
 
     if hide_credentials:
         assert config_dict["sh_client_secret"] == "*" * 11 + "x" * 4