Copyright (C) 2020-2026 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.

"""HERE platform catalog layer partition abstraction."""

import asyncio
import json
import logging
import time
import uuid
from contextlib import contextmanager
from datetime import datetime
from io import BytesIO
from pathlib import Path
from typing import TYPE_CHECKING, BinaryIO, Dict, Iterator, Optional, Union, cast

import backoff
from aiohttp import (
ClientConnectionError,
ClientError,
ClientResponse,
ClientSession,
ClientTimeout,
ConnectionTimeoutError,
SocketTimeoutError,
)
from here.platform.constants import DEFAULT_ITER_CHUNK_SIZE
from here.platform.exceptions import (
AuthenticationException,
DigestMismatchException,
InequalReadsException,
NonceAlreadyUsedException,
PlatformException,
ResourceLimitExceededException,
ServiceUnavailableException,
TooManyRequestsException,
)
from here.platform.utils.deprecation import _deprecate_remove_parameters
from here.platform.utils.file import checksum

if TYPE_CHECKING:
from here.platform.layer import IndexLayer, StreamLayer, VersionedLayer, VolatileLayer

logger = logging.getLogger(name)

[docs]
class Partition:
"""HERE platform partition abstraction."""

def init(
self,
data_handle: Optional[str] = None,
layer=None, # type: ignore
id: Optional[str] = None,
checksum: Optional[str] = None,
data_size: Optional[int] = None,
crc: Optional[str] = None,
):
"""Instantiate Partition given a data handle, layer and other parameters."""
assert layer
self.data_handle = data_handle
self.layer = layer
self.id = id
self.checksum = checksum
self.data_size = data_size
self.crc = crc
self.billing_tag = self.layer.billing_tag # type: ignore

[docs]
def get_blob(
self, stream: bool = False, chunk_size: int = DEFAULT_ITER_CHUNK_SIZE
) -> Union[bytes, Iterator[bytes]]:
"""
Get blob (raw bytes) inside the partition for this layer with given
data_handle.

:param stream: whether to stream data.
:param chunk_size: the size to request each iteration when streaming data.
:return: Content of the blob referenced by the data handle.
:raises ValueError: Unsupported content encoding or invalid data handle
"""
if not self.data_handle:
raise ValueError(f"data_handle not set for partition {self.id}")
volume_type = self.layer.configuration.json["volume"]["volumeType"]
if volume_type == "volatile":
blob = self.layer._data_volatile_blob_api.get_volatile_blob(
layer_id=self.layer.id,
data_handle=self.data_handle,
billing_tag=self.billing_tag,
stream=stream,
chunk_size=chunk_size,
)
else:
blob = self.layer._data_blob_api.get_blob_by_handle(
layer_id=self.layer.id,
data_handle=self.data_handle,
billing_tag=self.billing_tag,
stream=stream,
chunk_size=chunk_size,
storage_layer_access=self.layer.has_storage_layer_access,
)
if self.layer.configuration.content_encoding not in ("gzip", "identity", None):
raise ValueError(
f"Unsupported content encoding : {self.layer.configuration.content_encoding}"
)
return cast(Union[bytes, Iterator[bytes]], blob)

[docs]
@staticmethod
def generate_data_handle() -> str:
"""
Generate a unique data handle.

:return: A string representing a unique blob identifier
"""
return str(uuid.uuid4())

def __current_time(self):
"""This is a time.time() wrapper to make it testable in a deterministic way"""
return time.time()

def __generate_new_handle(self):
start_time = self.__current_time()
while self.__current_time() - start_time <= self.layer.catalog.platform._retry_max_time:
data_handle = self.generate_data_handle()
if self.layer._data_blob_api.check_handle_exists(
self.layer.id, data_handle=data_handle
):
time.sleep(self.layer.catalog.platform._polling_wait)
else:
self.data_handle = data_handle
break
if not self.data_handle:
raise ValueError("Data handle not generated, max retry timeout reached")

