Source code for sentinelhub.config

Module for managing configuration data from `config.json`
from __future__ import annotations

import copy
import json
import numbers
import os
from typing import Any, Dict, Iterable, List, Optional, Union

from .exceptions import deprecated_function

ConfigDict = Dict[str, Union[str, int, float]]

[docs]class SHConfig: # pylint: disable=too-many-instance-attributes """A sentinelhub-py package configuration class. The class reads during its first initialization the configurable settings from ``./config.json`` file: - `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 - `sh_client_secret`: User's OAuth client secret for Sentinel Hub service - `sh_base_url`: There exist multiple deployed instances of Sentinel Hub service, this parameter defines the location of a specific service instance. - `sh_auth_base_url`: Base 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. Usage in the code: * ``SHConfig().sh_base_url`` * ``SHConfig().instance_id`` """ CREDENTIALS = { "instance_id", "sh_client_id", "sh_client_secret", "aws_access_key_id", "aws_secret_access_key", "aws_session_token", } CONFIG_PARAMS = [ "instance_id", "sh_client_id", "sh_client_secret", "sh_base_url", "sh_auth_base_url", "geopedia_wms_url", "geopedia_rest_url", "aws_access_key_id", "aws_secret_access_key", "aws_session_token", "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", ] _cache: Optional[Dict[str, Any]] = None def __init__(self, hide_credentials: bool = False, use_defaults: bool = False): """ :param hide_credentials: If `True` then methods that provide the entire content of the config object will mask out all credentials. But credentials could still be accessed directly from config object attributes. The default is `False`. :param use_defaults: Does not load the configuration file, returns config object with defaults only. """ self.instance_id: str = "" self.sh_client_id: str = "" self.sh_client_secret: str = "" self.sh_base_url: str = "" self.sh_auth_base_url: str = "" self.geopedia_wms_url: str = "" self.geopedia_rest_url: str = "" self.aws_access_key_id: str = "" self.aws_secret_access_key: str = "" self.aws_session_token: str = "" self.aws_metadata_url: str = "" self.aws_s3_l1c_bucket: str = "sentinel-s2-l1c" self.aws_s3_l2a_bucket: str = "sentinel-s2-l2a" self.opensearch_url: str = "" 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 self._hide_credentials = hide_credentials if not use_defaults: for param, value in self._global_cache.items(): setattr(self, param, value) def _validate_values(self) -> None: """Ensures that the values are aligned with expectations.""" default = SHConfig(use_defaults=True) for param in self.CONFIG_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") def __getitem__(self, name: str) -> Union[str, int, float]: """Config parameters can also be accessed as items.""" if name in self.CONFIG_PARAMS: return getattr(self, name) raise KeyError(f"`{name}` is not a supported config parameter") def __str__(self) -> str: """Content of SHConfig in json schema. If `hide_credentials` is set to `True` then credentials will be masked. """ return json.dumps(self.get_config_dict(), indent=2) def __repr__(self) -> str: """Representation of SHConfig parameters. If `hide_credentials` is set to `True` then credentials will be masked. """ repr_list = [f"{self.__class__.__name__}("] for key, value in self.get_config_dict().items(): repr_list.append(f"{key}={repr(value)},") return "\n ".join(repr_list).strip(",") + "\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.CONFIG_PARAMS) @property def _global_cache(self) -> Dict[str, Any]: """Uses a class attribute to store a global instance of a class with config parameters.""" if SHConfig._cache is None: loaded_instance = SHConfig.load(self.get_config_location()) SHConfig._cache = {param: getattr(loaded_instance, param) for param in SHConfig.CONFIG_PARAMS} return SHConfig._cache
[docs] @classmethod def load(cls, filename: str) -> SHConfig: """Method that loads configuration parameters from a file. Does not affect global settings. :param filename: Path to file from which to read configuration. """ with open(filename, "r") as cfg_file: config_dict = json.load(cfg_file) config = cls(use_defaults=True) for param, value in config_dict.items(): if param in cls.CONFIG_PARAMS: setattr(config, param, value) config._validate_values() return config
[docs] def save(self, filename: Optional[str] = None) -> None: """Method that saves configuration parameter changes from instance of SHConfig class to global config class and to `config.json` file. :param filename: Optional name of file to which to save configuration. If not specified saves to global default. :Example: ``my_config = SHConfig()`` \n ``my_config.instance_id = '<new instance id>'`` \n ```` """ self._validate_values() is_changed = False for param in self.CONFIG_PARAMS: if getattr(self, param) != self._global_cache[param]: is_changed = True self._global_cache[param] = getattr(self, param) # pylint: disable=unsupported-assignment-operation if is_changed: config_dict = {param: getattr(self, param) for param in self.CONFIG_PARAMS} with open(filename or self.get_config_location(), "w") as cfg_file: json.dump(config_dict, cfg_file, indent=2)
[docs] def copy(self) -> SHConfig: """Makes a copy of an instance of `SHConfig`""" return copy.copy(self)
[docs] def reset(self, params: Union[str, Iterable[str], object] = ...) -> None: """Resets configuration class to initial values. Use `` method in order to save this change. :param params: Parameters which will be reset. Parameters can be specified with a list of names, e.g. ``['instance_id', 'aws_access_key_id', 'aws_secret_access_key']``, or as a single name, e.g. ``'sh_base_url'``. By default, all parameters will be reset and default value is ``Ellipsis``. """ default = SHConfig(use_defaults=True) if params is ...: params = self.get_params() if isinstance(params, str): self._reset_param(params, default) elif isinstance(params, Iterable): for param in params: self._reset_param(param, default) else: raise ValueError( f"Parameters must be specified in form of a list of strings or as a single string, instead got {params}" )
def _reset_param(self, param: str, default: SHConfig) -> None: """Resets a single parameter :param param: A configuration parameter """ if param not in self.get_params(): raise ValueError(f"Cannot reset unknown parameter `{param}`") setattr(self, param, getattr(default, param))
[docs] def get_params(self) -> List[str]: """Returns a list of parameter names :return: List of parameter names """ return list(self.CONFIG_PARAMS)
[docs] def get_config_dict(self) -> ConfigDict: """Get a dictionary representation of `SHConfig` class. If `hide_credentials` is set to `True` then credentials will be masked. :return: A dictionary with configuration parameters """ config_params = {param: getattr(self, param) for param in self.CONFIG_PARAMS} if self._hide_credentials: config_params = {param: self._mask_credentials(param, value) for param, value in config_params.items()} return config_params
[docs] @classmethod def get_config_location(cls) -> str: """Returns location of configuration file on disk :return: File path of `config.json` file """ config_file = os.path.join(os.path.dirname(__file__), "config.json") if not os.path.isfile(config_file): with open(config_file, "w") as cfg_file: default_dict = cls(use_defaults=True).get_config_dict() json.dump(default_dict, cfg_file, indent=2) return config_file
def _mask_credentials(self, param: str, value: object) -> object: """In case a parameter that holds credentials is given it will mask its value""" if not (param in self.CREDENTIALS and value): return value if not isinstance(value, str): raise ValueError(f"Parameter `{param}` should be a string but {value} found") hide_size = min(max(len(value) - 4, 10), len(value)) return "*" * hide_size + value[hide_size:]
[docs] @deprecated_function(message_suffix="Use `f'{config.sh_auth_base_url}/oauth/token'` instead.") def get_sh_oauth_url(self) -> str: """Provides URL for Sentinel Hub authentication endpoint :return: A URL endpoint """ return f"{self.sh_auth_base_url}/oauth/token"
[docs] @deprecated_function(message_suffix="Use `f'{config.sh_base_url}/api/v1/process'` instead.") def get_sh_process_api_url(self) -> str: """Provides URL for Sentinel Hub Process API endpoint :return: A URL endpoint """ return f"{self.sh_base_url}/api/v1/process"
[docs] @deprecated_function(message_suffix="Use `f'{config.sh_base_url}/ogc'` instead.") def get_sh_ogc_url(self) -> str: """Provides URL for Sentinel Hub OGC endpoint :return: A URL endpoint """ return f"{self.sh_base_url}/ogc"
[docs] @deprecated_function(message_suffix="Use `f'{config.sh_auth_base_url}/aux/ratelimit'` instead.") def get_sh_rate_limit_url(self) -> str: """Provides URL for Sentinel Hub rate limiting endpoint :return: A URL endpoint """ return f"{self.sh_auth_base_url}/aux/ratelimit"
[docs] def raise_for_missing_instance_id(self) -> None: """In case Sentinel Hub instance ID is missing it raises an informative error :raises: ValueError """ if not self.instance_id: raise ValueError( "Sentinel Hub instance ID is missing. " "Either provide it with SHConfig object or save it into config.json configuration file. " "Check for more info." )