Copyright (C) 2020-2022 HERE Global B.V. and its affiliate(s).

All rights reserved.

This software and other materials contain proprietary information

controlled by HERE and are protected by applicable copyright legislation.

Any use and utilization of this software and other materials and

disclosure to any third parties is conditional upon having a separate

agreement with HERE for the access, use, utilization or disclosure of this

software. In the absence of such agreement, the use of the software is not

allowed.

"""
This module provides a ApplicationConfig class to be used for configuration management.

A config object can be created from a application_config.conf file in the package, or from a
custom config file path provided.
"""
import os
from dataclasses import dataclass
from os.path import expanduser, expandvars
from pathlib import Path
from typing import Any, Dict

from here.platform.exceptions import ConfigException
from here.platform.utils.properties_helper import PropertiesDict

DEFAULT_APPLICATION_CONFIG_PATH = "~/.here/application_config.conf"

@dataclass
class _Constants:

field = (field_name, default_value)

retry_max_time = ("RETRY_MAX_TIME", 195)
retry_max_wait = ("RETRY_MAX_WAIT", 60)
connect_retry_max_time = ("CONNECT_RETRY_MAX_TIME", 10)
connect_timeout = ("CONNECT_TIMEOUT", 3.05)
read_timeout = ("READ_TIMEOUT", 60)
polling_wait = ("POLLING_WAIT", 5.0)
max_object_store_upload_part_size = (
"MAX_OBJECT_STORE_UPLOAD_PART_SIZE_MB",
96,
) # 96mb default maximum size for write_object

constants = Constants()

_ensureValue = (
lambda config, field_const: config[field_const[0]]
if field_const[0] in config
else field_const[1]
)

[docs]
class ApplicationConfig:
"""
ApplicationConfig provides functions for dealing with the pysdk application
Config.

Config can be read from the following locations:

• The default packaged file: application_config.conf
• The default location: "~/.here/application_config.conf"
• A custom path to a config file
  """

def init(
self,
retry_max_time: float,
read_timeout: float,
connect_retry_max_time: float,
connect_timeout: float,
polling_wait: float,
max_object_store_upload_part_size: int,
retry_max_wait: float,
):
"""
Instantiate the ApplicationConfig object.

:param retry_max_time: Maximum retry time.
:param read_timeout: Read timeout.
:param connect_retry_max_time: Maximum retry time to connect.
:param connect_timeout: Connection timeout.
:param polling_wait: Polling wait time
:param max_object_store_upload_part_size: Maximum object store upload part size in MB
:param retry_max_wait: Maximum time to wait for an individual retry.
"""
self.retry_max_time: float = retry_max_time
self.read_timeout: float = read_timeout
self.connect_retry_max_time: float = connect_retry_max_time
self.connect_timeout: float = connect_timeout
self.polling_wait: float = polling_wait
self.max_object_store_upload_part_size: int = max_object_store_upload_part_size
self.retry_max_wait: float = retry_max_wait
self.additional_parameters: Dict[str, Any] = dict()

[docs]
@classmethod
def from_default(cls) -> "ApplicationConfig":
"""Return the config read from the default config file path at
'~/.here/application_config.conf'.

If no default file is found, this method will try to read the
config from the package application_config.conf file.

:return: config
"""

application_conf = cls.from_config_file(DEFAULT_APPLICATION_CONFIG_PATH)

return application_conf

[docs]
@classmethod
def from_config_file(cls, path: str) -> "ApplicationConfig":
"""
Return the config object from a specified config file path.

:param path: path to a PySDK application_config.conf file.
:return: config
:raises ConfigException: Erroneous application_config.conf file in path
"""

try:
config path = expanduser(expandvars(path))
config = PropertiesDict.parse(config_path)
return ApplicationConfig(
retrymax_time=_ensureValue(config, constants.retry_max_time),
connectretry_max_time=_ensureValue(config, constants.connect_retry_max_time),
connecttimeout=_ensureValue(config, constants.connect_timeout),
readtimeout=_ensureValue(config, constants.read_timeout),
pollingwait=_ensureValue(config, constants.polling_wait),
max_object_store_upload_part_size=_ensureValue(
config,constants.max_object_store_upload_part_size
),
retrymax_wait=_ensureValue(config, constants.retry_max_wait),
)
except FileNotFoundError:
this_dir = Path(file).parent
default_path = this_dir / "application_config.conf"
if not default_path.exists():
raise ConfigException(f"Configuration file not found: {default_path}")
return cls.from_config_file(os.fspath(default_path))
except Exception as e:
raise ConfigException(f"Erroneous {config_path} file: {e}")