[docs]
def put_blob(self, path_or_data: Union[str, bytes, Path]):
"""
Upload data in single part.

:param path_or_data: Path of the file or data in bytes to be uploaded as a partition.
"""
if not self.data_handle:
self.__generate_new_handle()

if isinstance(path_or_data, bytes):
data = path_or_data
else:
with open(path_or_data, "rb") as file_data:
data = file_data.read()
data_length = len(data)
content_encoding = self.layer.configuration.content_encoding
content_type = self.layer.configuration.content_type
self.layer._data_blob_api.publish_blob_by_handle(
layer_id=self.layer.id,
data_handle=self.data_handle,
data=data,
content_length=data_length,
content_encoding=content_encoding,
content_type=content_type,
billing_tag=self.billing_tag,
)

[docs]
def multipart_upload(
self,
path_or_data: Union[str, bytes, Path],
part_size: int,
):
"""
Multipart upload data to a blob store.

:param path_or_data: Path of the file or data in bytes to be upload as a partition.
:param part_size: An int representing size in MB, to upload in multiple parts
minimum value is 5MB and and Maximum is 50MB.

:raises ValueError: Data handle not generated, max retry timeout reached
"""
if not (5 <= part_size <= 50):
raise ValueError(f"part_size must be between 5 and 50 MB, got {part_size}")

if not self.data_handle:
self.__generate_new_handle()

content_type = self.layer.configuration.content_type
assert content_type
content_encoding = self.layer.configuration.content_encoding or "identity"
multi_parts = self.layer._data_blob_api.start_multipart_upload_by_handle(
layer_id=self.layer.id,
data_handle=self.data_handle,
content_type=content_type,
billing_tag=self.billing_tag,
content_encoding=content_encoding,
)

upload_url = multi_parts["links"]["uploadPart"]["href"]
logger.debug(f"URL: {upload_url}")
try:
loop = asyncio.get_event_loop()
except RuntimeError:
logger.info("No current event loop, creating new one")
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
headers = self.layer._base_api.headers.copy()
future = asyncio.ensure_future(
self.read_and_upload_parts(path_or_data, upload_url, content_type, headers, part_size)
)
results = loop.run_until_complete(future)
parts = []
for result in results:
for k, v in result.items():

Case-insensitive header lookup for ETag

etag = v.get("Etag") or v.get("etag") or v.get("ETag")
if etag:
parts.append(dict(etag=etag, number=k))
else:

Throw error instead of returning result.

raise ValueError("Expected header ETag in response: ", result)

complete multipart upload

self.layer._data_blob_api.complete_multipart_upload_by_handle(
complete_href_url=multi_parts["links"]["complete"]["href"],
parts=parts,
billing_tag=self.billing_tag,
)

[docs]
async def read_and_upload_parts(
self,
path_or_data: Union[str, bytes, Path],
upload_url: str,
content_type: str,
headers: Dict[str, str],
part_size: int,
):
"""
Read input file in chunks and upload each part to the specified URL.

:param path_or_data: Path of the file or data to be read and upload as a partition.
:param upload_url: An URL to upload blob data.
:param content_type: A standard MIME type describing the format of the blob data.
:param headers: A dict containing http headers.
:param part_size: An int representing size in MB, to upload in multiple parts
minimum value is 5MB and and Maximum is 50MB.
:return: A list of dict with key as part number and value as url response body.
"""
tasks = []
completed_tasks = []
application_config = self.layer.platform.application_config
with self.get_data_handler(path_or_data) as data_handler:
timeout = ClientTimeout(
total=None,
connect=application_config.connect_timeout,
sock_read=application_config.read_timeout,
)
async with ClientSession(
timeout=timeout,
trust_env=True,
) as session:
data = data_handler.read(part_size 1024 1024) # reading in part_size MB chunks
part_num = 1
headers["Content-Type"] = content_type
while data:
logger.debug(f"Starting part number: {part_num}")
upd_url = f"{upload_url}?partNumber={part_num}"
if self.billing_tag:
upd_url = "".join([upd_url, f"&billingTag={self.billing_tag}"])
task = asyncio.ensure_future(
self.upload_part(session, upd_url, part_num, headers, data)
)
part_num += 1
tasks.append(task) # create list of tasks
if len(tasks) % 5 == 0:
completed_tasks.extend(await asyncio.gather(*tasks))
logger.info(f"Completed parts: {len(completed_tasks)}")
tasks = []
data = data_handler.read(part_size 1024 1024)
completed_tasks.extend(await asyncio.gather(*tasks))
return completed_tasks # gather task responses

