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

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

.. |metadata_api_reference| raw:: html

Metadata API Reference # noqa E501
"""

from typing import Iterable, Optional, Union, cast

from here.platform.api.base_api import BaseApi
from here.platform.api.stream import stream_json_response
from here.platform.auth import Auth
from here.platform.config import ApplicationConfig, PlatformConfig
from here.platform.constants import DEFAULT_ITER_CHUNK_SIZE
from here.platform.exceptions import PlatformException
from json_stream.base import StreamingJSONObject

[docs]
class DataMetadataApi(BaseApi):
"""
This class provides access to HERE platform Metadata APIs.

The metadata service provides a way to get information (metadata) about layers and partitions
stored in a catalog. This service exposes the metadata for all the partitions or all the
changed partitions. For a catalog with versioned layers, you can retrieve metadata for a
particular version or version range of the catalog.
"""

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

:param base_url: a Metadata API Url obtained from Lookup API for a given hrn.
: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(DataMetadataApi, self).init(
platform_config=platform_config,
application_config=application_config,
auth=auth,
proxies=proxies,
)
self.base_url = base_url

[docs]
def get_layers_version(self, version: int, billing_tag: Optional[str] = None) -> dict:
"""
Return information about layer versions for the catalog version.

:param version: Catalog version.
:param billing_tag: Billing Tag is an optional free-form tag which is used for grouping
billing records together.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/layerVersions"
url = self.format_url(self.base_url, path)
params = {"version": version, "billingTag": billing_tag}
resp = self.get(url=url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_changes(
self,
layer_id: str,
start_version: Optional[int] = None,
end_version: Optional[int] = None,
since_time: Optional[int] = None,
part: Optional[str] = None,
additional_fields: Optional[list] = ["dataSize", "checksum", "compressedDataSize", "crc"],
range_header: Optional[str] = None,
billing_tag: Optional[str] = None,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
) -> Union[dict, StreamingJSONObject]:
"""
Get the latest partition metadata in a version range for a versioned layer or a time
range for a volatile layer. For versioned layers the range is expressed as a start and
end version and might not return all changes for the partitions which were added and
removed in between the specified start and end versions.

:param layer_id: Unique layer id. Content of this parameter refers to a valid layer ID.
: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 part: Available/Required for versioned layers only; indicates which part of the
layer shall be queried.
:param additional_fields: Available values : dataSize, checksum, compressedDataSize, crc
:param range_header: an optional Range parameter to resume download of a large response
for versioned layers when there is a connection issue between the client and server.
Specify a single byte range offset like this: Range: bytes=10-. This parameter is
compliant with RFC 7233, but note that this parameter only supports a single byte
range. The range parameter can also be specified as a query parameter, i.e.
range=bytes=10-. For volatile layers use the pagination links returned in the response
body.
:param billing_tag: Billing Tag is an optional free-form tag which is used for grouping
billing records together.
:param stream: whether to stream parsing of the data. Passing True can reduce memory usage
for extremely large responses, but is slower than reading the full JSON in one go.
:param chunk_size: the size to request each iteration when streaming data.
: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, "part": part, "additionalFields": additional_fields_str, "billingTag": billing_tag,}
headers = self.headers
headers["Range"] = range_header
resp = self.get(url, params=params, headers=headers, stream=stream)
if resp.status_code in [200, 206]:
if stream:
return stream_json_response(self, resp, chunk_size)
return resp.json()
else:
raise PlatformException(resp)

[docs]
def get_changes_parts(self, layer_id: str, num_requested_parts: str) -> dict:
"""
Return a list of Part Ids which represent the parts that can be used to limit the scope
of queries of changes for the version range. This allows to run parallel queries with
multiple parts. The user has to provide the desired number of parts and the service will
return a list of Part Ids. Please note in some cases the requested number of parts will
make them too small and in this case the service might return lesser amount of the parts
than requested.

:param layer_id: Unique layer id. Content of this parameter refers to a valid layer ID.
:param num_requested_parts: Indicates requested number of parts.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/changes/parts"
url = self.format_url(self.base_url, path)
params = {"numRequestedParts": num_requested_parts}
resp = self.get(url=url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_partitions(
self,
layer_id: str,
version: Optional[int] = None,
part: Optional[str] = None,
additional_fields: Optional[list] = ["dataSize", "checksum", "compressedDataSize", "crc"],
range_header: Optional[str] = None,
billing_tag: Optional[str] = None,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
) -> Union[dict, StreamingJSONObject]:
"""
Get the metadata for all partitions in a specific layer.

:param layer_id: Unique layer id. Content of this parameter refers to a valid layer ID.
:param version: If you are getting metadata from a versioned layer, specify the version
of the layer you want. This parameter is required for versioned layers. If you are
getting metadata from another layer type, do not specify this parameter.
:param part: Available/Required for versioned layers only; indicates which part of the
layer shall be queried.
:param additional_fields: Available values : dataSize, checksum, compressedDataSize, crc
:param range_header: an optional Range parameter to resume download of a large response
for versioned layers when there is a connection issue between the client and server.
Specify a single byte range offset like this: Range: bytes=10-. This parameter is
compliant with RFC 7233, but note that this parameter only supports a single byte
range. The range parameter can also be specified as a query parameter, i.e.
range=bytes=10-. For volatile layers use the pagination links returned in the response
body.
:param billing_tag: Billing Tag is an optional free-form tag which is used for grouping
billing records together.
:param stream: whether to stream parsing of the data. Passing True can reduce memory usage
for extremely large responses, but is slower than reading the full JSON in one go.
:param chunk_size: the size to request each iteration when streaming data.
:return: response from the API suitable for streaming.
: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 = {"version": version, "part": part, "additionalFields": additional_fields_str, "billingTag": billing_tag,}
headers = self.headers
headers["Range"] = range_header

Always stream the request itself to allow for better retry behavior and avoid buffering

the data before parsing.

resp = self.get(url, params=params, headers=headers, stream=stream)
if resp.status_code in [200, 206]:
if stream:
return stream_json_response(self, resp, chunk_size)
return resp.json()
else:
raise PlatformException(resp)

[docs]
def get_partitions_parts(self, layer_id: str, num_requested_parts: str) -> dict:
"""
Return a list of Part Ids which represent the parts that can be used to limit the scope
of queries of the metadata for all partitions in a specific layer. This allows to run
parallel queries with multiple parts. The user has to provide the desired number of parts
and the service will return a list of Part Ids. Please note in some cases the requested
number of parts will make them too small and in this case the service might return lesser
amount of the parts than requested.

:param layer_id: Unique layer id. Content of this parameter refers to a valid layer ID.
:param num_requested_parts: Indicates requested number of parts.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/partitions/parts"
url = self.format_url(self.base_url, path)
params = {"numRequestedParts": num_requested_parts}
resp = self.get(url=url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def next_partitions(
self,
next_url: str,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
) -> StreamingJSONObject:
"""
Get the metadata for partitions queried either from get_partitions() or get_changes() with
the provided "next" URL when the number of partitions exceeds what can be returned with a
single query.

:param next_url: URL for the next partitions returned from a previous response.
:param stream: whether to stream parsing of the data. Passing True can reduce memory usage
for extremely large responses, but is slower than reading the full JSON in one go.
:param chunk_size: the size to request each iteration when streaming data.
:return: response from the API suitable for streaming.
:raises PlatformException: If platform responds with an HTTP error.
"""

Always stream the request itself to allow for better retry behavior and avoid buffering

the data before parsing.

resp = self.get(url=next_url, stream=stream)
if resp.status_code == 200:
if stream:
return stream_json_response(self, resp, chunk_size)
return resp.json()
else:
raise PlatformException(resp)

[docs]
def list_versions(
self,
start_version: int,
end_version: int,
billing_tag: Optional[str] = None,
) -> dict:
"""
Return information about specific catalog version(s). If the catalog doesn't contain any
versions, an empty dictionary is returned. Maximum number of versions to be returned
per call is 1000 versions. If requested range is bigger than 1000 versions,
400 Bad Request will be returned.

:param start_version: The version number after which list of returned versions will begin.
Minimum value -1 will return list starting with version 0.
:param end_version: The end of the range of versions you want to get (inclusive). This
must be a valid catalog version greater than the startVersion. The maximum value for
this parameter is returned from the /versions/latest endpoint. If this version does
not exist, 400 Bad Request is returned
:param billing_tag: Billing Tag is an optional free-form tag which is used for grouping
billing records together.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/versions"
url = self.format_url(self.base_url, path)
params = {"startVersion": start_version, "endVersion": end_version, "billingTag": billing_tag,}
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
elif resp.status_code == 404:
return
else:
raise PlatformException(resp)

[docs]
def compatible_versions(
self, catalog_dependencies: dict, limit: Optional[int] = 1, next: Optional[str] = None
) -> dict:
"""
Given a list of HRNs and versions provided by the user, returns a list of versions of this
catalog for which the listed HRN are either present in the direct or indirect dependencies
with the same version, or are not present. Please note that versions that don't depend on
any of the provided catalog HRNs, are also considered compatible. The compatible versions
are returned in reverse order, from the newest to the oldest. When there is no compatible
version the service returns an empty list. When the catalog has no version a 404 is
returned.

:param catalog_dependencies: The catalog dependencies we want to search for
:param limit: The numbers of items to return per page
:param next: The next url where the iteration will continue.
:return: response from the API.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/versions/compatibles"
url = next if next else self.format_url(self.base_url, path)
params = {"limit": limit,}
resp = self.post(url, data=catalog_dependencies, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_latest_version(
self, start_version=-1, billing_tag: Optional[str] = None
) -> Optional[int]:
"""
Return information about the latest version for the given catalog. If the catalog doesn't
contain any versions None will be returned.

:param start_version: The catalog version returned from a prior request to
/versions/latest. You should save the version from each request so that you can use it
in the startVersion parameter of subsequent requests. If you don't have the version
from a prior request, set the parameter to -1.
:param billing_tag: Billing Tag is an optional free-form tag which is used for grouping
billing records together.
:return: latest version number of the catalog if version exist else None
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/versions/latest"
url = self.format_url(self.base_url, path)
params = {"startVersion": start_version, "billingTag": billing_tag,}
resp = self.get(url, params=params)
if resp.status_code == 200:
return int(resp.json().get("version"))
elif resp.status_code == 404:
return None
else:
raise PlatformException(resp)

[docs]
def get_minimum_version(self, billing_tag: Optional[str] = None) -> Optional[int]:
"""
Return minimum version for the given catalog. If the catalog doesn't
contain any versions None will be returned.

:param billing_tag: Billing Tag is an optional free-form tag which is used for grouping
billing records together.
:return: an int representing minimum version, if any
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/versions/minimum"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}
resp = self.get(url, params=params)
if resp.status_code == 200:
return int(resp.json().get("version"))
elif resp.status_code == 404:
return None
else:
raise PlatformException(resp)

[docs]
def set_minimum_version(self, version_obj: dict, billing_tag: Optional[str] = None):
"""
Sets minimum version for the given catalog.

:param version_obj: A version object that contains new minimum version.
:param billing_tag: Billing Tag is an optional free-form tag which is used for grouping
billing records together.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/versions/minimum"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}
resp = self.post(url, data=version_obj, params=params)
if resp.status_code == 204:
return
else:
raise PlatformException(resp)