"""
Module for managing configuration data from `config.toml`
"""
from __future__ import annotations
import copy
import json
import os
import warnings
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
import tomli
import tomli_w
from .exceptions import SHDeprecationWarning
DEFAULT_PROFILE = "default-profile"
SH_PROFILE_ENV_VAR = "SH_PROFILE"
SH_CLIENT_ID_ENV_VAR = "SH_CLIENT_ID"
SH_CLIENT_SECRET_ENV_VAR = "SH_CLIENT_SECRET"
@dataclass(repr=False)
class _SHConfig: # pylint: disable=too-many-instance-attributes
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 | None = None
sh_token_url: str = "https://services.sentinel-hub.com/auth/realms/main/protocol/openid-connect/token"
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
max_retries: int | None = None
def __post_init__(self) -> None:
if self.sh_auth_base_url is not None:
self.sh_token_url = self.sh_auth_base_url + "/oauth/token"
warnings.warn(
"The parameter `sh_auth_base_url` of `SHConfig` has been replaced with `sh_token_url`. Please"
" update your configuration, for now the parameters were automatically adjusted to `sh_token_url ="
" sh_auth_base_url + '/oauth/token'`.",
category=SHDeprecationWarning,
)
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")
[docs]class SHConfig(_SHConfig):
"""A sentinelhub-py package configuration class.
The class reads the configurable settings from ``config.toml`` file on initialization:
- `instance_id`: An instance ID for Sentinel Hub service used for OGC requests.
- `sh_client_id`: User's OAuth client ID for Sentinel Hub service. Can be set via SH_CLIENT_ID environment
variable. The environment variable has precedence.
- `sh_client_secret`: User's OAuth client secret for Sentinel Hub service. Can be set via SH_CLIENT_SECRET
environment variable. The environment variable has precedence.
- `sh_base_url`: There exist multiple deployed instances of Sentinel Hub service, this parameter defines the
location of a specific service instance.
- `sh_token_url`: Url for Sentinel Hub Authentication service. Authentication is typically sent to the main
service deployment even if `sh_base_url` points to another deployment.
- `geopedia_wms_url`: Base url for Geopedia WMS services.
- `geopedia_rest_url`: Base url for Geopedia REST services.
- `aws_access_key_id`: Access key for AWS Requester Pays buckets.
- `aws_secret_access_key`: Secret access key for AWS Requester Pays buckets.
- `aws_session_token`: A session token for your AWS account. It is only needed when you are using temporary
credentials.
- `aws_metadata_url`: Base url for publicly available metadata files
- `aws_s3_l1c_bucket`: Name of Sentinel-2 L1C bucket at AWS s3 service.
- `aws_s3_l2a_bucket`: Name of Sentinel-2 L2A bucket at AWS s3 service.
- `opensearch_url`: Base url for Sentinelhub Opensearch service.
- `max_wfs_records_per_query`: Maximum number of records returned for each WFS query.
- `max_opensearch_records_per_query`: Maximum number of records returned for each Opensearch query.
- `max_download_attempts`: Maximum number of download attempts from a single URL until an error will be raised.
- `download_sleep_time`: Number of seconds to sleep between the first failed attempt and the next. Every next
attempt this number exponentially increases with factor `3`.
- `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.
- `max_retries`: Maximum number of retries until an exception is raised.
The location of `config.toml` for manual modification can be found with `SHConfig.get_config_location()`.
"""
CREDENTIALS = (
"instance_id",
"sh_client_id",
"sh_client_secret",
"aws_access_key_id",
"aws_secret_access_key",
"aws_session_token",
)
def __init__(self, profile: str | None = None, *, use_defaults: bool = False, **kwargs: Any):
"""
: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.
"""
profile = self._get_profile(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_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."""
return json.dumps(self.to_dict(mask_credentials=True), indent=2)
def __repr__(self) -> str:
"""Representation of `SHConfig`. Credentials are masked for safety."""
config_dict = self.to_dict(mask_credentials=True)
content = ",\n ".join(f"{key}={value!r}" for key, value in config_dict.items())
return f"{self.__class__.__name__}(\n {content},\n)"
@staticmethod
def _get_profile(profile: str | None) -> str:
return profile if profile is not None else os.environ.get(SH_PROFILE_ENV_VAR, default=DEFAULT_PROFILE)
[docs] @classmethod
def load(cls, profile: str | None = None) -> SHConfig:
"""Loads configuration parameters from the config file at `SHConfig.get_config_location()`.
:param profile: Which profile to load from the configuration file.
"""
profile = cls._get_profile(profile)
filename = cls.get_config_location()
if not os.path.exists(filename):
cls(use_defaults=True).save() # 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.")
return cls(use_defaults=True, **configurations_dict[profile])
[docs] def save(self, profile: str | None = None) -> None:
"""Saves configuration parameters to the config file at `SHConfig.get_config_location()`.
:param profile: Under which profile to save the configuration.
"""
profile = self._get_profile(profile)
file_path = Path(self.get_config_location())
file_path.parent.mkdir(parents=True, exist_ok=True)
if file_path.exists():
with open(file_path, "rb") as cfg_file:
current_configuration = tomli.load(cfg_file)
else:
current_configuration = {}
current_configuration[profile] = self._get_dict_of_diffs_from_defaults()
with open(file_path, "wb") as cfg_file:
tomli_w.dump(current_configuration, cfg_file)
def _get_dict_of_diffs_from_defaults(self) -> dict[str, str | float]:
"""Returns a dictionary containing key: value pairs for parameters that have values different from defaults."""
current_profile_config = self.to_dict(mask_credentials=False)
default_values = SHConfig(use_defaults=True).to_dict(mask_credentials=False)
return {key: value for key, value in current_profile_config.items() if default_values[key] != value}
[docs] def copy(self) -> SHConfig:
"""Makes a copy of an instance of `SHConfig`"""
return copy.copy(self)
[docs] def to_dict(self, mask_credentials: bool = True) -> dict[str, str | float]:
"""Get a dictionary representation of the `SHConfig` class.
:param mask_credentials: Wether to mask fields containing credentials.
:return: A dictionary with configuration parameters
"""
config_params = asdict(self)
if mask_credentials:
for param in self.CREDENTIALS:
config_params[param] = self._mask_credentials(config_params[param])
return config_params
def _mask_credentials(self, value: str) -> str:
"""In case a parameter that holds credentials is given it will mask its value"""
hide_size = min(max(len(value) - 4, 10), len(value))
return "*" * hide_size + value[hide_size:]
[docs] @classmethod
def get_config_location(cls) -> str:
"""Returns the default location of the user configuration file on disk."""
user_folder = os.path.expanduser("~")
return os.path.join(user_folder, ".config", "sentinelhub", "config.toml")