[docs]
@staticmethod
@contextmanager
def get_data_handler(path_or_data: Union[str, bytes, Path]):
"""Get the data handler to read from file path or data.

:param path_or_data: Path of the file or data to be read and uploaded as a partition.
:yield Union[BytesIO, BinaryIO]: Data Handle
"""
try:
if isinstance(path_or_data, bytes):
handle = BytesIO(path_or_data) # type: Union[BytesIO, BinaryIO]
yield handle
else:
handle = open(path_or_data, "rb")
yield handle
finally:
handle.close()

[docs]
async def upload_part(
self,
session: ClientSession,
upload_url: str,
part_num: int,
headers: Dict[str, str],
data: bytes,
) -> Dict:
"""
Upload a part data to a url, using specified ClientSession.

:param session: a ClientSession object for making requests.
:param upload_url: an URL to upload blob data.
:param part_num: an unique number for the part to be uploaded.
:param headers: HTTP headers to be send with the request.
:param data: blob data to be send with the request.

:return: Returns a dict with key as part number and value as response body.
"""

logger.debug(f"Upload URL: {upload_url}")

headers = headers.copy() # avoid mutating input headers
headers["User-Agent"] = self.layer._base_api._user_agent

Use x-idempotency-key to uniquely identify the request in case of retries.

headers["x-idempotency-key"] = str(uuid.uuid4())

if self.layer._has_gz_encoding():
headers["Content-Encoding"] = "gzip"

data_length = len(data)
logger.debug(f"Data Length: {data_length}")
headers["Content-Length"] = str(data_length)

headers["X-HERE-Digest"] = f"SHA-256:{checksum(data, 'sha256')}"

class ResponseWrapper:
"""Lightweight wrapper for aiohttp response compatible with exception handlers."""

def init(self, aio_response: ClientResponse, content: bytes):
self._aio_response = aio_response
self._content = content

@property
def status_code(self) -> int:
return self._aio_response.status

@property
def reason(self) -> str:
return self._aio_response.reason or ""

@property
def headers(self) -> Dict[str, str]:
return dict(self._aio_response.headers)

@property
def content(self) -> bytes:
return self._content

@property
def url(self) -> str:
return str(self._aio_response.url)

@property
def text(self) -> str:
return self._content.decode(errors="replace")

Inner backoff loop retries on connection error. Outer backoff loop retries based on too

many requests.

application_config = self.layer.platform.application_config

Get url protocol

protocol = None
if "://" in upload_url:
protocol = upload_url.split("://")[0].lower()

Get proxy from Platform if available

proxy = None
proxies = self.layer.platform.proxies
if proxies and protocol and protocol in proxies.keys():
proxy = proxies[protocol]

