Copyright (C) 2021-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:DataVolatileBlobApi class to perform API operations.

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

.. |volatile_blob_api_reference| raw:: html

By Key Object Blob API Reference By Key # noqa E501
By Handle Object Blob API Reference By Handle # noqa E501

"""

from typing import Any, Optional

from here.platform.api.base_api import BaseApi
from here.platform.auth import Auth
from here.platform.config.application_config import ApplicationConfig
from here.platform.config.platform_config import PlatformConfig
from here.platform.exceptions import PlatformException
from here.platform.utils.file import checksum
from requests import Response

[docs]
class DataObjectBlobApi(BaseApi):
"""
This class provides access to HERE platform Data Blob APIs v2.

The data-blob-api-v2 service supports the upload and retrieval of objects stored in
ObjectStore layers. Two types of uploads can be done with this layer - Single and Multipart.
The single part upload can be used for objects up to 192 MB.
Use multipart upload for any data objects larger than 192 MB. Every multipart upload uses
the token from previous multipart upload response. Before doing any multipart upload,
wait for the response from previous multipart upload, if any.
Also, while reading, objects can be read in part by passing a Range Header.
Example, Range: bytes=10-
"""

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

:param base_url: base_url for the request
: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().init(
platform_config=platform_config,
application_config=application_config,
auth=auth,
proxies=proxies,
)
self.base_url = base_url

[docs]
def key_exists(self, layer_id: str, key: str, billing_tag: Optional[str] = None) -> bool:
"""
Checks whether the object with :param:key exists in the object store layer.
"""
path = f"/layers/{layer_id}/keys/{key}"
params = {"billingTag": billing_tag}
url = "".format(self.base_url, path)
resp = self.head(url, params=params)
return resp.status_code in [200, 204]

[docs]
def get_object(
self,
layer_id: str,
key: str,
only_meta: bool,
billing_tag: Optional[str] = None,
stream: bool = False,
storage_layer_access=False,
) -> Response:
"""
Fetches the object against a :param:key in layer with ID :param:layer_id.

:param layer_id: ID of the object store layer
:param key: key for the object to be fetched
:param only_meta: fetch only metadata
:param billing_tag: A string to represent a grouping of billing records
:param stream: whether to stream data
:param storage_layer_access: whether to request direct access to the storage layer
:returns: object data against given :param:key
:raises PlatformException: If platform responds with an HTTP error
"""
path = f"/layers/{layer_id}/keys/{key}"
params = {"billingTag": billing_tag}

Read directly from S3 through redirect if available.

if not only_meta and not self.is_local and storage_layer_access:
params["accessType"] = "storageLayerAccess"
url = "".format(self.base_url, path)
resp = self.request("HEAD" if only_meta else "GET", url, params=params, stream=stream)
if resp.status_code in [200, 204]:
return resp
else:
raise PlatformException(resp)

[docs]
def put_object(
self,
layer_id: str,
key: str,
object_data: Optional[bytes] = None,
headers: Optional[dict] = None,
source: Optional[str] = None,
billing_tag: Optional[str] = None,
) -> Any:
"""
Uploads the :param:object_data against given :param:key
in layer with ID :param:layer_id.

:param layer_id: ID of the object store layer
:param key: key for the object to be uploaded
:param object_data: object to be uploaded
:param headers: additional request header fields like content-length,
and content-type which can be provided while uploading object.
:param source: source to copy an existing object, instead of uploading a new one,
If this parameter is present, the object_data must be empty.
:param billing_tag: A string to represent a grouping of billing records
:returns: status response after putting the object
:raises PlatformException: If platform responds with an HTTP error
"""
path = f"/layers/{layer_id}/keys/{key}"
params = {"billingTag": billing_tag}
if source:
params["source"] = source
url = "".format(self.base_url, path)
complete_headers = {**self.headers}
if headers:
complete_headers = {**self.headers, **headers}
if object_data:
complete_headers["X-HERE-Digest"] = f"SHA-256:{checksum(object_data, 'sha256')}"
resp = self.put(url, data=object_data, params=params, headers=complete_headers)
if resp.status_code in [200, 204]:
return resp.content
else:
raise PlatformException(resp)

[docs]
def delete_object(
self,
layer_id: str,
key: str,
billing_tag: Optional[str] = None,
):
"""
Deletes the object with :param:key from the object store layer.

:param layer_id: ID of the object store layer.
:param key: key for the object to be deleted.
:param billing_tag: A string to represent a grouping of billing records
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/keys/{key}"
params = {"billingTag": billing_tag}
url = "".format(self.base_url, path)
resp = self.delete(url, params=params)
if resp.status_code == 202:
return
else:
raise PlatformException(resp)

