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 contains a :class:DataConfigApi class to perform API operations.

The HERE API reference documentation used in this module can be found here:
|config_api_reference|

.. |config_api_reference| raw:: html

Config API Reference # noqa E501
"""

from typing import Any, Dict, Optional, Tuple
from urllib.parse import urlparse

from here.platform.api.base_api import BaseApi
from here.platform.auth import Auth
from here.platform.config import ApplicationConfig, PlatformConfig
from here.platform.exceptions import PlatformException

[docs]
class DataConfigApi(BaseApi):
"""
This class provides access to HERE platform Data Config APIs.

The config service provides basic catalog management operations. It manages all platform
resources needed for different kinds of catalogs and operations on them.
"""

def init(
self,
platform_config: PlatformConfig,
application_config: ApplicationConfig,
auth: Optional[Auth],
proxies: Optional[dict] = None,
):
"""
Instantiate DataConfigApi object.

:param platform_config: a mandatory :class:PlatformConfig object to provide
configuration information for the API.
:param application_config: a mandatory :class:ApplicationConfig object to provide
configuration information for the API.
:param auth: an Authentication instance.
:param proxies: an optional proxy configuration. Defaults to the environment proxy
configuration.
"""
super(DataConfigApi, self).init(
platform_config=platform_config,
application_config=application_config,
auth=auth,
proxies=proxies,
)

temporary work-around for LDS support: If api url has scheme, use value unchanged

as scheme and netloc.

parsed = urlparse(self.platform_config.api)
if parsed.scheme:
self.base_url = f"{self.platform_config.api}/config/v1"
else:
self.base_url = (
f"https://config.data.{self.platform_config.api}/config/v1" # noqa: E231
)

[docs]
def get_catalogs(self, **filters: str) -> dict:
"""
List (or search) catalogs and layers.

:param filters: a dictionary with search criteria as key/value pairs
:return: a dictionary with all catalog information found
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/catalogs"
url = self.format_url(self.base_url, path)

params = {"billingTag": billing_tag}

resp = self.get(url, params=filters)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def create_catalog(self, data: Dict[str, Any], billing_tag: Optional[str] = None) -> dict:
"""
Create a catalog.

:param data: a dict with a catalog metadata.
:param billing_tag: A string which is used for grouping billing records.
:return: a dict with the catalog creation status.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/catalogs"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}
resp = self.post(url, data, params)
if resp.status_code in [200, 202]:

Note: 200 is returned only by LDS

resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_catalog_details(self, catalog_hrn: str, billing_tag: Optional[str] = None) -> dict:
"""
Get the full catalog configuration for the requested catalog.

:param catalog_hrn: a HERE Resource Name
:param billing_tag: A string which is used for grouping billing records.
:return: a dictionary with catalog details
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/catalogs/{catalog_hrn}"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def catalog_exists(self, catalog_hrn: str, billing_tag: Optional[str] = None) -> bool:
"""
Check whether a catalog with the specified HRN exists.

:param catalog_hrn: a HERE Resource Name
:param billing_tag: A string which is used for grouping billing records.
:return: Boolean value based on Http response
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/catalogs/{catalog_hrn}"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}
resp = self.head(url, params)
if resp.status_code == 200:
return True
elif resp.status_code == 404:
return False
else:
raise PlatformException(resp)

[docs]
def update_catalog(
self, catalog_hrn: str, data: Dict[str, Any], billing_tag: Optional[str] = None
) -> dict:
"""
Update a catalog.

:param catalog_hrn: a HERE Resource Name.
:param data: body of the update catalog request.
:param billing_tag: A string which is used for grouping billing records.
:return: a dict with catalog update status.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/catalogs/{catalog_hrn}"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}
resp = self.put(url=url, data=data, params=params)
if resp.status_code == 202:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def patch_catalog(self, catalog_hrn: str, data: Dict[str, Any]) -> dict:
"""
Modify or patch a catalog.

:param catalog_hrn: a HERE Resource Name.
:param data: body of the modify catalog request.
:return: a dict with catalog update status.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/catalogs/{catalog_hrn}"
url = self.format_url(self.base_url, path)
resp = self.patch(url=url, data=data)
if resp.status_code == 202:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def delete_catalog(self, catalog_hrn: str, billing_tag: Optional[str] = None) -> dict:
"""
Delete a catalog.

