Source code for here.platform.api.data_blob_api

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

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

.. |blob_api_reference| raw:: html

Blob API Reference # noqa E501
"""

from typing import Optional, Union

from here.platform.api.base_api import BaseApi
from here.platform.api.stream import ChunkedGet
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 here.platform.utils.file import checksum

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

The blob service supports the upload and retrieval of large volumes of data from the storage
of a catalog.
"""

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

:param base_url: a Blob 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(DataBlobApi, self).init(
platform_config=platform_config,
application_config=application_config,
auth=auth,
proxies=proxies,
)
self.base_url = base_url

[docs]
def check_handle_exists(
self, layer_id: str, data_handle: str, billing_tag: Optional[str] = None
) -> bool:
"""
Check if a blob exists for the requested data handle.

:param layer_id: The ID of the parent layer for this blob.
:param data_handle: The data handle identifies a specific blob so that you can get that
blob's contents.
:param billing_tag: A string which is used for grouping billing records.
:return: a boolean indicating if the handle exists.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/data/{data_handle}"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}
resp = self.head(url, params=params)
if resp.status_code == 200:
return True
elif resp.status_code == 404:
return False
else:
raise PlatformException(resp)

[docs]
def get_blob_by_handle(
self,
layer_id: str,
data_handle: str,
range_header: Optional[str] = None,
billing_tag: Optional[str] = None,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
storage_layer_access=False,
) -> Union[bytes, ChunkedGet]:
"""
Get blob (raw bytes) for given layer ID and data-handle from storage.

:param layer_id: The ID of the parent layer for this blob.
:param data_handle: The data handle identifies a specific blob so that you can get that
blob's contents.
:param range_header: an optional Range parameter to resume download of a large response
when there is a connection issue between the client and server, or to fetch a specific
slice of the blob. To resume download after a connection issue, specify a single byte
range offset as follows: Range: bytes=10-. To fetch a specific slice of the blob,
specify a slice as follows: Range: bytes=10-100. 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-.
:param billing_tag: A string which is used for grouping billing records.
:param stream: whether to stream data.
:param chunk_size: the size to request each iteration when streaming data.
:param storage_layer_access: whether to request direct access to the storage layer
:return: a blob response as bytes
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/data/{data_handle}"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}

Read directly from S3 through redirect if available.

if not self.is_local and storage_layer_access:
params["accessType"] = "storageLayerAccess"
headers = self.headers
headers["Range"] = range_header
resp = self.get(url, params=params, headers=headers, stream=stream)
if resp.status_code in [200, 206]:
return ChunkedGet(self, resp, chunk_size) if stream else resp.content
else:
raise PlatformException(resp)

[docs]
def delete_blob_by_handle(
self, layer_id: str, data_handle: str, billing_tag: Optional[str] = None
):
"""
Delete blob (raw bytes) for given layer ID and data-handle from storage.

:param layer_id: The ID of the parent layer for this blob.
:param data_handle: The data handle identifies a specific blob so that you can get that
blob's contents.
: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}/data/{data_handle}"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}
resp = self.delete(url, params=params)
if resp.status_code == 202:
return
else:
raise PlatformException(resp)

[docs]
def publish_blob_by_handle(
self,
layer_id: str,
data_handle: str,
data: bytes,
content_length: int,
content_encoding: Optional[str] = None,
content_type: Optional[str] = None,
billing_tag: Optional[str] = None,
) -> None:
"""
Publish blob (raw bytes) for given layer ID.