@backoff.on_exception(
backoff.expo,
(
DigestMismatchException,
InequalReadsException,
NonceAlreadyUsedException,
ServiceUnavailableException,
SocketTimeoutError,
TooManyRequestsException,
),
max_time=application_config.retry_max_time,
max_value=application_config.retry_max_wait,
)
@backoff.on_exception(
backoff.constant,
(
ClientConnectionError,
ClientError,
ConnectionError,
ConnectionTimeoutError,
),
max_time=application_config.connect_retry_max_time,
)
async def upload_part_impl() -> Dict[str, str]:
"""
Perform a POST request of part upload API at the specified URL.

:raises TooManyRequestsException: If platform responds with HTTP 429.
:raises ServiceUnavailableException: If platform responds with HTTP 408, 500, 502, 503,
or 504.
:raises AuthenticationException: If platform responds with HTTP 401 or 403.
:raises NonceAlreadyUsedException: If platform responds with HTTP 401 due to nonce
re-use and retries have been exceeded.
:raises ResourceLimitExceededException: If platform responds with HTTP 402.
:raises DigestMismatchException: If platform responds with HTTP 400 due to the provided
digest not matching the data.
:raises PlatformException: If platform does not respond with 200, 201 or any of the
above.
:return: response headers from the API.
"""
logger.debug(f"POST {upload_url}")
start_time = datetime.now()

async with session.post(
upload_url,
headers=headers,
data=data,
proxy=proxy,
) as response:

Read response content

content = await response.read()
run_time = datetime.now() - start_time

Create lightweight response wrapper

resp = ResponseWrapper(response, content)

logger.debug(
f"POST {upload_url} - {resp.status_code} - {run_time.total_seconds()} sec."
)
logger.debug(f"Headers: {resp.headers}")

if resp.status_code == 429:
raise TooManyRequestsException(resp)

Request timeout, internal server error, invalid gateway, service unavailable,

or gateway timeout.

elif resp.status_code in [408, 500, 502, 503, 504]:
raise ServiceUnavailableException(resp)
elif resp.status_code == 402:
raise ResourceLimitExceededException(resp)
elif resp.status_code in [401, 403]:
try:

Special error case for the nonce already used, which just needs a retry.

is_nonce_error = (
json.loads(resp.content)["errorCode"]
== NonceAlreadyUsedException.AAA_ERROR_CODE
)
except Exception:

If the error code couldn't be retrieved, assume it's not a nonce error.

is_nonce_error = False
if is_nonce_error:
raise NonceAlreadyUsedException(resp)
raise AuthenticationException(resp)
elif resp.status_code == 400:

Check for digest mismatch.

is_digest_error = False
try:
for detail in json.loads(resp.content)["detail"]:
if detail.get("name") == "digest":
is_digest_error = True
break
except Exception:

If the error name couldn't be retrieved, assume it's not a digest error.

pass
if is_digest_error:
raise DigestMismatchException(resp)
elif resp.status_code not in [200, 201]:
raise PlatformException(resp)

Return response headers as a dict

return dict(response.headers)

response_headers = await upload_part_impl()
upload_part_resp: Dict = {part_num: response_headers}
logger.debug(f"Response: {upload_part_resp}")
return upload_part_resp

[docs]
def upload_volatile_blob(self, path_or_data: Union[str, Path, bytes]):
"""
Upload data to volatile blob.

:param path_or_data: Path of the file or data to be read and upload as a partition
:raises ValueError: in case data handle is not set
"""
if not self.data_handle:
raise ValueError(f"data_handle not set for partition {self.id}")
with Partition.get_data_handler(path_or_data) as data_handler:
data = data_handler.read()
self.layer._data_volatile_blob_api.put_volatile_blob(
layer_id=self.layer.id,
data_handle=self.data_handle,
body=data,
billing_tag=self.billing_tag,
)

[docs]
class VersionedPartition(Partition):
"""
Partition subclass used for VersionedLayers.

In addition to the fields present in the base class, this class adds:

• version: version of the partition
  """

def init(
self,
data_handle: Optional[str],
layer: "VersionedLayer",
id: str,
checksum: Optional[str] = None,
data_size: Optional[int] = None,
compressed_data_size: Optional[int] = None,
crc: Optional[str] = None,
version: Optional[int] = None,
):
"""Instantiate VersionedPartition given a data handle, layer and other parameters."""
super().init(
data_handle=data_handle,
layer=layer,
id=id,
checksum=checksum,
data_size=data_size,
crc=crc,
)
self.version = version
self.compressed_data_size = compressed_data_size