:param catalog_hrn: a HERE Resource Name.
:param billing_tag: a string which is used for grouping billing records.
:return: a dict with catalog deletion status.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/catalogs/{catalog_hrn}"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}
resp = self.delete(url, params)
if resp.status_code in [200, 202]:

Note: 200 is returned only by LDS

resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def patch_layer(self, catalog_hrn: str, layer_id: str, data: Dict[str, Any]) -> dict:
"""
Modify or patch a catalog layer.

:param catalog_hrn: a HERE Resource Name.
:param layer_id: a string which represents layer id.
:param data: body of the update catalog request.
:return: a dict with catalog layer update status.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/catalogs/{catalog_hrn}/layers/{layer_id}"
url = self.format_url(self.base_url, path)
resp = self.patch(url=url, data=data)
if resp.status_code == 202:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def delete_layer(self, catalog_hrn: str, layer_id: str) -> dict:
"""
Delete a catalog layer.

:param catalog_hrn: a HERE Resource Name.
:param layer_id: a string which represents layer id.
:return: a dict with catalog layer deletion status.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/catalogs/{catalog_hrn}/layers/{layer_id}"
url = self.format_url(self.base_url, path)
resp = self.delete(url)
if resp.status_code == 202:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def disable_automatic_version_deletion(self, catalog_hrn: str) -> dict:
"""
Disable automatic retired versions cleanup.

:param catalog_hrn: a HERE Resource Name.
:return: a dict with catalog automatic version deletion status.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/catalogs/{catalog_hrn}/automaticVersionDeletion"
url = self.format_url(self.base_url, path)
resp = self.delete(url)
if resp.status_code == 202:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_catalog_status(
self, catalog_status_href: str, billing_tag: Optional[str] = None
) -> Tuple[dict, bool]:
"""
Get the status of the catalog operations for the given token.

:param catalog_status_href: a catalog status href url.
:param billing_tag: A string which is used for grouping billing records.
:return: A tuple with dict with the status of the catalog/layer operation and bool set to
True if the operation completed or False if the operation is still in progress.
:raises PlatformException: If platform responds with an HTTP error.
"""
params = {"billingTag": billing_tag}
resp = self.get(url=catalog_status_href, params=params)
if resp.status_code in [200, 201, 202, 303]:

Note: 201 is returned only by LDS

resp_json: dict = resp.json()
return resp_json, resp.status_code != 202
else:
raise PlatformException(resp)

[docs]
def check_subscription_exists(self, subscription_hrn: str) -> bool:
"""
Checks whether a subscription with the specified HRN exists.

:param subscription_hrn: The HERE Resource Name (HRN) of subscription
:return: True if subscription exists.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/subscriptions/{subscription_hrn}"
url = self.format_url(self.base_url, path)
resp = self.head(url)
if resp.status_code == 200:
return True
elif resp.status_code == 404:
return False
else:
raise PlatformException(resp)

[docs]
def list_subscriptions(self, **filters: dict) -> dict:
"""
Lists all subscriptions that your account has access to.

:param filters: Http query params dict.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/subscriptions"
url = self.format_url(self.base_url, path)
resp = self.get(url, params=filters)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def create_subscription(self, configuration_object: dict) -> dict:
"""
Creates a subscription between source catalog/layer and target catalog/layer

:param configuration_object: A subscription configuration object.

:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/subscriptions"
url = self.format_url(self.base_url, path)
resp = self.post(url, data=configuration_object)
if resp.status_code == 202:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_subscription(self, subscription_hrn: str) -> dict:
"""
Returns configuration of the subscription associated with the HRN.

:param subscription_hrn: The HERE Resource Name (HRN) of subscription

:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/subscriptions/{subscription_hrn}"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def delete_subscription(self, subscription_hrn: str) -> dict:
"""
Deletes a subscription associated with the HRN.

:param subscription_hrn: The HERE Resource Name (HRN) of subscription.

:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/subscriptions/{subscription_hrn}"
url = self.format_url(self.base_url, path)
resp = self.delete(url)
if resp.status_code == 202:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_subscription_status(self, status_token: str) -> dict:
"""
Get the status of the subscription.

:param status_token: Status token from create/delete subscription response.

:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/subscriptions/status/{status_token}"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)