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:DataQueryApi class to perform API operations.

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

.. |query_api_reference| raw:: html

Query API Reference # noqa E501
"""

from typing import Iterable, Optional, cast

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 DataQueryApi(BaseApi):
"""
This class provides access to HERE platform Data Query APIs.

The query service provides a way to get information (metadata) about layers and partitions
stored in a catalog. This service exposes the metadata for single partitions that users can
query one by one or by specifying a parent tile.
"""

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

:param base_url: base url
: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(DataQueryApi, self).init(
platform_config=platform_config,
application_config=application_config,
auth=auth,
proxies=proxies,
)
self.base_url = base_url

[docs]
def get_changes_by_id(
self,
layer_id: str,
partition: list,
start_version: Optional[int] = None,
end_version: Optional[int] = None,
since_time: Optional[int] = None,
additional_fields: Optional[list] = ["dataSize", "checksum", "compressedDataSize", "crc"],
billing_tag: Optional[str] = None,
) -> dict:
"""
Get changes for the version or time range for the specific partition(s).

:param layer_id: Unique layer id. Content of this parameter refers to a valid layer ID.
:param partition: The partitions you want to include in the response. This allows you to
limit the response to specific partitions. You can specify multiple partitions by
using this parameter multiple times. The maximum number of partitions per call is 100.
:param start_version: Available/Required for versioned layers only; the beginning of the
range of versions you want to get (exclusive). By convention -1 indicates the initial
version before the first publication. After the first publication,the catalog version
is 0
:param end_version: Available/Required for versioned layers only; the end of the range of
versions you want to get (inclusive). This must be a valid catalog version greater
than the startVersion.
:param since_time: Available/Required for volatile layers only; will return partitions
whose data has been modified since this time, in milliseconds since epoch, inclusive.
:param additional_fields: Available values : dataSize, checksum, compressedDataSize, crc
:param billing_tag: A string which is used for grouping billing records.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/changes"
url = self.format_url(self.base_url, path)
additional_fields_str = (
",".join(cast(Iterable[str], additional_fields))
if type(additional_fields) == list
else ""
)
params = {"startVersion": start_version, "endVersion": end_version, "sinceTime": since_time, "additionalFields": additional_fields_str, "partition": partition, "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 get_partitions_by_id(
self,
layer_id: str,
partition: list,
version: Optional[int] = None,
additional_fields: Optional[list] = ["dataSize", "checksum", "compressedDataSize", "crc"],
billing_tag: Optional[str] = None,
) -> dict:
"""
Get metadata for specific partition ids. If the layer specified in the request does not
exist, the request results in an error. If a partition specified in the request does not
exist, the response does not include this partition. Maximum allowed number of partitions
ids per call is 100.

:param layer_id: The ID of the layer specified in the request. The content of this
parameter must refer to a valid layer already configured in the catalog configuration.
You can specify multiple partitions by using this parameter multiple times.
:param partition: Partition ids to use for filtering. You can specify multiple partitions
by using this parameter multiple times. Maximum allowed partitions ids per call is 100
:param version: The version of the catalog against which to run the query. Must be a valid
catalog version.
:param additional_fields: Additional fields - dataSize, checksum, compressedDataSize, crc.
:param billing_tag: A string which is used for grouping billing records.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/partitions"
url = self.format_url(self.base_url, path)
additional_fields_str = (
",".join(cast(Iterable[str], additional_fields))
if type(additional_fields) == list
else ""
)
params = {"partition": partition, "version": version, "additionalFields": additional_fields_str, "billingTag": billing_tag,}
resp = self.get(url, params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def quad_tree_index(
self,
layer_id: str,
version: int,
quad_key: str,
depth: int,
additional_fields: Optional[list] = ["dataSize", "checksum", "compressedDataSize", "crc"],
billing_tag: Optional[str] = None,
) -> dict:
"""
Get metadata for the requested index. Only available for versioned layers where the
partitioning scheme is heretile.

:param layer_id: The ID of the layer specified in the request. Content of this parameter
must refer to a valid layer already configured in the catalog configuration. Exactly
one layer ID must be provided.
:param version: The version of the catalog against which to run the query. Must be a valid
catalog version.
:param quad_key: The geometric area specified by an index in the request, represented as a
HERE tile.
:param depth: The recursion depth of the response. If set to 0, the response includes only
data for the quadKey specified in the request. In this way, depth describes the
maximum length of the subQuadKeys in the response. The maximum allowed value for the
depth parameter is 4.
:param additional_fields: Additional fields - dataSize, checksum, compressedDataSize, crc.
:param billing_tag: A string which is used for grouping billing records.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/versions/{version}/quadkeys/{quad_key}/depths/{depth}"
url = self.format_url(self.base_url, path)
additional_fields_str = (
",".join(cast(Iterable[str], additional_fields))
if type(additional_fields) == list
else ""
)
params = {"additionalFields": additional_fields_str, "billingTag": billing_tag}
resp = self.get(url, params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def quad_tree_index_volatile(
self,
layer_id: str,
quad_key: str,
depth: int,
additional_fields: Optional[list] = ["dataSize", "checksum", "compressedDataSize", "crc"],
billing_tag: Optional[str] = None,
) -> dict:
"""
Get metadata for the requested index. Only available for volatile layers where the
partitioning scheme is heretile.

:param layer_id: The ID of the layer specified in the request. Content of this parameter
must refer to a valid layer already configured in the catalog configuration. Exactly
one layer ID must be provided.
:param quad_key: The geometric area specified by an index in the request, represented as a
HERE tile.
:param depth: The recursion depth of the response. If set to 0, the response includes only
data for the quadKey specified in the request. In this way, depth describes the
maximum length of the subQuadKeys in the response. The maximum allowed value for the
depth parameter is 4.
:param additional_fields: Additional fields - dataSize, checksum, compressedDataSize, crc.
:param billing_tag: A string which is used for grouping billing records.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/quadkeys/{quad_key}/depths/{depth}"
url = self.format_url(self.base_url, path)
additional_fields_str = (
",".join(cast(Iterable[str], additional_fields))
if type(additional_fields) == list
else ""
)
params = {"additionalFields": additional_fields_str, "billingTag": billing_tag}
resp = self.get(url, params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)