[docs]
class VolatilePartition(Partition):
"""
Partition subclass used for VolatileLayers.

"""

def init(
self,
data_handle: Optional[str],
layer: "VolatileLayer",
id: Optional[str] = None,
checksum: Optional[str] = None,
data_size: Optional[int] = None,
compressed_data_size: Optional[int] = None,
crc: Optional[str] = None,
):
"""Instantiate VolatilePartition given a data handle, layer and other parameters."""
super().init(
data_handle=data_handle,
layer=layer,
id=id,
checksum=checksum,
data_size=data_size,
crc=crc,
)
self.compressed_data_size = compressed_data_size

[docs]
class IndexPartition(Partition):
"""
Partition subclass used for IndexLayers.

In addition to the fields present in the base class, this class adds:

• timestamp: timestamp of the partition
• fields: fields according to the index layer configuration
• additional_metadata: free-form additional metadata
  """

@_deprecate_remove_parameters(
remove_params=["id"],
since_version="2.12",
use_new_params=["data_handle"],
)
def init(
self,
layer: "IndexLayer",
data_handle: Optional[str] = None,
checksum: Optional[str] = None,
data_size: Optional[int] = None,
crc: Optional[str] = None,
timestamp: Optional[int] = None,
fields: Dict[str, Union[str, int, bool]] = ,
additional_metadata: Dict[str, str] = ,
):
"""Instantiate IndexPartition given a data handle, layer and other parameters."""

Index layers seems not to have partition ids,

the convention is to replicate data handles as ids

data handle is called id in the REST API

super().init(
data_handle=data_handle,
layer=layer,
id=data_handle,
checksum=checksum,
data_size=data_size,
crc=crc,
)
self.timestamp = timestamp
self.fields = fields
self.additional_metadata = additional_metadata

[docs]
class StreamPartition(Partition):
"""
Partition subclass used for StreamLayers.

In addition to the fields present in the base class, this class adds:

• data: inline data, alternative to data_handle
• timestamp: timestamp in milliseconds since the Unix epoch (1970-01-01T00:00:00 UTC)
• kafka_offset: the offset of the message in the Kafka stream partition
• kafka_partition: the Kafka stream partition number the offset is related to
  """

def init(
self,
layer: "StreamLayer",
data_handle: Optional[str] = None,
id: Optional[str] = None,
checksum: Optional[str] = None,
data_size: Optional[int] = None,
crc: Optional[str] = None,
data: Optional[bytes] = None,
timestamp: Optional[int] = None,
kafka_partition: Optional[int] = None,
kafka_offset: Optional[int] = None,
):
"""Instantiate StreamPartition given a data handle, layer and other parameters."""
super().init(
data_handle=data_handle,
layer=layer,
id=id,
checksum=checksum,
data_size=data_size,
crc=crc,
)
self.data = data
self.timestamp = timestamp
self.kafka_offset = kafka_offset
self.kafka_partition = kafka_partition

[docs]
def get_data(
self, stream: bool = False, chunk_size: int = DEFAULT_ITER_CHUNK_SIZE
) -> Union[bytes, Iterator[bytes]]:
"""
Get the data associated with the StreamPartition.

Data can be present directly in the data field
or retrieved via data_handle. This function returns
the data, regardless of where it is stored.

:return: Data associated with the stream partition
:param stream: whether to stream data
:param chunk_size: the size to request each iteration when streaming data
:raises ValueError: in case neither in data nor data_handle is available
"""
if self.data:
if self.layer.configuration.content_encoding not in ("gzip", "identity", None):
raise ValueError(
f"Unsupported content encoding : {self.layer.configuration.content_encoding}"
)
if stream:

Ensure consistency of API and expectations of return types.

def generator() -> Iterator[bytes]:
assert self.data
yield self.data

return generator()
else:
return self.data
elif self.data_handle:
return self.get_blob(stream=stream, chunk_size=chunk_size)
else:
raise ValueError("Unsupported stream format. Expected data or data_handle attribute.")