:param layer_id: The ID of the parent layer for this blob.
:param data_handle: The data handle (ID) represents an identifier for the data blob.
:param data: blob data in bytes
:param content_length: Size of the entity-body, in bytes. For more information,
see RFC 7230, section 3.3.2: Content-Length.
:param content_encoding: A string representing content encodings applied to data.
:param content_type: A string representing format of the data. The value of
this field must be equal to the one specified in the contentType field in
the catalog layer configuration.
: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}/data/{data_handle}"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}
headers = self.headers
headers["Content-Length"] = str(content_length)
headers["Content-Encoding"] = content_encoding
headers["Content-Type"] = content_type
headers["X-HERE-Digest"] = f"SHA-256:{checksum(data, 'sha256')}"
resp = self.put(url, data=data, headers=headers, params=params)
if resp.status_code == 200:
return
else:
raise PlatformException(resp)

[docs]
def start_multipart_upload_by_handle(
self,
layer_id: str,
data_handle: str,
content_type: str,
content_encoding: str = "identity",
billing_tag: Optional[str] = None,
) -> dict:
"""
Publish large data blobs where the data payload needs to be split into multiple parts.
The multipart upload start is to be followed by the individual parts upload and completed
with a call to complete the upload. The limit of the blob uploaded this way is 50GB.

:param layer_id: The ID of the parent layer for this blob.
:param data_handle: The data handle (ID) represents an identifier for the data blob.
:param content_type: A string representing format of the blob data. The value of
this field must be equal to the one specified in the contentType field in
the catalog layer configuration.
:param content_encoding: A string representing content encodings applied to blob.
: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}/data/{data_handle}/multiparts"
url = self.format_url(self.base_url, path)
params = {"billingTag": billing_tag}
data = dict()
data["contentEncoding"] = content_encoding
data["contentType"] = content_type
resp = self.post(url, data=data, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_multipart_upload_status_by_handle(
self,
layer_id: str,
data_handle: str,
multipart_token: str,
billing_tag: Optional[str] = None,
) -> dict:
"""
Get the status of a multipart upload by handle.

[docs]
def complete_multipart_upload_by_handle(
self,
complete_href_url: str,
parts: list,
billing_tag: Optional[str] = None,
):
"""
Completes a multipart upload by handle.

:param complete_href_url: The complete href url.
:param parts: A list of dict representing part_ids uploaded in multipart upload.
:param billing_tag: A string which is used for grouping billing records.
:raises PlatformException: If platform responds with an HTTP error.
"""
params = {"billingTag": billing_tag}
data = dict(parts=parts)
resp = self.put(complete_href_url, data=data, params=params)
if resp.status_code in [200, 204]:

Note: 200 is returned only by LDS

return
else:
raise PlatformException(resp)

[docs]
def cancel_multipart_upload_by_handle(
self,
layer_id: str,
data_handle: str,
multipart_token: str,
billing_tag: Optional[str] = None,
):
"""
Cancel an entire multipart upload operation by handle.

[docs]
def upload_part_by_handle(
self,
layer_id: str,
data_handle: str,
multipart_token: str,
part_number: int,
content_length: int,
content_type: str,
billing_tag: Optional[str] = None,
) -> dict:
"""
Upload a single part of a multipart upload for the blob. Every uploaded part except the
last one must have a minimum 5 MB of data and a maximum of 96 MB. The length of every part
except the last one must be a multiple of 1MB (1024KB). The maximum number of parts is
10,000.

:param layer_id: The ID of the parent layer for this blob.
:param data_handle: The data handle (ID) represents an identifier for the data blob.
:param multipart_token: The identifier of the multipart upload (token). Content of this
parameter must refer to a valid token which when the multipart upload was 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. Do not reuse the same partNumber
when retrying an upload in an error situation (network problems, 4xx or 5xx responses)
Reusing the same partNumber in a retry may cause the publication to fail.
:param content_length:Size of the entity-body, in bytes. For more information, see
RFC 7230, section 3.3.2: Content-Length.
: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 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}/data/{data_handle}/multiparts/{multipart_token}/parts"
url = self.format_url(self.base_url, path)
params = {"partNumber": part_number, "billingTag": billing_tag}
headers = self.headers
headers["Content-Length"] = content_length
headers["Content-Type"] = content_type
resp = self.post(url, params=params, headers=headers)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)