[docs]
def list_keys(self, layer_id: str, params: dict) -> Any:
"""
List all the keys present in the object store layer.
"""
path = f"/layers/{layer_id}/keys"
url = "".format(self.base_url, path)
resp = self.get(url, params=params)
if resp.status_code in [200, 204]:
return resp.json()
else:
raise PlatformException(resp)

[docs]
def start_multipart_upload(
self,
layer_id: str,
key: str,
content_type: str,
billing_tag: Optional[str] = None,
content_encoding: Optional[str] = None,
):
"""
The multipart upload start is to be followed by the individual parts upload and
completed with a call to complete the upload

:param layer_id: ID of the object store layer.
:param key:The key identifies a specific blob so that you can get that blob's contents
:param content_type: A standard MIME type describing the format of the blob data.
:param billing_tag: A string to represent a grouping of billing records
:param content_encoding: Content-encoding of the object. This field is optional.
For more information, see https://tools.ietf.org/html/rfc2616#section-14.11
:return: returns multipart token
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/keys/{key}"
params = {"billingTag": billing_tag, "flags": "alignToMB"}
url = "".format(self.base_url, path)
data = dict()
data["contentType"] = content_type
if content_encoding:
data["contentEncoding"] = content_encoding
resp = self.post(url, data=data, params=params)
if resp.status_code == 200:
return resp.json()
else:
raise PlatformException(resp)

[docs]
def get_multipart_upload_status(
self,
layer_id,
multipart_token,
billing_tag: Optional[str] = None,
):
"""
Gets the status of a multipart upload.
The status can be received only when the upload has been completed.

:param layer_id: ID of the object store layer.
:param multipart_token: The identifier of the multipart upload (token).
This token is returned when the multipart upload is initiated.
:param billing_tag: A string to represent a grouping of billing records
:return: returns status of multipart upload.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/keysMultipart/{multipart_token}"
params = {"billingTag": billing_tag}
url = "".format(self.base_url, path)
resp = self.get(url, params=params)
if resp.status_code == 200:
return resp.json()
else:
raise PlatformException(resp)

[docs]
def cancel_multipart_upload(
self,
layer_id,
multipart_token,
billing_tag: Optional[str] = None,
):
"""
Cancels an entire multipart upload operation. You can only cancel
a multipart upload before
it has been completed.

:param layer_id: ID of the object store layer.
:param multipart_token: The identifier of the multipart upload (token).
This token is returned when the multipart upload is initiated.
:param billing_tag: A string to represent a grouping of billing records
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/keysMultipart/{multipart_token}"
params = {"billingTag": billing_tag}
url = "".format(self.base_url, path)
resp = self.delete(url, params=params)
if resp.status_code == 202:
return
else:
raise PlatformException(resp)

[docs]
async def upload_blob_part(
self,
layer_id: str,
multipart_token: str,
part_number: int,
object_part: bytes,
content_type: str,
billing_tag: Optional[str] = None,
):
"""
The multipart upload start is to be followed by the individual parts upload and
completed with a call to complete the upload

:param layer_id: ID of the object store layer.
:param multipart_token: The identifier of the multipart upload (token).
This token is returned when the multipart upload is initiated.
:param part_number: The number of the part for the multi part upload. The numbers of the
upload parts must start from 1, be no greater than 10,000 and be consecutive. Parts
uploaded with the same partNumber are overridden.
:param object_part: object part to be uploaded
:param content_type: A standard MIME type describing the format of the blob data. For more
information, see RFC 2616, section 14.17: Content-Type. The value of this header must
match the content type specified in the contentType field when the multipart upload
was initialized, and this content type must also match the content type specified in
the layer's configuration.
:param billing_tag: A string to represent a grouping of billing records
:return: returns multipart token
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/keysMultipart/{multipart_token}/parts"
url = "".format(self.base_url, path)
params = {"partNumber": part_number, "billingTag": billing_tag,}
headers = self.headers
headers["Content-Type"] = content_type
response = self.post(url, data=object_part, params=params, headers=headers)
if response.status_code == 200:
return response.content
else:
raise PlatformException(response)

[docs]
def complete_multipart_upload(
self,
layer_id: str,
multipart_token: str,
data: dict,
billing_tag: Optional[str] = None,
):
"""
Completes a multipart upload by handle.

:param layer_id: ID of the object store layer.
:param multipart_token: The identifier of the multipart upload (token).
This token is returned when the multipart upload is initiated.
:param data: It is list of dict object which holds id of uploaded parts and part_number
:param billing_tag: A string which is used for grouping billing records.
:raises PlatformException: If platform responds with an HTTP error.
"""

path = f"/layers/{layer_id}/keysMultipart/{multipart_token}"
params = {"billingTag": billing_tag}
url = "".format(self.base_url, path)
resp = self.put(url, data=data, params=params)
if resp.status_code in [200, 204]:

Note: 200 is returned only by LDS

return
else:
raise PlatformException(resp)