Source code for here.platform.layer

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:Layer class to access data in HERE platform catalogs.
"""
import asyncio
import copy
import enum
import gzip
import hashlib
import json
import logging
import os
import re
import time
import webbrowser
from ast import literal_eval
from base64 import b64decode, b64encode
from dataclasses import dataclass
from datetime import datetime
from itertools import islice, zip_longest
from pathlib import Path
from typing import (
TYPE_CHECKING,
Any,
Dict,
Generator,
Iterable,
Iterator,
List,
Mapping,
Optional,
Tuple,
Union,
cast,
)
from uuid import uuid4

from geojson import Feature, FeatureCollection
from geojson.geometry import Geometry
from geojson.mapping import GEO_INTERFACE_MARKER
from here.platform.adapter import Adapter
from here.platform.adapter_default import DefaultAdapter
from here.platform.api.base_api import BaseApi
from here.platform.api.data_blob_api import DataBlobApi
from here.platform.api.data_config_api import DataConfigApi
from here.platform.api.data_index_api import DataIndexApi
from here.platform.api.data_ingest_api import DataIngestApi
from here.platform.api.data_interactive_api import DataInteractiveApi
from here.platform.api.data_metadata_api import DataMetadataApi
from here.platform.api.data_object_blob_api import DataObjectBlobApi
from here.platform.api.data_publish_api import DataPublishApi
from here.platform.api.data_query_api import DataQueryApi
from here.platform.api.data_statistics_api import DataStatisticsApi
from here.platform.api.data_stream_api import DataStreamApi
from here.platform.api.data_volatile_blob_api import DataVolatileBlobApi
from here.platform.api.stream import ChunkedGet, find_list_of_objects
from here.platform.constants import (
DEFAULT_INLINE_STREAM_DATA_LIMIT,
DEFAULT_ITER_CHUNK_SIZE,
MAX_SIZE_VOLATILE_PARTITION,
)
from here.platform.exceptions import (
LayerConfigurationException,
PlatformException,
UnsupportedContentTypeDecodeException,
UnsupportedContentTypeEncodeException,
)
from here.platform.model import Publication
from here.platform.model.subscription import (
InteractiveMapSubscription,
InteractiveMapSubscriptionType,
)
from here.platform.models import Coverage, DurableVolume, Partitioning, VolatileVolume, Volume
from here.platform.partition import (
IndexPartition,
Partition,
StreamPartition,
VersionedPartition,
VolatilePartition,
)
from here.platform.schema import Schema
from here.platform.utils import JsonDictDocument
from here.platform.utils.collection import grouper, iter_tuples
from here.platform.utils.file import checksum, get_crc, get_readable_bytes
from json_stream.base import StreamingJSONObject, TransientAccessException
from kafka import KafkaConsumer, KafkaProducer
from kafka.oauth.abstract import AbstractTokenProvider
from requests import Response

if TYPE_CHECKING:
import geopandas as gpd
import pandas as pd
from here.platform.catalog import Catalog

logger = logging.getLogger(name)

[docs]
class LayerType(enum.Enum):
"""
LayerType enum defines the different layer types supported.

Supported types: versioned, index, volatile, stream, interactivemap, objectstore.

The string representation is lowercase, to match with strings
used in the platform APIs.
"""

UNKNOWN = 0 # This is to be forward compatible with new layer types

VERSIONED = 1
INDEX = 2
VOLATILE = 3
STREAM = 4
INTERACTIVEMAP = 5
OBJECTSTORE = 6

[docs]
@classmethod
def from_str(cls, s: str) -> "LayerType":
"""Create a LayerType from a string, this is case-insensitive"""
s_upper = s.upper()
return LayerType[s_upper] if s_upper in LayerType.members else LayerType.UNKNOWN

def str(self) -> str:
"""The layer type as string, lowercase"""
return self.name.lower()

[docs]
class LayerConfiguration(JsonDictDocument):
"""
The configuration of a layer, including its most significant properties
"""

@property
def id(self) -> Optional[str]:
"""The ID of the layer"""
return self.json.get("id")

@property
def hrn(self) -> Optional[str]:
"""The HERE Resource Name (HRN) of the layer"""
return self.json.get("hrn")

@property
def name(self) -> str:
"""The name of the layer"""
return str(self.json["name"])

@property
def summary(self) -> Optional[str]:
"""The summary of the layer"""
return self.json.get("summary")

@property
def description(self) -> Optional[str]:
"""A longer description of the layer"""
return self.json.get("description")

@property
def coverage(self) -> Optional[Coverage]:
"""The geographic area that this layer covers"""
return Coverage(self.json["coverage"]) if "coverage" in self.json else None

@property
def partitioning(self) -> Optional[Partitioning]:
"""Describes the way in which data is partitioned within the layer"""
return Partitioning(self.json["partitioning"]) if "partitioning" in self.json else None

@property
def volume(self) -> Optional[Volume]:
"""Describes the volume to be used for storing the layer's data content"""
if "volume" in self.json:
volume_config = self.json["volume"]
if "volumeType" in volume_config and volume_config["volumeType"] == "durable":
return DurableVolume(volume_config)
elif "volumeType" in volume_config and volume_config["volumeType"] == "volatile":
return VolatileVolume(volume_config)
return None

@property
def tags(self) -> List[str]:
"""List of user-defined tags applied to the layer"""
return [str(t) for t in self.json.get("tags", [])]

@property
def billing_tag(self) -> Optional[List[str]]:
"""List of billing tags used for grouping billing records together for the layer"""
return [str(t) for t in self.json.get("billingTags", [])]

@property
def type(self) -> LayerType:
"""The type of the layer."""
return LayerType.from_str(self.json["layerType"])

@property
def content_type(self) -> Optional[str]:
"""The MIME type of the blobs stored in the layer, e.g. application/x-protobuf."""
return str(self.json["contentType"]) if "contentType" in self.json else None

@property
def content_encoding(self) -> Optional[str]:
"""The content transfer encoding used to transfer blobs, typically gzip or empty"""
return str(self.json["contentEncoding"]) if "contentEncoding" in self.json else None

@property
def created(self) -> datetime:
"""Timestamp, in ISO 8601 format, when the layer was initially created"""
return datetime.strptime(self.json["created"], "%Y-%m-%dT%H:%M:%S.%fZ")

@property
def schema(self) -> Optional[Dict[str, str]]:
"""
Describes a HRN for the layer schema.
Can be updated by the user for any kind of layer.
:return: Dict of schema or None
"""
return self.json.get("schema")

@property
def properties(self) -> Optional[Dict[str, Any]]:
"""
Returns additional properties depending on layer type.
:return: Dict of layer-specific properties or None if no extra properties are set.
"""
layer_type_properties = {"versioned": "versionedLayerProperties", "volatile": "volatileProperties", "stream": "streamProperties", "index": "indexProperties", "objectstore": "objectStoreProperties", "interactivemap": "interactiveMapProperties",}
this_layer_type_properties = layer_type_properties.get(self.json["layerType"])
return self.json.get(this_layer_type_properties) if this_layer_type_properties else None

[docs]
class Layer:
"""This base class provides access to data stored in catalog layers.

Instances can read their schemas for data stored in protobuf format,
all available partition IDs as well as the raw data blobs inside such
partitions. You have to use the :class:Schema class to access the
decoded protobuf data.
"""

def init(self, layer_id: str, catalog: "Catalog"):
"""Initialize layer instance.

:param layer_id: a string with the layer ID of this layer
:param catalog: the instance of the Catalog this layer belongs to
"""
self.id = layer_id
self.platform = catalog.platform
self.catalog: Catalog = catalog
self._base_api: BaseApi = catalog._base_api

lazy-loaded

self.__data_volatile_blob_api: Optional[DataVolatileBlobApi] = None
self.__data_blob_api: Optional[DataBlobApi] = None
self.__data_metadata_api: Optional[DataMetadataApi] = None
self.__data_publish_api: Optional[DataPublishApi] = None
self.__data_query_api: Optional[DataQueryApi] = None
self.__data_index_api: Optional[DataIndexApi] = None
self.__data_stream_api: Optional[DataStreamApi] = None
self.__data_config_api: Optional[DataConfigApi] = None
self.__data_ingest_api: Optional[DataIngestApi] = None
self.__data_interactive_api: Optional[DataInteractiveApi] = None
self.__data_object_blob_api: Optional[DataObjectBlobApi] = None
self.__data_statistics_api: Optional[DataStatisticsApi] = None

self._schema_registry = self.catalog.platform.schema_registry
self._adapter: Adapter = catalog.adapter
self._default_adapter: Adapter = catalog.platform.default_adapter
self.billing_tag = self.catalog.billing_tag # type: ignore

Pre-fetch this property so it doesn't need to be looked up for every read of data.

layer_properties = self.configuration.properties
self.has_storage_layer_access = (
layer_properties.get("encryptionType") == "cloudEncryption"
if layer_properties
else False
)

@property
def _data_volatile_blob_api(self) -> DataVolatileBlobApi:
"""
Lazy loads DataVolatileBlobApi API.

:return: DataVolatileBlobApi instance
"""
if self.__data_volatile_blob_api is None:
self.__data_volatile_blob_api = self.catalog._data_volatile_blob_api
return self.__data_volatile_blob_api

@property
def _data_blob_api(self) -> DataBlobApi:
"""
Lazy loads DataBlobApi API.

:return: DataBlobApi instance
"""
if self.__data_blob_api is None:
self.__data_blob_api = self.catalog._data_blob_api
return self.__data_blob_api

@property
def _data_metadata_api(self) -> DataMetadataApi:
"""
Lazy loads DataMetadataApi API.

:return: DataMetadataApi instance
"""
if self.__data_metadata_api is None:
self.__data_metadata_api = self.catalog._data_metadata_api
return self.__data_metadata_api

@property
def _data_publish_api(self) -> DataPublishApi:
"""
Lazy loads DataPublishApi API.

:return: DataPublishApi instance
"""
if self.__data_publish_api is None:
self.__data_publish_api = self.catalog._data_publish_api
return self.__data_publish_api

@property
def _data_query_api(self) -> DataQueryApi:
"""
Lazy loads DataQueryApi API.

:return: DataQueryApi instance
"""
if self.__data_query_api is None:
self.__data_query_api = self.catalog._data_query_api
return self.__data_query_api

@property
def _data_index_api(self) -> DataIndexApi:
"""
Lazy loads DataIndexApi API.

:return: DataIndexApi instance
"""
if self.__data_index_api is None:
self.__data_index_api = self.catalog._data_index_api
return self.__data_index_api

@property
def _data_stream_api(self) -> DataStreamApi:
"""
Lazy loads DataStreamApi API.

:return: DataStreamApi instance
"""
if self.__data_stream_api is None:
self.__data_stream_api = self.catalog._data_stream_api
return self.__data_stream_api

@property
def _data_config_api(self) -> DataConfigApi:
"""
Lazy loads DataConfigApi API.

:return: DataConfigApi instance
"""
if self.__data_config_api is None:
self.__data_config_api = self.catalog._data_config_api
return self.__data_config_api

@property
def _data_ingest_api(self) -> DataIngestApi:
"""
Lazy loads DataIngest API.

This API is not implemented on the Local Data Service.

:return: DataIngestApi instance
"""
if self.__data_ingest_api is None:
self.__data_ingest_api = self.catalog._data_ingest_api
return self.__data_ingest_api

@property
def _data_interactive_api(self) -> DataInteractiveApi:
"""
Lazy loads DataInteractive API.

This API is not implemented on the Local Data Service.

:return: DataInteractiveApi instance
"""
if self.__data_interactive_api is None:
self.__data_interactive_api = self.catalog._data_interactive_api
return self.__data_interactive_api

@property
def _data_object_blob_api(self) -> DataObjectBlobApi:
"""
Lazy loads DataInteractive API.

This API is not implemented on the Local Data Service.

:return: DataObjectBlobApi instance
"""
if self.__data_object_blob_api is None:
self.__data_object_blob_api = self.catalog._data_object_blob_api
return self.__data_object_blob_api

@property
def _data_statistics_api(self) -> DataStatisticsApi:
"""
Lazy loads DataStatistics API.

This API is not implemented on the Local Data Service.

:return: DataStatisticsApi instance
"""
if self.__data_statistics_api is None:
self.__data_statistics_api = self.catalog._data_statistics_api
return self.__data_statistics_api

[docs]
def is_versioned(self) -> bool:
"""
Check if this is a versioned layer.

:return: True if this is a versioned layer otherwise False
"""
return self.configuration.type == LayerType.VERSIONED

[docs]
def is_volatile(self) -> bool:
"""
Check if this is a volatile layer.

:return: True if this is a volatile layer otherwise False
"""
return self.configuration.type == LayerType.VOLATILE

[docs]
def is_stream(self) -> bool:
"""
Check if this is a stream layer.

:return: True if this is a stream layer otherwise False
"""
return self.configuration.type == LayerType.STREAM

[docs]
def is_index(self) -> bool:
"""
Check if this is an index layer.

:return: True if this is an index layer otherwise False
"""
return self.configuration.type == LayerType.INDEX

[docs]
def is_interactivemap(self) -> bool:
"""
Check if this is an interactive map layer.

:return: True if this is an interactive map layer otherwise False
"""
return self.configuration.type == LayerType.INTERACTIVEMAP

[docs]
def is_objectstore(self) -> bool:
"""
Check if this is an objectstore layer.

:return: True if this is an objectstore layer otherwise False
"""
return self.configuration.type == LayerType.OBJECTSTORE

def _verify_schema_exists(self):
"""
Verify the layer has a schema (and it can be obtained)
as precondition before trying to decode partition.

:raises ValueError: in case the layer has no schema
"""
if not self.get_schema():
raise ValueError(f"Schema not found for layer {self.id}")

def _verify_adapter_encoder(self, adapter: Adapter):
"""
Verify if an :class:Adapter can encode the layer content type.

:param adapter: the adapter to test
:raises LayerConfigurationException: in case encoding is requested but the
layer doesn't have any content type configured
:raises UnsupportedContentTypeEncodeException: in case encoding is requested but
the adapter does not support the content type of the layer requested
"""
content_type = self.configuration.content_type
if not content_type:
raise LayerConfigurationException(f"Content type required to encode layer {self.id}")
elif content_type not in adapter.encoder.supported_content_types:
raise UnsupportedContentTypeEncodeException(content_type=content_type)

def _verify_adapter_decoder(self, adapter: Adapter):
"""
Verify if an :class:Adapter can decode the layer content type.

:param adapter: the adapter to test
:raises LayerConfigurationException: in case decoding is requested but the
layer doesn't have any content type configured
:raises UnsupportedContentTypeDecodeException: in case decoding is requested but
the adapter does not support the content type of the layer requested
"""
content_type = self.configuration.content_type
if not content_type:
raise LayerConfigurationException(f"Content type required to decode layer {self.id}")
elif content_type not in adapter.decoder.supported_content_types:
raise UnsupportedContentTypeDecodeException(content_type=content_type)

def _validate_publication(self, publication: "Publication"):
if self.id not in [layer.id for layer in publication.layers]:
raise ValueError(f"Layer {self.id} not included in publication") # noqa: E713
if not publication.is_active:
raise ValueError("Publication is not active")

@property
def configuration(self) -> LayerConfiguration:
"""The configuration of the layer"""
self.catalog._ensure_catalog_config_loaded()
assert (
self.catalog._layer_configuration is not None
and self.id in self.catalog._layer_configuration
)
return self.catalog._layer_configuration[self.id]

[docs]
def get_details(self) -> Dict[str, Any]:
"""
Get layer details from the platform.

:return: a dictionary with the layer details
"""
return self.configuration.json

[docs]
def has_schema(self) -> bool:
"""
Check whether the layer has a schema defined.
This does not obtain and register the schema.

:return: whether the layer has a schema
"""
return "schema" in self.configuration.json

[docs]
def get_schema(self) -> Optional[Schema]:
"""
Return the schema of the layer, if available.

This allows for parsing the partition data.
It only works for layers which define a protobuf schema.

:return: a Schema instance
"""
if not self.has_schema():
return None

The schema is defined, we expect to find it on the artifact service

schema_hrn = self.configuration.json["schema"]["hrn"]
return self._schema_registry.obtain_and_register_schema(
schema_hrn, self.configuration.content_type
)

[docs]
def open_in_portal(self):
"""Open the layer page on the HERE platform portal."""
portal_url = self.platform.platform_config.portal_url
if portal_url is None:
raise ValueError("here_platform_portal_url is not present in configuration.")
webbrowser.open_new(f"{portal_url}/data/{self.catalog.hrn}/{self.id}")

def _has_gz_encoding(self) -> bool:
"""
Return whether the data of the layer has to be encoded/decoded
using the gzip algorithm when transmitted/received.

This doesn't affect the content type of the layer
and is always transparent to the user.
If, for example, the content type is JSON, then uncompressed JSON
is always accepted/returned from/to the user.
This property just affects the encoding
of the transmission to/from the platform services.

If the content type is zip or gzip, it's convenient for the user not to
configure the layer encoding to gzip as this would determine a second
compression/decompression of the content. User content is always
accepted/returned as they passed it. In no case zip or gzip files have
to be compressed/decompressed on behalf of the user.

:return: whether the data must be encoded/decoded in gzip before/after transmission
"""

return self.configuration.content_encoding == "gzip"

def _validate_write_part_size(self, part_size: int):
"""
Validate part size for write.
"""
if part_size < 5 or part_size > 50:
raise ValueError(
f"Invalid part_size provided: {part_size}. It must be between 5MB and 50MB."
)

def _validate_write_data_size(self, path_or_data: Union[str, Path, bytes], size_limit: int):
"""
Validate data size for write.

:param path_or_data: file path or data to write to versioned layers.
:param size_limit: data size to validate against.
:return: data size
:raises ValueError: data validation failed.
"""
if isinstance(path_or_data, bytes):
data_size = len(path_or_data)
if data_size > size_limit:
readable_size = get_readable_bytes(size_limit)
raise ValueError(
f"Data bytes size is greater than the permissible limit of {readable_size}"
)
return data_size
else:
if not Path(path_or_data).is_file():
raise ValueError(f"File: {path_or_data} does not exist.")

data_size = os.path.getsize(path_or_data)
if data_size > size_limit:
readable_size = get_readable_bytes(size_limit)
raise ValueError(
f"File: {path_or_data} size is greater than the permissible limit of"
f" {readable_size}"
)
return data_size

def _get_checksum_crc(self, data: Union[str, Path, bytes]):
layer_details = self.configuration.json
cksum = None
data_crc = None
if "digest" in layer_details:
cksum = checksum(
path_or_data=data,
hash_algo=layer_details["digest"].lower(),
)
if "crc" in layer_details:
data_crc = get_crc(data)
return data_crc, cksum

def _upload_blob(
self,
partition: Partition,
path_or_data: Union[str, bytes, Path],
part_size: int = 50,
single_part_upload_limit: int = 52428800,
) -> "Partition":
"""
Verify gzip to compress the data and upload blob.

:param partition: partition object referencing the uploaded data.
:param path_or_data: content to be uploaded, it must match the layer content type, if set.
:param part_size: An int representing size in MB, to upload in multiple parts
minimum value is 5MB and and Maximum is 50MB.
:param single_part_upload_limit: the max. number of bytes for single part data uploads,
data size greater than this limit will be uploaded in chunks. Default is 50mb.
:return: partition object referencing the uploaded data
:raises ValueError: Unsupported content encoding.
"""
if self.configuration.content_encoding not in ("gzip", "identity", None):
raise ValueError(
f"Unsupported content encoding: {self.configuration.content_encoding}"
)
if self._has_gz_encoding():

This part needs to be improved because it will

load the complete file data bytes in memory to compress.

if isinstance(path_or_data, bytes):
path_or_data = gzip.compress(path_or_data)
else:
if not Path(path_or_data).is_file():
raise ValueError(f"File: {path_or_data} does not exist.")
with open(path_or_data, "rb") as file_data:
file_data_read = file_data.read()
path_or_data = gzip.compress(file_data_read)
if (
self.configuration.type == LayerType.VOLATILE
or self.configuration.type == LayerType.VERSIONED
):
assert isinstance(partition, (VersionedPartition, VolatilePartition))
partition.compressed_data_size = len(path_or_data)

if self.configuration.type == LayerType.VOLATILE:
partition.upload_volatile_blob(path_or_data=path_or_data)
else:
if isinstance(path_or_data, bytes):
data_size = len(path_or_data)
else:
data_size = os.path.getsize(path_or_data)
if data_size <= single_part_upload_limit:
partition.put_blob(path_or_data=path_or_data)
else:
partition.multipart_upload(
path_or_data=path_or_data,
part_size=part_size,
)
return partition

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

: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.
"""
return self._data_blob_api.check_handle_exists(
layer_id=self.id, data_handle=data_handle, billing_tag=billing_tag or self.billing_tag
)

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

: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
: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.
:return: a blob response as bytes or iterator of bytes if stream is True
"""
return self._data_blob_api.get_blob_by_handle(
layer_id=self.id,
data_handle=data_handle,
range_header=range_header,
billing_tag=billing_tag or self.billing_tag,
stream=stream,
chunk_size=chunk_size,
storage_layer_access=self.has_storage_layer_access,
)

def _delete_blob(self, data_handle: str, billing_tag: Optional[str] = None) -> bool:
"""
Wrapper for data_blob_api delete_blob_by_handle method
Delete blob (raw bytes) for given layer ID and data-handle from storage.

: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 flag true on successful delete
"""
delete_handle = self._data_blob_api.delete_blob_by_handle(
layer_id=self.id, data_handle=data_handle, billing_tag=billing_tag or self.billing_tag
)
return True if delete_handle is None else False

def _put_blob(
self,
path_or_data: Union[str, bytes, Path],
publication: Optional["Publication"] = None,
partition_id: Optional[str] = None,
data_handle: Optional[str] = None,
part_size: int = 50,
fields: Dict[str, Union[str, int, bool]] = ,
additional_metadata: Dict[str, str] = ,
timestamp: Optional[int] = None,
inline_stream_data_limit: Optional[int] = DEFAULT_INLINE_STREAM_DATA_LIMIT,
) -> Partition:
"""
Wrapper for data_blob_api publish_blob_by_handle method
Validate data, generate partition object with metadata,
upload a blob to the storage for a layer, and return a Partition object
referencing to it via its data_handle, populated with its metadata.

:param path_or_data: content to be uploaded, it must match the layer content type, if set.
:param publication: the publication this operation is part of
:param partition_id: partition identifier the blob relates to for stream, versioned
and volatile layers.
:param data_handle: data handle to use for the blob, in case already available,
if not available an appropriate one is generated and returned.
:param part_size: An int representing size in MB, to upload in multiple parts
minimum value is 5MB and maximum is 50MB.
:param fields: A dict representing the fields of index record for
data being uploaded for index layer only.
:param additional_metadata: A dict of additional metadata about data being
uploaded for index layer only.
:param timestamp: timestamp, in milliseconds since Unix epoch (1970-01-01T00:00:00 UTC)
:param inline_stream_data_limit: threshold data size in bytes to decide if
inline stream data field should be populated, if data size
is less than the inline_data_limit then the data would
be added to StreamPartition.data field or else blob would
be uploaded and its data_handle will be added to
StreamPartition.data_handle field.
:return: partition object referencing the uploaded data
:raises ValueError: Publication is not active or Layer not found in the publication.
"""
publication_layer_ids = []
if not self.configuration.type == LayerType.INDEX and publication:
publication_layer_ids = [x.id for x in publication.layers]
if (
self.configuration.type == LayerType.INDEX
or self.configuration.type == LayerType.STREAM
or (
publication.is_active and self.id in publication_layer_ids
if publication
else False
)
):
if not data_handle:
data_handle = Partition.generate_data_handle()
if self.configuration.type == LayerType.INDEX:
assert isinstance(self, IndexLayer)
data_size = self._validate_write_layer(
part_size=part_size, path_or_data=path_or_data, fields=fields
)
partition_id = data_handle
elif self.configuration.type == LayerType.VOLATILE:
assert isinstance(self, VolatileLayer)
data_size_list = self._validate_write_layer(path_or_data_list=[path_or_data])
data_size = data_size_list[0]
else:
assert isinstance(self, (StreamLayer, VersionedLayer))
data_size_list = self._validate_write_layer(
part_size=part_size, path_or_data_list=[path_or_data]
)
data_size = data_size_list[0]
data_crc, cksum = self._get_checksum_crc(data=path_or_data)
metadata_dict = {"id": partition_id, "data_size": data_size, "data_handle": data_handle, "checksum": cksum, "crc": data_crc, "timestamp": timestamp, "fields": fields, "additional_metadata": additional_metadata,}
if isinstance(self, StreamLayer):
assert isinstance(inline_stream_data_limit, int)
inline_data = self._read_and_validate_inline_data(
path_or_data=path_or_data, inline_data_limit=inline_stream_data_limit
)
if inline_data:
metadata_dict["data"] = inline_data
metadata_dict["data_handle"] = None
partition = self._create_partition(**metadata_dict)
if isinstance(partition, StreamPartition) and partition.data:
return partition
partition = cast(
Union[IndexPartition, VolatilePartition, StreamPartition, VersionedPartition],
self._upload_blob(
partition=partition,
path_or_data=path_or_data,
part_size=part_size,
),
)
logger.debug(
f"Multipart Upload completed successfully for layer_id: {self.id} "
f"partition_id: {partition_id}"
)
return partition
else:
if publication and not publication.is_active:
raise ValueError("Publication is not active.")
else:
raise ValueError("Layer not found in the publication.")

[docs]
class IndexLayer(Layer):
"""
This class provides access to data stored in index layers.
"""

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

: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
: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.
:return: a blob response as bytes or iterator of bytes if stream is True
"""
return self._get_blob(
data_handle,
range_header,
billing_tag=billing_tag or self.billing_tag,
stream=stream,
chunk_size=chunk_size,
)

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

: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.
"""
return self._blob_exists(data_handle, billing_tag=billing_tag or self.billing_tag)

[docs]
def put_blob(
self,
path_or_data: Union[str, bytes, Path],
publication: Optional["Publication"] = None,
partition_id: Optional[str] = None,
data_handle: Optional[str] = None,
part_size: int = 50,
fields: Dict[str, Union[str, int, bool]] = ,
additional_metadata: Dict[str, str] = ,
timestamp: Optional[int] = None,
) -> Partition:
"""
Upload a blob to the durable blob service.

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

: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 flag true on successful delete
"""
return self._delete_blob(data_handle, billing_tag=billing_tag or self.billing_tag)

def _validate_write_layer(
self,
part_size: int,
path_or_data: Union[str, Path, bytes],
fields: Dict[str, Union[str, int, bool]] = ,
):
"""
Validate parameters for write versioned layers.

:param part_size: An int representing size in MB, to upload in multiple parts minimum
value is 5MB and maximum is 50MB.
:param path_or_data: file path or data to write to versioned layers.
:param fields: a dict representing the fields of index record for data being uploaded
:return: data size
:raises ValueError: index layer write data validation failed.

"""
layer_info: dict = cast(dict, self.catalog._layer_details)[self.id]
self._validate_write_part_size(part_size=part_size)
data_size = self._validate_write_data_size(
path_or_data=path_or_data, size_limit=50 1024 1024 * 1024
)
if fields:
if len(fields) > 4:
raise ValueError(
"Index attributes can not be more than four. "
"Please check index attributes in layer configuration."
)
layer_index_attrs = layer_info["indexProperties"]["indexDefinitions"]
layer_attrs_set = set()
index_attrs_set = set(fields.keys())
time_window_attr = ""
for attr in layer_index_attrs:
if attr["type"] == "timewindow":
time_window_attr = attr["name"]
layer_attrs_set.add(attr["name"])

if time_window_attr not in index_attrs_set:
raise ValueError(
f"TimeWindow attribute: {time_window_attr} is mandatory."
f"It is not provided in: {fields}"
)

if index_attrs_set.intersection(layer_attrs_set) != index_attrs_set:
raise ValueError(
f"There is mismatch in index attributes defined in layer: "
f"{layer_attrs_set} and input index attributes: {index_attrs_set}"
)
return data_size

def _create_partition(self, **kwargs) -> IndexPartition:
fields = cast(Dict[str, Union[str, int, bool]], kwargs.get("fields"))
additional_metadata = cast(Dict[str, str], kwargs.get("additional_metadata"))

TODO: remove the cast once the decorator of the IndexPartition init is gone

return cast(
IndexPartition,
IndexPartition(
data_handle=kwargs.get("data_handle"),
layer=self,
data_size=kwargs.get("data_size"),
checksum=kwargs.get("checksum"),
crc=kwargs.get("crc"),
timestamp=kwargs.get("timestamp"),
fields=fields,
additional_metadata=additional_metadata,
),
)

[docs]
def get_partitions_metadata(
self,
query: str,
adapter: Optional[Adapter] = None,
part: Optional[str] = None,
billing_tag: Optional[str] = None,
**kwargs,
) -> Union[Iterator[IndexPartition], "pd.DataFrame"]:
"""Get list of all partitions matching the query.

The query must be in RSQL format, see also: https://github.com/jirutka/rsql-parser.

:param query: the RSQL query
:param adapter: the Adapter to transform and return the result.
None to use the default adapter of the catalog.
:param part: Indicates which part of the layer shall be queried.
:param billing_tag: A string which is used for grouping billing records.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: Iterator of :class:IndexPartition objects, or adapter-specific
"""
adapter = adapter or self._adapter

meta_it = self._get_partitions_list(query, part, billing_tag)

return adapter.from_index_metadata(meta_it, **kwargs)

def _get_partitions_list(
self,
query: str,
part: Optional[str] = None,
billing_tag: Optional[str] = None,
) -> Generator[IndexPartition, None, None]:
"""Get list of all partition objects matching the query.

The query must be in RSQL format, see also: https://github.com/jirutka/rsql-parser.

:param query: the RSQL query
:param part: Indicates which part of the layer shall be queried.
:param billing_tag: A string which is used for grouping billing records.
:yield: :class:Partition objects
"""
partition_info = self._data_index_api.perform_query(
self.id, query=query, part=part, billing_tag=billing_tag or self.billing_tag
)
partition_metadata = partition_info["data"]
for p in partition_metadata:
partition_obj = IndexPartition(
data_handle=p["id"],
layer=self,
checksum=p.get("checksum"),
data_size=p.get("size"),
crc=p.get("crc"),
timestamp=p.get("timestamp"),
fields={

TODO: this modifies the field names set by the users,

not sure this is what we want

x: p[x]
for x in p
if x
not in timestamp
},
additional_metadata=literal_eval(p["metadata"]) if "metadata" in p else None,
)
yield partition_obj

[docs]
def read_partitions(
self,
query: str,
decode: bool = True,
adapter: Optional[Adapter] = None,
part: Optional[str] = None,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
**kwargs,
) -> Union[
Iterator[Tuple[IndexPartition, bytes]], # raw
Iterator[Tuple[IndexPartition, Iterator[bytes]]], # stream
Iterator[Tuple[IndexPartition, Any]], # from default adapter
"pd.DataFrame", # from geopandas adapter
]:
"""
Read of all partition data matching the query.

The query must be in RSQL format, see also: https://github.com/jirutka/rsql-parser.

:param query: the RSQL query
:param decode: whether to decode the data through an adapter or return raw bytes
:param adapter: the Adapter to transform and return the result.
None to use the default adapter of the catalog.
:param part: Indicates which part of the layer shall be queried.
:param stream: whether to stream data. This implies decode=False.
:param chunk_size: the size to request each iteration when streaming data.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: Iterator of :class:IndexPartition objects each with its raw data,
in case decode=False, adapter-specific otherwise
:raises ValueError: in case decoding is requested but the adapter does not support
the content type of the layer requested, or invalid parameters # noqa
:raises LayerConfigurationException: in case decoding is requested but the
layer doesn't have any content type configured # noqa
"""
adapter = adapter or self._adapter

if decode and not stream:
self._verify_adapter_decoder(adapter)

meta_it = self.get_partitions_metadata(query, adapter=self._default_adapter, part=part)

partitions_data = map(
lambda p: (p, p.get_blob(stream=stream, chunk_size=chunk_size)), meta_it
)

if decode and not stream:
content_type = self.configuration.content_type
assert content_type
schema = self.get_schema()
return adapter.from_index_data(partitions_data, content_type, schema, **kwargs)
else:
return partitions_data

[docs]
def set_partitions_metadata(
self,
update: Optional[Iterable[IndexPartition]] = None,
delete: Optional[Iterable[str]] = None,
):
"""
Update the metadata of the layer as part of a publication
by publishing updated partitions and/or deleting partitions.

:param update: the complete partitions to update.
:param delete: the data handles to delete.
"""
if update or delete:
update = update or []
delete = delete or []
update_partition_metadata = []
assert all(x.layer.id == self.id for x in update)
for partition in update:
update_partition_metadata.append(
{"checksum": partition.checksum, "crc": partition.crc, "fields": partition.fields, "id": partition.data_handle, "metadata": partition.additional_metadata, "size": partition.data_size,}
)

def split_metadata(chunk_size, update_metadata, delete_metadata):
update_iter = iter(update_metadata)
update_piece = list(islice(update_iter, chunk_size))
delete_iter = iter(delete_metadata)
delete_piece = list(islice(delete_iter, chunk_size))
while update_piece or delete_piece:
yield (update_piece, delete_piece)
update_piece = list(islice(update_iter, chunk_size))
delete_piece = list(islice(delete_iter, chunk_size))

for update_chunk, delete_chunk in split_metadata(
1000, update_partition_metadata, delete
):
body = {"additions": update_chunk, "removals": delete_chunk}
self._data_index_api.perform_update(
layer_id=self.id,
body=body,
billing_tag=self.billing_tag,
)

[docs]
def write_single_partition(
self,
data: Union[str, Path, bytes, "pd.DataFrame"], # encode / raw
timestamp: Optional[int] = None,
fields: Dict[str, Union[str, int, bool]] = ,
additional_metadata: Dict[str, str] = ,
part_size: int = 50,
encode: bool = True,
adapter: Optional[Adapter] = None,
**kwargs,
):
"""
Upload content to the layer and publish the related partition metadata.

:param data: data to upload to the layer and derive metadata from.
:param timestamp: timestamp
:param fields: a dict representing the fields of index record for data being uploaded
:param additional_metadata: a dict of additional metadata about data being uploaded
:param part_size: An int representing size in MB, to upload in multiple parts
minimum value is 5MB and and Maximum is 50MB.
:param encode: whether to encode the data through an adapter or store raw bytes
:param adapter: the Adapter to transform the input data.
None to use the default adapter of the catalog.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
"""

adapter = adapter or self._adapter

if encode:
self._verify_adapter_encoder(adapter)
content_type = self.configuration.content_type
assert content_type
schema = self.get_schema()
encoded = adapter.to_index_single_data(data, content_type, schema, **kwargs)
assert isinstance(encoded, bytes)
data = encoded

self._validate_write_layer(part_size=part_size, path_or_data=data, fields=fields)
index_partition = self._put_blob(
path_or_data=data,
part_size=int(part_size),
timestamp=timestamp,
fields=fields,
additional_metadata=additional_metadata,
)
assert isinstance(index_partition, IndexPartition)

Why isn't it possible to use set_partitions_metadata?

index_data: Dict[str, Any] =
layer_details = self.configuration.json
if "digest" in layer_details:
index_data["checksum"] = index_partition.checksum
if "crc" in layer_details:
index_data["crc"] = index_partition.crc
index_data["fields"] = index_partition.fields
index_data["metadata"] = index_partition.additional_metadata
index_data["size"] = index_partition.data_size
index_data["id"] = index_partition.data_handle
self._data_index_api.insert_indexes(
layer_id=self.id, body=[index_data], billing_tag=self.billing_tag
)

[docs]
def delete_partitions(self, query: str):
"""
Delete the partitions that match the query in an index layer.

The query must be in RSQL format, see
also: https://github.com/jirutka/rsql-parser.

:param query: A string representing a RSQL query.
:return : true when delete partitions succeeds.
:raises ValueError: delete partitions failed.
"""
del_resp = self._data_index_api.perform_delete(layer_id=self.id, delete_query=query)
del_id = del_resp["deleteId"]
logger.debug(f"Delete RequestId generated: {del_id}")

while True:
logger.debug(f"status polling wait {self.catalog.platform._polling_wait} sec.")
time.sleep(self.catalog.platform._polling_wait)
del_status = self._data_index_api.get_delete_request_status(
layer_id=self.id, delete_id=del_id
)
status = del_status["state"]
logger.debug(f"partition delete status: {status}")
if status == "Succeeded":
logger.info(
f"partitions deleted successfully for the query: {query}, "
f"number of index records deleted: {del_status['count']}"
)
return True
elif status == "Failed":
raise ValueError(f"Delete partitions failed: {del_status['message']}")

[docs]
def get_parts(self, num_requested_parts: int = 1, billing_tag: Optional[str] = None) -> dict:
"""
Return a list of Part Ids which represent the layer parts that can be used to limit the
scope of a query operation. 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 num_requested_parts: Indicates requested number of layer parts.
:param billing_tag: A string which is used for grouping billing records.
:return: dict of parts as per num_requested_parts.
"""

parts = self._data_index_api.get_parts(
layer_id=self.id,
num_requested_parts=num_requested_parts,
billing_tag=billing_tag or self.billing_tag,
)

return parts

[docs]
class StreamSubscription:
"""
Represent a subscription to consume data from a stream layer.

The subscription must be closed by unsubscribing to free resources on the service.
"""

[docs]
class Mode(enum.Enum):
"""
Mode of a stream subscription.
"""

SERIAL = "serial"
PARALLEL = "parallel"

def init(
self,
layer: "StreamLayer",
sub_id: str,
sub_mode: Mode,
node_base_url: str,
):
"""
Initialize a new subscription to a stream layer.

:param layer: layer related to the subscription
:param sub_id: identifier of the subscription, as returned by the service
:param sub_mode: mode of the subscription
:param node_base_url: base URL of the stream processing node for subsequent API calls
"""
self._layer = layer
self._active = True
self._node_base_url = node_base_url
self.id = sub_id
self.mode = sub_mode

def enter(self):
"""
Entry point for stream subscription context manager
"""
return self

def exit(self, exc_type, exc_value, exc_tb):
"""
Exit point for stream subscription context manager
"""
try:
self.unsubscribe()
except Exception as e:
logger.error(f"Error occured when performing unsubscribe: {e}")

[docs]
def unsubscribe(self, strict: bool = False):
"""
Disable message consumption for this layer.

After unsubscribing, you need to subscribe to the stream layer again
to be able to resume the data consumption.

:param strict: True to require that the subscription exists, False to allow it to have
already been cancelled.
"""
if self._active:
self._layer._data_stream_api.delete_subscription(
node_base_url=self._node_base_url,
layer_id=self._layer.id,
subscription_id=self.id,
mode=self.mode.value,
strict=strict,
)
self._active = False

[docs]
def seek_to_offsets(self, offsets: Dict[int, int]):
"""
Seek to stream offsets for a stream layer subscription. It will
start reading data from a specified offsets.

:param offsets: Dict of offset {<Partition ID>:<Offset Number>, <Partition ID>:<Offset Number>}. Partition id is kafka partition id.
"""
if self._active and offsets:
list_offsets = [{"partition": k, "offset": v} for k, v in offsets.items()]
self._layer._data_stream_api.seek_to_offset(
node_base_url=self._node_base_url,
layer_id=self._layer.id,
subscription_id=self.id,
offsets=list_offsets,
mode=self.mode.value,
)

[docs]
def commit_offsets(self, offsets: Dict[int, int]):
"""
Commit specified offsets once read is done.

:param offsets: Dict of offset {<Partition ID>:<Offset Number>, <Partition ID>:<Offset Number>}. Partition id is kafka partition id.
"""
if self._active and offsets:
list_offsets = [{"partition": k, "offset": v} for k, v in offsets.items()]
self._layer._data_stream_api.commit_offsets(
node_base_url=self._node_base_url,
layer_id=self._layer.id,
subscription_id=self.id,
offsets=list_offsets,
mode=self.mode.value,
)

[docs]
class StreamIngestion(JsonDictDocument):
"""
Response of the stream layer to confirm successful data ingestion.
"""

@property
def message_list_id(self) -> str:
"""The identifier assigned to the ingested SDII message list."""
return str(self.json["traceId"]["parentId"])

@property
def message_ids(self) -> List[str]:
"""The identifiers assigned to each SDII message ingested."""
return [str(msg_id) for msg_id in self.json["traceId"]["generatedIds"]]

[docs]
class StreamLayer(Layer):
"""
This class provides access to data stored in stream layers.
"""

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

: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.
"""
return self._blob_exists(data_handle, billing_tag=billing_tag or self.billing_tag)

[docs]
def put_blob(
self,
path_or_data: Union[str, bytes, Path],
partition_id: Optional[str] = None,
data_handle: Optional[str] = None,
inline_stream_data_limit: Optional[int] = DEFAULT_INLINE_STREAM_DATA_LIMIT,
) -> Partition:
"""
Upload a blob to the durable blob service.

:param path_or_data: content to be uploaded, it must match the layer content type, if set.
:param partition_id: partition identifier the blob relates to.
:param data_handle: data handle to use for the blob, in case already available,
if not available an appropriate one is generated and returned.
:param inline_stream_data_limit: threshold data size in bytes to decide if
inline stream data field should be populated, if data size
is less than the inline_data_limit then the data would
be added to StreamPartition.data field or else blob would
be uploaded and its data_handle will be added to
StreamPartition.data_handle field.
:return: partition object referencing the uploaded data
"""
return self._put_blob(
path_or_data,
partition_id=partition_id,
data_handle=data_handle,
inline_stream_data_limit=inline_stream_data_limit,
)

def _validate_write_layer(
self, part_size: int, path_or_data_list: List[Union[str, Path, bytes]]
):
"""
Validate parameters for write stream layers.

:param part_size: An int representing size in MB, to upload in multiple parts minimum
value is 5MB and maximum is 50MB.
:param path_or_data_list: List of file paths or data to write to stream layers.
:return: list of data size
"""
self._validate_write_part_size(part_size=part_size)
data_size_list = []
for path_or_data in path_or_data_list:
data_size = self._validate_write_data_size(
path_or_data=path_or_data, size_limit=5 1024 1024 * 1024
)
data_size_list.append(data_size)
return data_size_list

def _read_and_validate_inline_data(
self,
path_or_data: Union[str, Path, bytes],
inline_data_limit: int,
) -> Optional[bytes]:
if isinstance(path_or_data, bytes):
data = path_or_data
else:
if not Path(path_or_data).is_file():
raise ValueError(f"File: {path_or_data} does not exist.")
data_size = os.path.getsize(path_or_data)
if data_size < (10 * inline_data_limit):
with open(path_or_data, "rb") as file_data:
data = file_data.read()
else:
return None
if len(data) <= inline_data_limit:
return data
elif (
self._has_gz_encoding()
and len(data) < (10 * inline_data_limit)
and len(gzip.compress(data)) < inline_data_limit
):
return data
else:
return None

def _create_partition(self, **kwargs) -> StreamPartition:
return StreamPartition(
id=kwargs.get("id"),
data_handle=kwargs.get("data_handle"),
data=kwargs.get("data"),
layer=self,
data_size=kwargs.get("data_size"),
checksum=kwargs.get("checksum"),
crc=kwargs.get("crc"),
timestamp=kwargs.get("timestamp"),
)

def _verify_subscription(self, subscription: StreamSubscription):
if not (
subscription._layer.id == self.id
and subscription._layer.catalog.hrn == self.catalog.hrn
):
raise ValueError("Subscription unrelated with the StreamLayer")
if not subscription._active:
raise ValueError("Subscription is not active")

[docs]
def subscribe(
self,
mode: StreamSubscription.Mode = StreamSubscription.Mode.SERIAL,
consumer_id: Optional[str] = None,
kafka_consumer_properties: Optional[dict] = None,
group_id: Optional[str] = None,
auto_offset_reset: Optional[str] = None,
subscription_id: Optional[str] = None,
) -> StreamSubscription:
"""
Enable message consumption for this layer.

:param mode: The subscription mode for this subscription. By default value is serial.
:param consumer_id: The Id to use to identify this consumer.
It must be unique within the consumer group. If you do not provide one,
the system will generate one.
:param kafka_consumer_properties: Properties to configure the kafka consumer on the service
:param group_id: set the consumer group id
:param auto_offset_reset: to seek to some predefined locations in the stream.
earliest: automatically reset the offset to the earliest offset
latest: automatically reset the offset to the latest offset
none: the service will return an error if no previous offset is
found for the consumer's group
:param subscription_id: subscription id returned from a previous call to subscribe(). This
allows a previously created subscription (e.g. saved to persistent storage between
application runs) to be restored.

For other kafka consumer available settings, see
https://kafka.apache.org/documentation/#consumerconfigs.
:return: a new subscription to the stream layer
"""
prop = dict(kafka_consumer_properties) if kafka_consumer_properties else
if group_id and "group.id" not in prop:
prop["group.id"] = group_id
if auto_offset_reset and "auto.offset.reset" not in prop:
prop["auto.offset.reset"] = auto_offset_reset
kafka_prop = {"kafkaConsumerProperties": prop} or None

subscribe_resp: dict = self._data_stream_api.subscribe(
layer_id=self.id,
subscription_id=subscription_id,
mode=mode.value,
consumer_id=consumer_id,
kafka_consumer_properties=kafka_prop,
)
return StreamSubscription(
self,
subscribe_resp["subscriptionId"],
mode,
subscribe_resp["nodeBaseURL"],
)

[docs]
def get_stream_metadata(
self,
subscription: StreamSubscription,
commit_offsets: bool = True,
adapter: Optional[Adapter] = None,
**kwargs,
) -> Union[Iterator[StreamPartition], "pd.DataFrame"]:
"""
Consume metadata for a subscription.

It does not download blobs, use read_stream for that.

The function consumes and returns messages for the stream subscription.
The amount of messages retrieved depends on a variety of factors and it is not possible
to assume that all the available messages are returned with one single invocation:
users should invoke this function multiple times to read all the content present
in the stream, at least until some data is returned, if this is what is wanted.

When no more data is returned, users can reasonably assume the end of stream is reached.
However, when operating with a distributed, asynchronous messaging system like the one
employed in this case, producers can append new messages at any point in time and there
may be a delay between the moment when data is produced and the moment when data is
available for consumption.

While no message is lost, the end of stream can't always be detected reliably.

:param subscription: the subscription from where to consume the data
:param commit_offsets: automatically commit offsets so next read starts at the end of the
last consumed message.
:param adapter: the Adapter to transform and return the result.
None to use the default adapter of the catalog.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: Iterator of :class:StreamPartition objects, or adapter-specific
:raises ValueError: in case the subscription is invalid # noqa
"""
self._verify_subscription(subscription)

adapter = adapter or self._adapter

meta_it = self._get_stream_metadata_list(
subscription=subscription, commit_offsets=commit_offsets
)

return adapter.from_stream_metadata(meta_it, **kwargs)

def _get_stream_metadata_list(
self, subscription: StreamSubscription, commit_offsets: bool
) -> Generator[StreamPartition, None, None]:
"""
Consume data for a subscription.

It does not download blobs whose handle is given or parse any.
Use read_stream for that.

:param subscription: the subscription from where to consume the data
:param commit_offsets: automatically commit offsets so next read starts at the end of the
last consumed message.
:yield: :class:Partition objects.
:raises ValueError: in case the subscription is invalid # noqa
"""
self._verify_subscription(subscription)
res = self._data_stream_api.consume_data(
node_base_url=subscription._node_base_url,
layer_id=self.id,
subscription_id=subscription.id,
mode=subscription.mode.value,
)

if commit_offsets:
offsets = {m["offset"]["partition"]: m["offset"]["offset"] for m in res["messages"] if "offset" in m}
subscription.commit_offsets(offsets=offsets)

def obtain_data(data_field) -> bytes:
if self.configuration.content_encoding not in ("gzip", "identity", None):
raise ValueError(
f"Unsupported content encoding: {self.configuration.content_encoding}"
)
if self._has_gz_encoding():
return gzip.decompress(b64decode(data_field))
else:
return b64decode(data_field)

for m in res["messages"]:
offset = m.get("offset")
p = m["metaData"]

'or None' is needed to convert empty strings to None

partition_obj = StreamPartition(
data_handle=p.get("dataHandle") or None,
id=p.get("partition") or None,
layer=self,
checksum=p.get("checksum") or None,
data_size=p.get("dataSize") or None,
crc=p.get("crc") or None,
data=obtain_data(p["data"]) if ("data" in p and p["data"]) else None,
timestamp=p.get("timestamp") or None,
kafka_partition=offset["partition"] if offset else None,
kafka_offset=offset["offset"] if offset else None,
)
yield partition_obj

[docs]
def read_stream(
self,
subscription: StreamSubscription,
commit_offsets: bool = True,
decode: bool = True,
adapter: Optional[Adapter] = None,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
**kwargs,
) -> Union[
Iterator[Tuple[StreamPartition, bytes]], # raw
Iterator[Tuple[StreamPartition, Iterator[bytes]]], # stream
Iterator[Tuple[StreamPartition, Any]], # from default adapter
"pd.DataFrame", # from geopandas adapter
]:
"""
Consume data for this subscription. Download and decode the blobs.

While no message is lost, the end of stream can't always be detected reliably.

:param subscription: the subscription from where to consume the data
:param commit_offsets: automatically commit offset so next read starts at the end of the
last message.
:param decode: whether to decode the data through an adapter or return raw bytes
:param adapter: the Adapter to transform and return the result.
None to use the default adapter of the catalog.
:param stream: whether to stream data. This implies decode=False.
:param chunk_size: the size to request each iteration when streaming data.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: Iterator of :class:StreamPartition objects each with its raw data,
in case decode=False, adapter-specific otherwise
:raises ValueError: in case the subscription is invalid # noqa
:raises ValueError: in case decoding is requested but the adapter does not support
the content type of the layer requested, or invalid parameters
:raises LayerConfigurationException: in case decoding is requested but the
layer doesn't have any content type configured
"""
self._verify_subscription(subscription)

adapter = adapter or self._adapter

if decode and not stream:
self._verify_adapter_decoder(adapter)

meta_it = self.get_stream_metadata(
subscription, commit_offsets, adapter=self._default_adapter
)

partitions_data = map(
lambda p: (p, p.get_data(stream=stream, chunk_size=chunk_size)), meta_it
)

if decode and not stream:
content_type = self.configuration.content_type
assert content_type
schema = self.get_schema()
return adapter.from_stream_data(partitions_data, content_type, schema, **kwargs)
else:
return partitions_data

[docs]
def append_stream_metadata(
self,
partitions: Union[
Iterable[StreamPartition], # to default adapter
"pd.DataFrame", # to geopandas adapter
],
publication: Optional["Publication"] = None,
adapter: Optional[Adapter] = None,
**kwargs,
) -> None:
"""
Append new partition metadata to the stream layer directly as messages to the stream.

:param publication: the publication this operation is part of
:param partitions: the partitions to append as messages, or adapter-specific
:param adapter: the Adapter to transform the input.
None to use the default adapter of the catalog.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
"""

adapter = adapter or self._adapter

meta_it = adapter.to_stream_metadata(self, partitions, **kwargs)

self._append_stream_metadata(meta_it)

def _append_stream_metadata(
self,
partitions: Iterable[StreamPartition],
trace_id: Optional[str] = None,
) -> None:
update_partition_metadata = []
for partition in partitions:
if partition.data and partition.data_handle:
raise ValueError(
"data and data_handle fields cannot be populated "
f"at the same time for partition id: {partition.id}."
)
inline_data = None
if partition.data:
if self.configuration.content_encoding not in ("gzip", "identity", None):
raise ValueError(
"Unsupported content encoding: " f"{self.configuration.content_encoding}"
)
data = partition.data
if self._has_gz_encoding():
data = gzip.compress(data)
if len(data) >= 998400:
raise ValueError("Data exceeds the current limit of 1Mb.")
inline_data = b64encode(data).decode("utf-8")
update_partition_metadata.append(
{"dataHandle": partition.data_handle if not partition.data else None, "data": inline_data, "partition": partition.id, "checksum": partition.checksum, "dataSize": partition.data_size, "crc": partition.crc, "timestamp": partition.timestamp,}
)
if update_partition_metadata:
body = {"partitions": update_partition_metadata}
self._data_ingest_api.ingest_partitions(
layer_id=self.id,
partitions=body,
traceid_header=trace_id,
billing_tag=self.billing_tag,
)

[docs]
def write_stream(
self,
data: Union[
Iterable[ # raw
Union[
Tuple[Union[str, int], Union[str, Path, bytes]], # id, data
Tuple[Union[str, int], Union[str, Path, bytes], Optional[int]], # id, data, ts
]
],
Mapping[Union[str, int], Union[str, Path, bytes]], # id, data
Iterable[ # to default adapter
Union[
Tuple[Union[str, int], Any], # id, data
Tuple[Union[str, int], Any, Optional[int]], # id, data, ts
]
],
Mapping[Union[str, int], Any], # to default adapter: id, data
"pd.DataFrame", # to geopandas adapter
],
timestamp: Optional[int] = None,
encode: bool = True,
inline_data_limit: int = 819200,
adapter: Optional[Adapter] = None,
**kwargs,
):
"""
Write new content to the layer and push the related
partition metadata to the stream as part of a publication.

:param data: data to upload to the layer and derive metadata from:
a sequence of elements, each either (id, data) or (id, data, timestamp).
Timestamp is optional and in milliseconds since Unix epoch (1970-01-01T00:00:00 UTC)
:param encode: whether to encode the data or upload raw bytes
:param timestamp: optional timestamp for all the messages, if none is specified in data:
in milliseconds since Unix epoch (1970-01-01T00:00:00 UTC)
:param inline_data_limit: threshold data size in bytes to decide if
inline stream data field should be populated ,if data size
is less than the inline_data_limit then the data would
be added to StreamPartition.data field or else blob would
be uploaded and its data_handle will be added to
StreamPartition.data_handle field.
Default is 819200 bytes.
:param adapter: the Adapter to transform the input.
None to use the default adapter of the catalog.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
"""

There's a similar code in the default adapter but

there's also no clear way of how to remove this duplication

def normalize(
entry: Union[
Tuple[Union[str, int], Union[str, Path, bytes]], # id, data
Tuple[Union[str, int], Union[str, Path, bytes], Optional[int]], # id, data, ts
]
) -> Tuple[Union[str, int], Union[str, Path, bytes], Optional[int]]:
if len(entry) == 3:
id, data, ts = cast(
Tuple[Union[str, int], Union[str, Path, bytes], Optional[int]], entry
)
return id, data, (ts or timestamp)
elif len(entry) == 2:
id, data = cast(Tuple[Union[str, int], Union[str, Path, bytes]], entry)
return id, data, timestamp
else:
raise ValueError("Unexpected format for data")

adapter = adapter or self._adapter

Encode if needed

if encode:
adapter = adapter or self._adapter
self._verify_adapter_encoder(adapter)
content_type = self.configuration.content_type
assert content_type
schema = self.get_schema()
encoded_it: Iterator[
Tuple[Union[str, int], Union[str, Path, bytes], Optional[int]]
] = adapter.to_stream_data(self, data, content_type, schema, timestamp, **kwargs)
else:
encoded_it = map(normalize, iter_tuples(data))

partition_list: List[StreamPartition] = []
for partition_id, path_or_data, partition_ts in encoded_it:
self._validate_write_data_size(
path_or_data=path_or_data, size_limit=5 1024 1024 * 1024
)
pid = str(partition_id) # convert int to str
partition_metadata = self._put_blob(
path_or_data=path_or_data,
partition_id=pid,
timestamp=partition_ts,
inline_stream_data_limit=inline_data_limit,
)
assert isinstance(partition_metadata, StreamPartition)
partition_list.append(partition_metadata)

self._append_stream_metadata(partitions=partition_list)

def _kafka_endpoint(self, kafka_type: str) -> Dict:
"""
Returns kafka endpoint.

:param kafka_type: Identifies whether the properties returned in the response
are for a producer or consumer.
Valid values for this is 'producer' or 'consumer'.
:return: kafka endpoint based on kafka_type i.e. producer or consumer
:raises ValueError: Incorrect kafka type.
"""
endpoint =
if kafka_type not in ["producer", "consumer"]:
raise ValueError("Incorrect kafka type.")
else:
endpoint = self._data_stream_api.kafka_endpoint(
layer_id=self.id, kafka_type=kafka_type
)
return endpoint

[docs]
def get_kafka_topic(self) -> str:
"""
Returns topic for stream layer.
:return: topic
"""
details = self._data_stream_api.kafka_endpoint(layer_id=self.id, kafka_type="consumer")
return str(details["topic"])

[docs]
def kafka_consumer(
self,
group_id: Optional[str] = None,
**kwargs,
) -> KafkaConsumer:
"""
Instantiate and return a new KafkaConsumer pre-configured to operate with the layer.

:param group_id: identifies the consumer group this consumer belongs to.
:param kwargs: Kafka consumer properties. See also:
https://kafka.apache.org/11/documentation.html#newconsumerconfigs
:return: kafka consumer
"""
token_provider = KafkaTokenProvider(self)
consumer_prop = self._kafka_endpoint(kafka_type="consumer")
kafka_client_id = consumer_prop["clientId"]
kafka_topic = consumer_prop["topic"]
bootstrap_servers_key = (
"bootstrapServersInternal" if kwargs.pop("internal", False) else "bootstrapServers"
)
bootstrap_servers = consumer_prop[bootstrap_servers_key]
kafka_bootstrap_servers = [
bootstrap_servers[0]["hostname"] + ":" + str(bootstrap_servers[0]["port"])
]
assert isinstance(kafka_bootstrap_servers, list)
assert isinstance(kafka_bootstrap_servers[0], str)
kafka_group_id = (consumer_prop["consumerGroupPrefix"] + group_id) if group_id else None
security_protocol = "SASL_SSL"
sasl_mechanism = "OAUTHBEARER"
api_version_auto_timeout_ms = 10000
check_crcs = False
consumer = KafkaConsumer(
kafka_topic,
group_id=kafka_group_id,
bootstrap_servers=kafka_bootstrap_servers,
client_id=kafka_client_id,
security_protocol=security_protocol,
sasl_mechanism=sasl_mechanism,
api_version_auto_timeout_ms=api_version_auto_timeout_ms,
check_crcs=check_crcs,
sasl_oauth_token_provider=token_provider,
**kwargs,
)
return consumer

[docs]
def kafka_producer(self, **kwargs) -> KafkaProducer:
"""
Instantiate and return a new KafkaProducer pre-configured to operate with the layer.

:param kwargs: Kafka producer properties.
:return: Kafka producer
"""
tokenProvider = KafkaTokenProvider(self)
producer_prop = self._kafka_endpoint(kafka_type="producer")
kafka_client_id = producer_prop["clientId"]
bootstrap_servers_key = (
"bootstrapServersInternal" if kwargs.pop("internal", False) else "bootstrapServers"
)
bootstrap_servers = producer_prop[bootstrap_servers_key]
kafka_bootstrap_servers = [
bootstrap_servers[0]["hostname"] + ":" + str(bootstrap_servers[0]["port"])
]
security_protocol = "SASL_SSL"
sasl_mechanism = "OAUTHBEARER"
api_version_auto_timeout_ms = 10000
producer = KafkaProducer(
bootstrap_servers=kafka_bootstrap_servers,
client_id=kafka_client_id,
security_protocol=security_protocol,
sasl_mechanism=sasl_mechanism,
api_version_auto_timeout_ms=api_version_auto_timeout_ms,
sasl_oauth_token_provider=tokenProvider,
**kwargs,
)
return producer

[docs]
class KafkaTokenProvider(AbstractTokenProvider):
"""
This class provides token to Kafka consumer and producer.
"""

def init(self, stream_layer: StreamLayer):
"""
Initializes the class KafkaTokenProvider.
:param stream_layer: the instance of the stream layer
"""
self.stream_layer = stream_layer

[docs]
def token(self):
"""
Returns token for Kafka consumer.
:return: token for consumer
"""
return self.stream_layer.catalog.platform.auth.token

[docs]
class VersionedLevelSummary(JsonDictDocument):
"""
Response of the versioned layer containing level summary.
"""

@property
def bounding_box(self) -> dict:
"""The bounding box of the level."""
return dict(self.json["boundingBox"])

@property
def size(self) -> int:
"""The size in bytes in the level ."""
return int(self.json["size"])

@property
def processed_timestamp(self) -> int:
"""The processed timestamp of level."""
return int(self.json["processedTimestamp"])

@property
def min_partition_size(self) -> int:
"""The minimum partition size of level."""
return int(self.json["minPartitionSize"])

@property
def max_partition_size(self) -> int:
"""The max partition size of level."""
return int(self.json["maxPartitionSize"])

@property
def version(self) -> int:
"""The version of level."""
return int(self.json["version"])

@property
def total_partitions(self) -> int:
"""The total number of partitions in level."""
return int(self.json["totalPartitions"])

[docs]
class VersionedLayerStatistics(JsonDictDocument):
"""
Response of the versioned layer containing layer statistics.
"""

@property
def level_summary(self) -> Dict[int, VersionedLevelSummary]:
"""The summary of each level."""
return {key: VersionedLevelSummary.from_dict(value) for key, value in self.json["levelSummary"].items()}

[docs]
class VersionedLayerStatisticsMap:
"""
Response of the versioned layer containing layer bitmap (bytes) data type response handling.
"""

def init(self, data: bytes) -> None:
"""Set data bytes variable"""
self._data: bytes = data

@property
def data(self) -> bytes:
"""The raw data bytes."""
return self._data

@property
def image(self):
"""
The representation of the raw bytes as an IPython.display.Image.
Note: IPython toolkit needs to be installed.

:return: IPython.display.Image
:raises RuntimeError: in case IPython toolkit is not installed
"""
try:
from IPython.display import Image
except ImportError:
raise RuntimeError("Please install IPython toolkit if you want to use this function.")
return Image(data=self._data)

[docs]
class VersionedLayer(Layer):
"""
This class provides access to data stored in versioned layers.
"""

:param data_handle: The data handle identifies a specific blob so that you can get that
blob's contents.
:param stream: whether to stream data.
:param chunk_size: the size to request each iteration when streaming data.
:param range_header: an optional Range parameter to resume download of a large response
:param billing_tag: A string which is used for grouping billing records.
:return: a blob response as bytes or iterator of bytes if stream is True
"""
return self._get_blob(
data_handle,
range_header,
billing_tag=billing_tag or self.billing_tag,
stream=stream,
chunk_size=chunk_size,
)

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

: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.
"""
return self._blob_exists(data_handle, billing_tag=billing_tag or self.billing_tag)

create fake publication

publication = Publication(
self.catalog,
[self],
publication_info=(
str(uuid4()),
self.catalog.configuration.version,
),
)
return self._put_blob(path_or_data, publication, partition_id, data_handle)

def _validate_write_layer(
self, part_size: int, path_or_data_list: List[Union[str, Path, bytes]]
):
"""
Validate parameters for write versioned layers.

:param part_size: An int representing size in MB, to upload in multiple parts minimum
value is 5MB and maximum is 50MB.
:param path_or_data_list: List of file paths or data to write to versioned layers.
:return: list of data size
"""
self._validate_write_part_size(part_size=part_size)
data_size_list = []
for path_or_data in path_or_data_list:
data_size = self._validate_write_data_size(
path_or_data=path_or_data, size_limit=5 1024 1024 * 1024
)
data_size_list.append(data_size)
return data_size_list

def _create_partition(self, **kwargs) -> VersionedPartition:
return VersionedPartition(
id=kwargs["id"],
data_handle=kwargs.get("data_handle"),
layer=self,
data_size=kwargs.get("data_size"),
checksum=kwargs.get("checksum"),
crc=kwargs.get("crc"),
)

[docs]
def get_partitions_metadata(
self,
partition_ids: Optional[List[Union[str, int]]] = None,
version: Optional[int] = None,
part: Optional[str] = None,
additional_fields: Optional[List[str]] = [
"dataSize",
"checksum",
"compressedDataSize",
"crc",
],
adapter: Optional[Adapter] = None,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
**kwargs,
) -> Union[Iterator[VersionedPartition], "pd.DataFrame"]:
"""Get list of all partition objects for the catalog with the given version.

:param partition_ids: The list of partition IDs. If not specified, all
partitions are returned
:param version: the catalog version. If not specified, the latest
catalog version will be used
:param part: indicates which part of the layer shall be queried. If not specified,
return all the partitions. It cannot be specified together with partition_ids
:param additional_fields: Additional metadata fields dataSize, checksum,
compressedDataSize, crc. By default considers all.
:param adapter: the Adapter to transform and return the result.
None to use the default adapter of the catalog.
:param stream: whether to stream data. This can reduce memory usage for very large lists of
partitions, but is generally slower.
:param chunk_size: the size to request each iteration when streaming data.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: Iterator of :class:VersionedPartition objects, or adapter-specific
:raises ValueError: in case of invalid parameter combination
"""
if partition_ids and part:
raise ValueError("partition_ids and part parameters are exclusive")
elif partition_ids:
partition_ids = [str(partition) for partition in partition_ids]

adapter = adapter or self._adapter

meta_it = self._get_partitions_list(
partition_ids=partition_ids,
version=version,
part=part,
additional_fields=additional_fields,
stream=stream,
chunk_size=chunk_size,
)
return adapter.from_versioned_metadata(meta_it, **kwargs)

[docs]
def get_partition_changes(
self,
since_version: Optional[int] = None,
version: Optional[int] = None,
part: Optional[str] = None,
additional_fields: Optional[List[str]] = [
"dataSize",
"checksum",
"compressedDataSize",
"crc",
],
adapter: Optional[Adapter] = None,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
**kwargs,
) -> Union[Iterator[VersionedPartition], "pd.DataFrame"]:
"""Get list of all partition objects for the catalog with the given version.

:param since_version: version from which partitions need to be tracked.
:param version: the catalog version. If not specified, the latest
catalog version will be used
:param part: indicates which part of the layer shall be queried. If not specified,
return all the partitions. It cannot be specified together with partition_ids
:param additional_fields: Additional metadata fields dataSize, checksum,
compressedDataSize, crc. By default considers all.
:param adapter: the Adapter to transform and return the result.
None to use the default adapter of the catalog.
:param stream: whether to stream data. This can reduce memory usage for very large lists of
partitions, but is generally slower.
:param chunk_size: the size to request each iteration when streaming data.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: Iterator of :class:VersionedPartition objects, or adapter-specific
"""

adapter = adapter or self._adapter

meta_it = self._get_partitions_changes(
since_version=since_version,
version=version,
part=part,
additional_fields=additional_fields,
stream=stream,
chunk_size=chunk_size,
)
return adapter.from_versioned_metadata(meta_it, **kwargs)

def _partition_json_to_metadata(
self,
p: Union[dict, StreamingJSONObject],
) -> VersionedPartition:
"""
Create a partition object based on the dict from an API response.

:param p: the partition dict from an API response.
:return: the partition instance.
"""
return VersionedPartition(
data_handle=p.get("dataHandle"),
id=p.get("partition", ""),
layer=self,
checksum=p.get("checksum"),
data_size=p.get("dataSize"),
compressed_data_size=p.get("compressedDataSize"),
crc=p.get("crc"),
version=p.get("version"),
)

def _get_partitions_changes(
self,
since_version: Optional[int],
version: Optional[int],
part: Optional[str],
additional_fields: Optional[List[str]],
stream: bool,
chunk_size: int,
) -> Generator[VersionedPartition, None, None]:
"""Get list of all partition objects for the catalog with the given version.

:param since_version: version from which partitions need to be tracked.
:param version: the catalog version. If None specified, the latest
catalog version will be used
:param part: indicates which part of the layer shall be queried. If None specified,
return all the partitions.
:param additional_fields: Additional metadata fields dataSize, checksum,
compressedDataSize, crc. By default considers all.
:param stream: whether to stream data. This can reduce memory usage for very large lists of
partitions, but is generally slower.
:param chunk_size: the size to request each iteration when streaming data.
:yield: :class:Partition objects.
:raises ValueError: if since_version is greater than the version.
"""
if version is None:
version = self.catalog.latest_version()

New catalog: no partitions available and will give a 400 response if you try.

if version is None or version < 0:
return

if since_version is not None:
if since_version == version:
return
elif since_version > version:
raise ValueError("since_version should not be greater than version")
else:
part_info = self._data_metadata_api.get_changes(
layer_id=self.id,
start_version=since_version,
end_version=version,
part=part,
additional_fields=additional_fields,
billing_tag=self.billing_tag,
stream=stream,
chunk_size=chunk_size,
)
else:
part_info = self._data_metadata_api.get_partitions(
layer_id=self.id,
version=version,
part=part,
additional_fields=additional_fields,
billing_tag=self.billing_tag,
stream=stream,
chunk_size=chunk_size,
)

while True:
try:
for p in find_list_of_objects(part_info, "partitions"):
yield self._partition_json_to_metadata(p)

next_url = part_info.get("next")
if next_url:
part_info = self._data_metadata_api.next_partitions(next_url)
else:
break
except TransientAccessException:
break

def _get_partitions_list(
self,
partition_ids: Optional[List[Union[str, int]]],
version: Optional[int],
part: Optional[str],
additional_fields: Optional[List[str]],
stream: bool,
chunk_size: int,
) -> Generator[VersionedPartition, None, None]:
"""Get list of all partition objects for the catalog with the given version.

:param partition_ids: The list of partition IDs. If not specified, all
partitions are returned.
:param version: the catalog version. If None specified, the latest
catalog version will be used
:param part: indicates which part of the layer shall be queried. If None specified,
return all the partitions. It cannot be specified together with partition_ids
:param additional_fields: Additional metadata fields dataSize, checksum,
compressedDataSize, crc. By default considers all.
:param stream: whether to stream data. This can reduce memory usage for very large lists of
partitions, but is generally slower.
:param chunk_size: the size to request each iteration when streaming data.
:yield: :class:Partition objects.
"""
if version is None:
version = self.catalog.latest_version()

New catalog: no partitions available and will give a 400 response if you try.

if version is None or version < 0:
return

if not partition_ids:
part_info = self._data_metadata_api.get_partitions(
layer_id=self.id,
version=version,
part=part,
additional_fields=additional_fields,
billing_tag=self.billing_tag,
stream=stream,
chunk_size=chunk_size,
)
while True:
try:
for p in find_list_of_objects(part_info, "partitions"):
yield self._partition_json_to_metadata(p)

next_url = part_info.get("next")
if next_url:
part_info = self._data_metadata_api.next_partitions(next_url)
else:
break
except TransientAccessException:
break
else:
max_size = 100

def _grouper(n, iterable, fillvalue=None):
args = [iter(iterable)] * n
return zip_longest(fillvalue=fillvalue, *args)

This loop will ignore the last group if less than max_size long.

for group in _grouper(max_size, partition_ids):

Never stream the response as the groups are too small to justify it.

part_info = self._data_query_api.get_partitions_by_id(
layer_id=self.id,
partition=group,
version=version,
additional_fields=additional_fields,
billing_tag=self.billing_tag,
)
for p in part_info.get("partitions", []):
yield self._partition_json_to_metadata(p)

[docs]
def read_partitions(
self,
partition_ids: Optional[List[Union[str, int]]] = None,
version: Optional[int] = None,
part: Optional[str] = None,
decode: bool = True,
adapter: Optional[Adapter] = None,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
**kwargs,
) -> Union[
Iterator[Tuple[VersionedPartition, bytes]], # raw
Iterator[Tuple[VersionedPartition, Iterator[bytes]]], # stream
Iterator[Tuple[VersionedPartition, Any]], # from default adapter
"pd.DataFrame", # from geopandas adapter
]:
"""
Read partition data from a layer.

:param partition_ids: The list of partition IDs. If not specified, all
partitions are read.
:param version: the catalog version. If not specified, the latest
catalog version will be used.
:param part: indicates which part of the layer shall be queried. If not specified,
return all the partitions. It cannot be specified together with partition_ids
:param decode: whether to decode the data through an adapter or return raw bytes
:param stream: whether to stream data. This implies decode=false.
:param chunk_size: the size to request each iteration when streaming data.
:param adapter: the Adapter to transform and return the result.
None to use the default adapter of the catalog.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: Iterator of :class:VersionedPartition objects each with its raw data,
in case decode=False, adapter-specific otherwise
:raises ValueError: in case decoding is requested but the adapter does not support
the content type of the layer requested, or invalid parameters
:raises LayerConfigurationException: in case decoding is requested but the
layer doesn't have any content type configured # noqa
"""
if partition_ids and part:
raise ValueError("partition_ids and part parameters are exclusive")
elif partition_ids:
partition_ids = [str(partition) for partition in partition_ids]

adapter = adapter or self._adapter

if decode and not stream:
self._verify_adapter_decoder(adapter)

meta_it = self.get_partitions_metadata(
partition_ids, version=version, part=part, adapter=self._default_adapter
)
partitions_data = map(
lambda p: (p, p.get_blob(stream=stream, chunk_size=chunk_size)), meta_it
)

if decode and not stream:
content_type = self.configuration.content_type
assert content_type
schema = self.get_schema()
return adapter.from_versioned_data(partitions_data, content_type, schema, **kwargs)
else:
return partitions_data

[docs]
def set_partitions_metadata(
self,
publication: "Publication",
update: Union[
None, # nothing to update, any adapter
Iterable[VersionedPartition], # to default adapter
"pd.DataFrame", # to geopandas adapter
] = None,
delete: Union[
None, # nothing to delete, any adapter
Iterable[Union[str, int]], # to default adapter
"pd.Series", # to geopandas adapter
] = None,
adapter: Optional[Adapter] = None,
**kwargs,
):
"""
Update the metadata of the layer as part of a publication
by publishing updated partitions and/or deleting partitions.

:param publication: the publication this operation is part of
:param update: the complete partitions to update, if any, or adapter-specific
:param delete: the partition ids to delete, if any, or adapter-specific
:param adapter: the Adapter to transform the input.
None to use the default adapter of the catalog.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
"""
self._validate_publication(publication)

adapter = adapter or self._adapter

update_it, delete_it = adapter.to_versioned_metadata(self, update, delete, **kwargs)

TODO improve _set_partitions_metadata to pass down iterators

to_update = list(update_it)
to_delete = [str(pid) for pid in delete_it]
self._set_partitions_metadata(publication, update=to_update, delete=to_delete)

def _set_partitions_metadata(
self,
publication: "Publication",
update: Iterable[VersionedPartition],
delete: Iterable[str],
) -> None:

TODO: refactor as this is almost a complete duplicate of volatile layer

n = 1000
if update:
assert all(x.layer.id == self.id for x in update)
update_partition_metadata = []
for partition in update:
update_partition_metadata.append(
{"dataHandle": partition.data_handle, "partition": partition.id, "checksum": partition.checksum, "dataSize": partition.data_size, "compressedDataSize": partition.compressed_data_size, "crc": partition.crc,}
)
pub_update_metadata_list = [
update_partition_metadata[i : i + n]
for i in range(0, len(update_partition_metadata), n)
]
for pub_metadata in pub_update_metadata_list:
body = {"partitions": pub_metadata}
self._data_publish_api.upload_partitions(
layer_id=self.id,
publication_id=publication.publication_id,
body=body,
billing_tag=self.billing_tag,
)
if delete:
delete_partition_metadata = []
for partition_id in delete:
delete_partition_metadata.append({"dataHandle": "", "partition": partition_id})
pub_delete_metadata_list = [
delete_partition_metadata[i : i + n]
for i in range(0, len(delete_partition_metadata), n)
]
for delete_pub_metadata in pub_delete_metadata_list:
delete_body = {"partitions": delete_pub_metadata}
self._data_publish_api.upload_partitions(
layer_id=self.id,
publication_id=publication.publication_id,
body=delete_body,
billing_tag=self.billing_tag,
)

[docs]
def write_partitions(
self,
publication: "Publication",
data: Union[
Iterable[Tuple[Union[str, int], Union[str, Path, bytes]]], # raw
Mapping[Union[str, int], Union[str, Path, bytes]], # raw
Iterable[Tuple[Union[str, int], Any]], # to default adapter
Mapping[Union[str, int], Any], # to default adapter
"pd.DataFrame", # to geopandas adapter
],
encode: bool = True,
adapter: Optional[Adapter] = None,
**kwargs,
):
"""
Upload content to the layer and publish the related
partition metadata as part of a publication.

:param publication: the publication this operation is part of
:param data: data to upload to the versioned layer, or adapter-specific
:param encode: whether to encode the data or upload raw bytes.
:param adapter: the Adapter to transform the input.
None to use the default adapter of the catalog.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
"""

TODO: refactor as this is almost a complete duplicate of volatile layer

self._validate_publication(publication)

Encode if needed

if encode:
adapter = adapter or self._adapter
self._verify_adapter_encoder(adapter)
content_type = self.configuration.content_type
assert content_type
schema = self.get_schema()
encoded_it: Iterator[Tuple[Union[str, int], bytes]] = adapter.to_versioned_data(
self, data, content_type, schema, **kwargs
)
else:
encoded_it = iter_tuples(data)

Upload blobs

partition_list: List[VersionedPartition] = []
for partition_id, path_or_data in encoded_it:
self._validate_write_data_size(
path_or_data=path_or_data, size_limit=50 1024 1024 * 1024
)
pid = str(partition_id) # convert int to str
partition_metadata = self._put_blob(
path_or_data=path_or_data,
publication=publication,
partition_id=pid,
)
assert isinstance(partition_metadata, VersionedPartition)
partition_list.append(partition_metadata)

Update metadata

self.set_partitions_metadata(
publication=publication, update=partition_list, adapter=self._default_adapter
)

[docs]
def get_statistics(self) -> VersionedLayerStatistics:
"""
Retrieve layer statistics.

:return: VersionedLayerStatistics object containing layer statistics.
"""
stats_data = self._data_statistics_api.get_summary(self.id)
return VersionedLayerStatistics(stats_data)

[docs]
def get_tile_map(self, data_level: str) -> VersionedLayerStatisticsMap:
"""
Retrieve layer tile map.

:param data_level: One of the Data Levels configured for this layer.
By default, assets generated at deepest data level are returned.
Note that assets returned for data levels greater than 11 represent data at data level 11.

:return: VersionedLayerStatisticsMap object containing properties data, image.
"""
tile_map_data = self._data_statistics_api.get_tile_map(
layer_id=self.id, data_level=data_level
)
return VersionedLayerStatisticsMap(tile_map_data)

[docs]
def get_size_map(self, data_level: str) -> VersionedLayerStatisticsMap:
"""
Retrieve layer size map.

:return: VersionedLayerStatisticsMap object containing properties data, image.
"""
size_map_data = self._data_statistics_api.get_size_map(
layer_id=self.id, data_level=data_level
)
return VersionedLayerStatisticsMap(size_map_data)

[docs]
def get_age_map(self, data_level: str) -> VersionedLayerStatisticsMap:
"""
Retrieve layer age map.

:return: VersionedLayerStatisticsMap object containing properties data, image.
"""
age_map_data = self._data_statistics_api.get_age_map(
layer_id=self.id, data_level=data_level
)
return VersionedLayerStatisticsMap(age_map_data)

[docs]
class VolatileLayer(Layer):
"""
This class provides access to data stored in volatile layers.
"""

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

: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.
:param stream: whether to stream data
:param chunk_size: the size to request each iteration when streaming data.
:return: a blob response as bytes or iterator of bytes if stream is True
"""
return self._data_volatile_blob_api.get_volatile_blob(
layer_id=self.id,
data_handle=data_handle,
billing_tag=billing_tag or self.billing_tag,
stream=stream,
chunk_size=chunk_size,
)

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

: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.
"""
return self._data_volatile_blob_api.check_handle_exists(
layer_id=self.id, data_handle=data_handle, billing_tag=billing_tag or self.billing_tag
)

create fake publication

publication = Publication(
self.catalog,
[self],
publication_info=(
str(uuid4()),
0,
),
)
return self._put_blob(path_or_data, publication, partition_id, data_handle)

def _validate_write_layer(self, path_or_data_list: List[Union[str, Path, bytes]]):
"""
Validate parameters for write volatile layers.

:param path_or_data_list: List of file paths or data to write to volatile layers.
:return: list of data size
"""
data_size_list = []
for path_or_data in path_or_data_list:
data_size = self._validate_write_data_size(
path_or_data=path_or_data, size_limit=MAX_SIZE_VOLATILE_PARTITION
)
data_size_list.append(data_size)
return data_size_list

def _create_partition(self, **kwargs) -> VolatilePartition:
return VolatilePartition(
id=kwargs.get("id"),
data_handle=kwargs.get("data_handle"),
layer=self,
data_size=kwargs.get("data_size"),
checksum=kwargs.get("checksum"),
crc=kwargs.get("crc"),
)

[docs]
def get_partitions_metadata(
self,
partition_ids: Optional[List[Union[str, int]]] = None,
additional_fields: Optional[List[str]] = [
"dataSize",
"checksum",
"compressedDataSize",
"crc",
],
adapter: Optional[Adapter] = None,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
**kwargs,
) -> Union[Iterator[VolatilePartition], "pd.DataFrame"]:
"""Get list of all partition objects for the catalog.

:param partition_ids: The list of partition IDs. If not specified, all
partitions are read.
:param additional_fields: Additional metadata fields dataSize, checksum,
compressedDataSize, crc. By default considers all.
:param adapter: the Adapter to transform and return the result.
None to use the default adapter of the catalog.
:param stream: whether to stream data. This can reduce memory usage for very large lists of
partitions, but is generally slower.
:param chunk_size: the size to request each iteration when streaming data.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: Iterator of :class:VolatilePartition objects, or adapter-specific
"""
if partition_ids:
partition_ids = [str(partition) for partition in partition_ids]

adapter = adapter or self._adapter

meta_it = self._get_partitions_list(
partition_ids=partition_ids,
additional_fields=additional_fields,
stream=stream,
chunk_size=chunk_size,
)

return adapter.from_volatile_metadata(meta_it, **kwargs)

def _partition_json_to_metadata(
self, p: Union[dict, StreamingJSONObject]
) -> VolatilePartition:
"""
Create a partition object based on the dict from an API response.

:param p: the partition dict from an API response.
:return: the partition instance.
"""
return VolatilePartition(
data_handle=p.get("dataHandle"),
id=p.get("partition"),
layer=self,
checksum=p.get("checksum"),
data_size=p.get("dataSize"),
compressed_data_size=p.get("compressedDataSize"),
crc=p.get("crc"),
)

def _get_partitions_list(
self,
partition_ids: Optional[List[Union[int, str]]],
additional_fields: Optional[List[str]],
stream: bool,
chunk_size: int,
) -> Generator[VolatilePartition, None, None]:
"""Get list of all partition objects for the catalog.

:param partition_ids: The list of partition IDs. If None specified, all
partitions are read.
:param additional_fields: Additional metadata fields dataSize, checksum,
compressedDataSize, crc. By default considers all.
:param stream: whether to stream data. This can reduce memory usage for very large lists of
partitions, but is generally slower.
:param chunk_size: the size to request each iteration when streaming data.
:yield: :class:Partition objects.
"""
if not partition_ids:
part_info = self._data_metadata_api.get_partitions(
layer_id=self.id,
additional_fields=additional_fields,
billing_tag=self.billing_tag,
stream=stream,
chunk_size=chunk_size,
)
while True:
try:
for p in find_list_of_objects(part_info, "partitions"):
yield self._partition_json_to_metadata(p)

next_url = part_info.get("next")
if next_url:
part_info = self._data_metadata_api.next_partitions(next_url)
else:
break
except TransientAccessException:
break
else:
max_size = 100

def _grouper(n, iterable, fillvalue=None):
args = [iter(iterable)] * n
return zip_longest(fillvalue=fillvalue, *args)

This loop will ignore the last group if less than max_size long.

for group in _grouper(max_size, partition_ids):
part_info = self._data_query_api.get_partitions_by_id(
layer_id=self.id,
partition=group,
additional_fields=additional_fields,
billing_tag=self.billing_tag,
)
try:
partitions = part_info.get("partitions", [])
except TransientAccessException:
pass
for p in partitions:
yield self._partition_json_to_metadata(p)

[docs]
def read_partitions(
self,
partition_ids: Optional[List[Union[str, int]]] = None,
decode: bool = True,
adapter: Optional[Adapter] = None,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
**kwargs,
) -> Union[
Iterator[Tuple[VolatilePartition, bytes]], # raw
Iterator[Tuple[VolatilePartition, Iterator[bytes]]], # stream
Iterator[Tuple[VolatilePartition, Any]], # from default adapter
"pd.DataFrame", # from geopandas adapter
]:
"""
Read partition data from a layer.

:param partition_ids: The list of partition IDs. If not specified, all
partitions are read.
:param decode: whether to decode the data through an adapter or return raw bytes
:param stream: whether to stream data. This implies decode=false.
:param adapter: the Adapter to transform and return the result.
None to use the default adapter of the catalog.
:param chunk_size: the size to request each iteration when streaming data.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: Iterator of :class:VolatilePartition objects each with its raw data,
in case decode=False, adapter-specific otherwise
:raises ValueError: in case decoding is requested but the adapter does not support
the content type of the layer requested, or invalid parameters
:raises LayerConfigurationException: in case decoding is requested but the
layer doesn't have any content type configured # noqa
"""
if partition_ids:
partition_ids = [str(partition) for partition in partition_ids]

adapter = adapter or self._adapter

if decode and not stream:
self._verify_adapter_decoder(adapter)

meta_it = self.get_partitions_metadata(partition_ids, adapter=self._default_adapter)

def get_blob_optional(partition) -> Optional[Tuple[VolatilePartition, bytes]]:
try:
return partition, partition.get_blob(stream=stream, chunk_size=chunk_size)
except PlatformException as e:
if e.resp.status_code == 404:
return None
else:
raise

partitions_data = filter(None, map(get_blob_optional, meta_it))

if decode and not stream:
content_type = self.configuration.content_type
assert content_type
schema = self.get_schema()
return adapter.from_volatile_data(partitions_data, content_type, schema, **kwargs)
else:
return partitions_data

def _data_handle_exists(self, partition_id: str) -> Optional[str]:
"""
Get data handle for a partition if it exists else None.
"""
partition_resp = self._data_query_api.get_partitions_by_id(
layer_id=self.id, partition=[partition_id], version=None, billing_tag=self.billing_tag
)
partitions = partition_resp["partitions"]
if partitions:
data_handle: str = partitions[0]["dataHandle"]
return data_handle
else:
return None

[docs]
def set_partitions_metadata(
self,
publication: "Publication",
update: Union[
None, # nothing to update, any adapter
Iterable[VolatilePartition], # to default adapter
"pd.DataFrame", # to geopandas adapter
] = None,
delete: Union[
None, # nothing to delete, any adapter
Iterable[Union[str, int]], # to default adapter
"pd.Series", # to geopandas adapter
] = None,
adapter: Optional[Adapter] = None,
**kwargs,
) -> None:
"""
Update the metadata of the layer as part of a publication
by publishing updated partitions and/or deleting partitions.

adapter = adapter or self._adapter

update_it, delete_it = adapter.to_volatile_metadata(self, update, delete, **kwargs)

TODO improve _set_partitions_metadata to pass down iterators

to_update = list(update_it)
to_delete = [str(pid) for pid in delete_it]
self._set_partitions_metadata(publication, update=to_update, delete=to_delete)

def _set_partitions_metadata(
self,
publication: "Publication",
update: Iterable[VolatilePartition],
delete: Iterable[str],
) -> None:
n = 1000
if update:
assert all(x.layer.id == self.id for x in update)
update_partition_metadata = []
for partition in update:
update_partition_metadata.append(
{"dataHandle": partition.data_handle, "partition": partition.id, "checksum": partition.checksum, "dataSize": partition.data_size, "compressedDataSize": partition.compressed_data_size, "crc": partition.crc,}
)
pub_update_metadata_list = [
update_partition_metadata[i : i + n]
for i in range(0, len(update_partition_metadata), n)
]
for pub_metadata in pub_update_metadata_list:
body = {"partitions": pub_metadata}
self._data_publish_api.upload_partitions(
layer_id=self.id,
publication_id=publication.publication_id,
body=body,
billing_tag=self.billing_tag,
)
if delete:
delete_partition_metadata = []
for partition_id in delete:
delete_partition_metadata.append({"dataHandle": "", "partition": partition_id})
pub_delete_metadata_list = [
delete_partition_metadata[i : i + n]
for i in range(0, len(delete_partition_metadata), n)
]
for delete_pub_metadata in pub_delete_metadata_list:
delete_body = {"partitions": delete_pub_metadata}
self._data_publish_api.upload_partitions(
layer_id=self.id,
publication_id=publication.publication_id,
body=delete_body,
billing_tag=self.billing_tag,
)

[docs]
def write_partitions(
self,
publication: "Publication",
data: Union[
Iterable[Tuple[Union[str, int], Union[str, Path, bytes]]], # raw
Mapping[Union[str, int], Union[str, Path, bytes]], # raw
Iterable[Tuple[Union[str, int], Any]], # to default adapter
Mapping[Union[str, int], Any], # to default adapter
"pd.DataFrame", # to geopandas adapter
],
encode: bool = True,
adapter: Optional[Adapter] = None,
**kwargs,
) -> None:
"""
Upload content to the layer and publish the related
partition metadata as part of a publication.

:param publication: the publication this operation is part of.
:param data: data to upload to the volatile layer, or adapter-specific
:param encode: whether to encode the data or upload raw bytes.
:param adapter: the Adapter to transform the input.
None to use the default adapter of the catalog.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
"""
self._validate_publication(publication)

Encode if needed

if encode:
adapter = adapter or self._adapter
self._verify_adapter_encoder(adapter)
content_type = self.configuration.content_type
assert content_type
schema = self.get_schema()
encoded_it: Iterator[Tuple[Union[str, int], bytes]] = adapter.to_volatile_data(
self, data, content_type, schema, **kwargs
)
else:
encoded_it = iter_tuples(data)

Upload blobs

partition_list: List[VolatilePartition] = []
for partition_id, path_or_data in encoded_it:
self._validate_write_data_size(path_or_data=path_or_data, size_limit=2 1024 1024)
pid = str(partition_id) # convert int to str
partition_metadata = self._put_blob(
path_or_data=path_or_data, publication=publication, partition_id=pid
)
assert isinstance(partition_metadata, VolatilePartition)
partition_list.append(partition_metadata)

Update metadata

self.set_partitions_metadata(
publication=publication, update=partition_list, adapter=self._default_adapter
)

[docs]
def delete_partitions(
self, publication: "Publication", partitions: Iterable[VolatilePartition]
):
"""
Delete content to selected partitions of the layer.

:param publication: the publication this operation is part of
:param partitions: identifiers of the volatile partitions to delete
"""
self._validate_publication(publication)

partition_ids: List[str] = []
for partition in partitions:
partition_ids.append(str(partition.id))

Write the metadata before deleting the blobs to avoid dangling references if the metadata

update fails. (e.g. publication was invalidated server-side)

self.set_partitions_metadata(publication=publication, delete=partition_ids)
for partition in partitions:
if partition.data_handle:
self._data_volatile_blob_api.delete_volatile_blob(
layer_id=self.id,
data_handle=partition.data_handle,
billing_tag=self.billing_tag,
)

[docs]
@dataclass
class HexbinClustering:
"""
This class defines attributes for hexbin clustering algorithm.
"""

clustering_type: str = "hexbin"
absolute_resolution: Optional[int] = None
resolution: Optional[int] = None
relative_resolution: Optional[int] = None
property: Optional[str] = None
pointmode: Optional[bool] = None

[docs]
@dataclass
class QuadbinClustering:
"""
This class defines attributes for quadbin clustering algorithm.
"""

clustering_type: str = "quadbin"
no_buffer: bool = False
relative_resolution: Optional[int] = None
resolution: Optional[int] = None
countmode: Optional[str] = None

[docs]
class InteractiveMapLayer(Layer):
"""
This class provides access to data stored in Interactive Map layers.
"""

@property
def statistics(self) -> dict:
"""
The statistical information of the layer.
"""
stats: dict = self._data_interactive_api.get_statistics(layer_id=self.id, skip_cache=True)
return stats

[docs]
def get_feature(
self,
feature_id: str,
selection: Optional[List[str]] = None,
force_2d: bool = False,
) -> Feature:
"""
Return GeoJSON feature for the provided feature_id.

:param feature_id: Feature id which is to fetched.
:param selection: A list, only these properties will be present in returned feature.
:param force_2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:return: :class:Feature object.
"""
feature = self._data_interactive_api.get_feature(
layer_id=self.id, feature_id=feature_id, selection=selection, force2d=force_2d
)
return Feature(
id=feature["id"], geometry=feature["geometry"], properties=feature["properties"]
)

[docs]
def get_features(
self,
feature_ids: List[str],
selection: Optional[List[str]] = None,
force_2d: bool = False,
**kwargs,
) -> Union[FeatureCollection, "gpd.GeoDataFrame"]:
"""
Return GeoJSON FeatureCollection for the provided feature_ids.

:param feature_ids: A list of feature identifiers to fetch.
:param selection: A list, only these properties will be present in returned features.
:param force_2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:raises ValueError: If feature_ids is empty list.
:return: FeatureCollection object, or adapter-specific
"""

if not feature_ids:
raise ValueError("Invalid input, please provide at least single feature_id")
result = self._data_interactive_api.get_features(
layer_id=self.id, feature_ids=feature_ids, selection=selection, force2d=force_2d
)
return self._adapter.from_geo_features(iter(result["features"]), **kwargs)

[docs]
def search_features(
self,
limit: int = 30000,
params: Optional[Dict[str, Union[str, list, tuple]]] = None,
selection: Optional[List[str]] = None,
skip_cache: bool = False,
force_2d: bool = False,
**kwargs,
) -> Union[FeatureCollection, "gpd.GeoDataFrame"]:
"""
Search for features in the layer based on the properties.

:param limit: A maximum number of features to return in the result. Default is 30000.
Hard limit is 100000.
:param params: A dict to represent additional filters on features to be searched.

Properties initiated with 'p.' are used to access values in the stored feature
which are under the 'properties' property.

params={"p.name": "foo"}
returns all features with a value of property p.name equal to foo.

Properties initiated with 'f.' are used to access values which are added by default in
the stored feature.The possible values are: 'f.id', 'f.createdAt' and 'f.updatedAt'.

params={"f.createdAt": 1634}
returns all features with a value of property f.createdAt equal to 1634

The query can also be written by using the long operators: "=gte", "=lte", "=gt",
"=lt" and "=cs"

params={"p.count=gte": 10}
returns all features with a value of property p.count greater than or
equal to 10.
params={"p.count=lte": 10}
returns all features with a value of property p.count less than or equal
to 10.
params={"p.count=gt": 10}
returns all features with a value of property p.count greater than 10.
params={"p.count=lt": 10}
returns all features with a value of property p.count less than 10.
params={"p.name=cs": "bar"}
returns all features with a value of property p.name which contains
bar.
:param selection: A list, only these properties will be present in returned features.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:param force_2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: FeatureCollection object, or adapter-specific
"""
result = self._data_interactive_api.search_features(
layer_id=self.id,
limit=limit,
params=params,
selection=selection,
skip_cache=skip_cache,
force2d=force_2d,
)
return self._adapter.from_geo_features(iter(result["features"]), **kwargs)

[docs]
def iter_features(
self,
chunk_size: int = 30000,
selection: Optional[List[str]] = None,
skip_cache: bool = False,
force_2d: bool = False,
) -> Iterator[Feature]:
"""
Return all the features in a Layer as Generator.

:param chunk_size: A number of features to return in single iteration.
:param selection: A list, only these properties will be present in returned features.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:param force_2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:yields: A :class:Feature object

"""
page_token = None
while True:
resp = self._data_interactive_api.iter_features(
layer_id=self.id,
limit=chunk_size,
page_token=page_token,
selection=selection,
skip_cache=skip_cache,
force2d=force_2d,
)
page_token = resp.get("nextPageToken")
features = resp["features"]
for f in features:
yield Feature(id=f["id"], geometry=f["geometry"], properties=f["properties"])
if page_token is None:
break

[docs]
def get_features_in_bounding_box(
self,
bounds: Tuple[float, float, float, float],
clip: bool = False,
limit: int = 30000,
params: Optional[Dict[str, Union[str, list, tuple]]] = None,
selection: Optional[List[str]] = None,
skip_cache: bool = False,
clustering: Optional[Union[HexbinClustering, QuadbinClustering]] = None,
force_2d: bool = False,
**kwargs,
) -> Union[FeatureCollection, "gpd.GeoDataFrame"]:
"""
Return the features which are inside a bounding box stipulated by bounds parameter.

:param bounds: A tuple of four numbers representing the West, South,
East and North margins, respectively, of the bounding box.
:param clip: A Boolean indicating if the result should be clipped
(default: False)
:param limit: A maximum number of features to return in the result. Default is 30000.
Hard limit is 100000.
:param params: A dict to represent additional filters on features to be searched.

Properties initiated with 'p.' are used to access values in the stored feature
which are under the 'properties' property.

params={"p.name": "foo"}
returns all features with a value of property p.name equal to foo.

Properties initiated with 'f.' are used to access values which are added by default
in the stored feature.The possible values are: 'f.id', 'f.createdAt' and 'f.updatedAt'

params={"f.createdAt": 1634}
returns all features with a value of property f.createdAt equal to 1634.

The query can also be written by using the long operators: "=gte", "=lte", "=gt",
"=lt" and "=cs"

params={"p.count=gte": 10}
returns all features with a value of property p.count greater than or
equal to 10.
params={"p.count=lte": 10}
returns all features with a value of property p.count less than or equal
to 10.
params={"p.count=gt": 10}
returns all features with a value of property p.count greater than 10.
params={"p.count=lt": 10}
returns all features with a value of property p.count less than 10.
params={"p.name=cs": "bar"}
returns all features with a value of property p.name which contains
bar.
:param selection: A list, only these properties will be present in returned features.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:param clustering: An object of either :class:HexbinClustering
or :class:QuadbinClustering.
:param force_2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: FeatureCollection object, or adapter-specific
"""

clustering_params =
if clustering:
clustering_options = copy.deepcopy(vars(clustering))
clustering_type = clustering_options.pop("clustering_type")
for key, val in clustering_options.items():
if val is not None:
init, *temp = key.split("_")
if temp:
new_key = "".join([init.lower(), *map(str.title, temp)])
clustering_params[new_key] = str(val).lower()
else:
clustering_params[init] = str(val).lower()
else:
clustering_type = None

result = self._data_interactive_api.get_features_by_bbox(
layer_id=self.id,
bbox=bounds,
clip=clip,
limit=limit,
params=params,
selection=selection,
skip_cache=skip_cache,
clustering=clustering_type,
clustering_params=clustering_params if clustering_params else None,
force2d=force_2d,
)
return self._adapter.from_geo_features(iter(result["features"]), **kwargs)

[docs]
def spatial_search(
self,
lng: float,
lat: float,
radius: int,
limit: int = 30000,
params: Optional[Dict[str, Union[str, list, tuple]]] = None,
selection: Optional[List[str]] = None,
skip_cache: bool = False,
force_2d: bool = False,
**kwargs,
) -> Union[FeatureCollection, "gpd.GeoDataFrame"]:
"""
Return the features which are inside the specified radius.

:param lng: The longitude in WGS'84 decimal degree (-180 to +180) of the center Point.
:param lat: The latitude in WGS'84 decimal degree (-90 to +90) of the center Point.
:param radius: Radius in meter which defines the diameter of the search request.
:param limit: The maximum number of features in the response. Default is 30000.
Hard limit is 100000.
:param params: A dict to represent additional filters on features to be searched.

Properties initiated with 'p.' are used to access values in the stored feature
which are under the 'properties' property.

params={"p.name": "foo"}
returns all features with a value of property p.name equal to foo.

Properties initiated with 'f.' are used to access values which are added by default
in the stored feature.The possible values are: 'f.id', 'f.createdAt' and 'f.updatedAt'

params={"f.createdAt": 1634}
returns all features with a value of property f.createdAt equal to 1634

The query can also be written by using the long operators: "=gte", "=lte", "=gt",
"=lt" and "=cs"

params={"p.count=gte": 10}
returns all features with a value of property p.count greater than or
equal to 10.
params={"p.count=lte": 10}
returns all features with a value of property p.count less than or equal
to 10.
params={"p.count=gt": 10}
returns all features with a value of property p.count greater than 10.
params={"p.count=lt": 10}
returns all features with a value of property p.count less than 10.
params={"p.name=cs": "bar"}
returns all features with a value of property p.name which contains
bar.
:param selection: A list, only these properties will be present in returned features.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:param force_2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: FeatureCollection object or 'Geo dataframe' specific to adapter.
"""
result = self._data_interactive_api.get_features_with_radius_search(
layer_id=self.id,
lng=lng,
lat=lat,
radius=radius,
limit=limit,
params=params,
selection=selection,
skip_cache=skip_cache,
force2d=force_2d,
)
return self._adapter.from_geo_features(iter(result["features"]), **kwargs)

[docs]
def spatial_search_geometry(
self,
geometry: Union[Feature, Geometry, dict, Any],
radius: Optional[int] = None,
limit: int = 30000,
params: Optional[Dict[str, Union[str, list, tuple]]] = None,
selection: Optional[List[str]] = None,
skip_cache: bool = False,
force_2d: bool = False,
**kwargs,
) -> Union[FeatureCollection, "gpd.GeoDataFrame"]:
"""
Return the features which are inside the specified radius and geometry.

The origin point is calculated based on the provided geometry.

:param geometry: Geometry which will be used in intersection. It supports
GeoJSON Feature, GeoJSON Geometry, or __geo_interface__.
:param radius: Radius in meter which defines the diameter of the search request.
:param limit: The maximum number of features in the response. Default is 30000.
Hard limit is 100000.
:param params: A dict to represent additional filters on features to be searched.

Properties initiated with 'p.' are used to access values in the stored
feature which are under the 'properties' property.

params={"p.name": "foo"}
returns all features with a value of property p.name equal to foo.

Properties initiated with 'f.' are used to access values which are added by default
in the stored feature.The possible values are: 'f.id', 'f.createdAt' and 'f.updatedAt'

params={"f.createdAt": 1634}
returns all features with a value of property f.createdAt equal to 1634.

The query can also be written by using the long operators: "=gte", "=lte", "=gt",
"=lt" and "=cs"

params={"p.count=gte": 10}
returns all features with a value of property p.count greater than or
equal to 10.
params={"p.count=lte": 10}
returns all features with a value of property p.count less than or equal
to 10.
params={"p.count=gt": 10}
returns all features with a value of property p.count greater than 10.
params={"p.count=lt": 10}
returns all features with a value of property p.count less than 10.
params={"p.name=cs": "bar"}
returns all features with a value of property p.name which contains
bar.
:param selection: A list, only these properties will be present in returned features.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:param force_2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: FeatureCollection object, or adapter-specific
"""
if hasattr(geometry, GEO_INTERFACE_MARKER):
geometry = getattr(geometry, GEO_INTERFACE_MARKER)
if hasattr(geometry, "geometry"):
geometry = getattr(geometry, "geometry")
result = self._data_interactive_api.get_features_with_geometry_intersection(
layer_id=self.id,
data=geometry,
radius=radius,
limit=limit,
params=params,
selection=selection,
skip_cache=skip_cache,
force2d=force_2d,
)
return self._adapter.from_geo_features(iter(result["features"]), **kwargs)

[docs]
def write_feature(self, feature_id: str, data: Union[Feature, dict]) -> None:
"""
Write GeoJSON feature to Layer.

:param feature_id: Identifier for the feature.
:param data: GeoJSON feature which is written to layer.
"""
self._data_interactive_api.write_feature(
layer_id=self.id, feature_id=feature_id, data=data
)

[docs]
def update_feature(self, feature_id: str, data: Union[Feature, dict]) -> None:
"""
Update the GeoJSON feature in the Layer.

:param feature_id: A feature_id to be updated.
:param data: A GeoJSON Feature object to update.
"""
self._data_interactive_api.update_feature(
layer_id=self.id, feature_id=feature_id, data=data
)

[docs]
def delete_feature(self, feature_id: str) -> None:
"""
Delete feature from the layer.

:param feature_id: A feature_id to be deleted.
"""
self._data_interactive_api.delete_feature(layer_id=self.id, feature_id=feature_id)

[docs]
def write_features(
self,
features: Optional[
Union[FeatureCollection, dict, Iterator[Feature], List[Feature], "gpd.GeoDataFrame"]
] = None,
from_file: Optional[Union[str, Path]] = None,
feature_count: int = 2000,
**kwargs,
) -> None:
"""
Write GeoJSON FeatureCollection to layer.

As API has a limitation on the size of features, features are divided into groups,
and each group has number of features based on feature_count.

:param features: Features represented by :class:FeatureCollection, Dict,
:class:Iterator, list of features, or adapter-specific
:param from_file: Path of GeoJSON file.
:param feature_count: An int representing a number of features to upload at a time.
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
"""
if from_file:
with open(from_file) as fh:
feature_col = json.load(fh)
feature_groups = grouper(size=feature_count, iterable=feature_col["features"])
else:
features = list(self._adapter.to_geo_features(data=features, **kwargs))
feature_groups = grouper(size=feature_count, iterable=features)
if feature_groups:
self._upload_features(feature_groups=feature_groups)

def _upload_features(self, feature_groups: Iterator[Union[Feature, Dict]]) -> None:
features_set = set()
for group in feature_groups:
features_list = []
for feature in group:
if feature:
if "id" not in feature:
feature["id"] = hashlib.md5(
json.dumps(feature, sort_keys=True).encode("utf-8")
).hexdigest()
if feature["id"] not in features_set:
features_set.add(feature["id"])
features_list.append(feature)
else:
logger.debug(
f"feature with id {feature['id']} is skipped due to duplicate id"
)
feature_collection = FeatureCollection(features=features_list)
self._data_interactive_api.write_features(layer_id=self.id, data=feature_collection)

[docs]
def update_features(
self, data: Union[FeatureCollection, dict, "gpd.GeoDataFrame"], **kwargs
) -> None:
"""
Update multiple features provided as FeatureCollection object.

:param data: A :class:FeatureCollection, dict, or adapter-specific
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
"""
data_feature = FeatureCollection(list(self._adapter.to_geo_features(data=data, **kwargs)))
self._data_interactive_api.update_features(layer_id=self.id, data=data_feature)

[docs]
def delete_features(self, feature_ids: Union[List[str], "pd.Series"], **kwargs) -> None:
"""
Delete features from layer.

:param feature_ids: A list of feature_ids to be deleted, or adapter-specific
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
"""
feature_ids = list(self._adapter.to_feature_ids(data=feature_ids, **kwargs))
if feature_ids:
self._data_interactive_api.delete_features(layer_id=self.id, feature_ids=feature_ids)

[docs]
def subscribe(
self,
subscription_name: str,
description: str,
destination_catalog_hrn: str,
destination_layer_id: str,
interactive_map_subscription_type: InteractiveMapSubscriptionType,
) -> InteractiveMapSubscription:
"""
Method to Subscribe to a Stream Layer from Layer's Catalog HRN.
Source Layer is the current layer and Source Catalog is Layer's Catalog which it belongs.

:param subscription_name: Name of the subscription.
:param description: Description of the subscription.
:param destination_catalog_hrn: Catalog HRN of the destination Catalog.
:param destination_layer_id: Layer Id of the destination Stream Layer.
:param interactive_map_subscription_type: InteractiveMapSubscriptionType containing type of
subscription.
:raises KeyError: in case statusToken in Response of createSubscription.
:raises ValueError: in case Created Subscription Status is NOT Active after
multiple retry till max retry time.
:return: InteractiveMapSubscription object containing details of the created subscription.
"""

source_catalog_hrn = self.catalog.hrn
source_layer = self.id

configuration_object: dict = dict(
{"subscriptionName": subscription_name, "description": description, "sourceCatalog": source_catalog_hrn, "sourceLayer": source_layer, "destinationCatalog": destination_catalog_hrn, "destinationLayer": destination_layer_id, "interactiveMapSubscription": {"type": interactive_map_subscription_type.value},}
)

subscription_status_resp = dict({"status": ""})
start_time = time.time()

create_subscription_resp: dict = self._data_config_api.create_subscription(
configuration_object=configuration_object
)

logger.debug(f"Create Subscription Response : {create_subscription_resp}")

if "statusToken" not in create_subscription_resp:
raise KeyError("Missing statusToken in Response of createSubscription.")

time.sleep(min(0.1, self.catalog.platform._polling_wait))

while time.time() - start_time < self.catalog.platform._retry_max_time:
subscription_status_resp = self._data_config_api.get_subscription_status(
status_token=create_subscription_resp["statusToken"]
)
logger.debug(f"Subscription Status Response : {subscription_status_resp}")

if subscription_status_resp["status"] == "active":
break

time.sleep(self.catalog.platform._polling_wait)

logger.debug(
f"Total wait time for Subscription Status Response : {time.time() - start_time}"
)

if "status" in subscription_status_resp and subscription_status_resp["status"] != "active":
raise ValueError(
f"Created Subscription Status is NOT Active "
f"for statusToken {create_subscription_resp['statusToken']}."
)

subscription_resp: dict = self._data_config_api.get_subscription(
subscription_hrn=subscription_status_resp["subscriptionHrn"]
)

subscription_resp.update(create_subscription_resp)

return InteractiveMapSubscription(subscription_resp)

[docs]
class ObjectType(enum.Enum):
"""
ObjectType defines the different types of object stored in an :class:ObjectStoreLayer.
"""

DIRECTORY = "commonPrefix"
OBJECT = "object"

[docs]
@dataclass
class ObjectMetadata:
"""
Metadata and details of an object stored in an :class:ObjectStoreLayer.

This includes, among others, object type and size, HTTP content type and last modified date.
"""

key: str
last_modified: Optional[str]
size: Optional[int]
object_type: ObjectType
content_type: Optional[str]
content_encoding: Optional[str]

[docs]
class ObjectStoreLayer(Layer):
"""
This class provides access to data stored in object store layers.
"""

MIN_UPLOAD_PART_SIZE = 5
MAX_UPLOAD_PART_SIZE = 96
MB = 1024 * 1024
__key_regex = re.compile(r"^[a-zA-Z0-9.[]=()/-`]450$")

def init(self, layer_id: str, catalog: "Catalog"):
"""Initialize ObjectStoreLayer instance.

:param layer_id: a string with the layer ID of this layer
:param catalog: the instance of the Catalog this layer belongs to
"""
super().init(layer_id, catalog)
self._max_upload_part_size = (
self.catalog.platform.application_config.max_object_store_upload_part_size
)

@staticmethod
def _validate_key(key: str) -> None:
if not ObjectStoreLayer.__key_regex.match(key):
raise ValueError("Invalid key passed")

[docs]
def key_exists(self, key: str) -> bool:
"""
Check if the layer contains an object with the given key.

:param key: the object key to check
:return: if the layer contain an object with the given key
"""
self._validate_key(key)
return self._data_object_blob_api.key_exists(layer_id=self.id, key=key)

[docs]
def get_object_metadata(self, key: str) -> ObjectMetadata:
"""
Get the metadata of the object with the given key.

:param key: key of the object to fetch metadata of
:return: object metadata of the given object
"""
self._validate_key(key)
resp = self._data_object_blob_api.get_object(layer_id=self.id, key=key, only_meta=True)
assert not resp.content
opt_size = resp.headers.get("Content-Length")
return ObjectMetadata(
key=key,
last_modified=resp.headers.get("Last-Modified"),
size=int(opt_size) if opt_size else None,
object_type=ObjectType.OBJECT,
content_type=resp.headers.get("Content-Type"),
content_encoding=resp.headers.get("Content-Encoding"),
)

[docs]
def read_object(
self,
key: str,
include_metadata: bool = False,
stream: bool = False,
chunk_size: int = DEFAULT_ITER_CHUNK_SIZE,
) -> Union[bytes, Iterator[bytes], Tuple[Union[bytes, Iterator[bytes]], ObjectMetadata]]:
"""
Read and return the content of an object.

Optionally, also return the corresponding object metadata.

:param key: key for the object to read
:param include_metadata: whether to also return the object metadata
:param stream: whether to stream data
:param chunk_size: the size to request each iteration when streaming data
:return: the content of the object and, if requested, also its metadata
"""

self._validate_key(key)
resp = self._data_object_blob_api.get_object(
layer_id=self.id,
key=key,
only_meta=False,
stream=stream,
storage_layer_access=self.has_storage_layer_access,
)
content = cast(
Union[bytes, Iterator[bytes]],
ChunkedGet(self._data_object_blob_api, resp, chunk_size) if stream else resp.content,
)
if include_metadata:
opt_size = resp.headers.get("Content-Length")
object_metadata = ObjectMetadata(
key=key,
last_modified=resp.headers.get("Last-Modified"),
size=int(opt_size) if opt_size else None,
object_type=ObjectType.OBJECT,
content_type=resp.headers.get("Content-Type"),
content_encoding=resp.headers.get("Content-Encoding"),
)
return content, object_metadata
else:
return content

def _check_upload_part_size(self, size: int):
"""
Checks the maximum size of uploaded parts in MB (megabytes).
The default is 96. Minimum is 5MB and maximum is 96 (from server side).

:param size: max. size of uploaded parts.
:raises ValueError: in case the value is less than 5 or greater than 96.
"""
if size < ObjectStoreLayer.MIN_UPLOAD_PART_SIZE:
raise ValueError(
"upload part size needs to be greater than or equal to ; actual ".format(
ObjectStoreLayer.MIN_UPLOAD_PART_SIZE,
size,
)
)
if size > ObjectStoreLayer.MAX_UPLOAD_PART_SIZE:
raise ValueError(
"upload part size needs to be less than or equal to ; actual ".format(
ObjectStoreLayer.MAX_UPLOAD_PART_SIZE,
size,
)
)

[docs]
def set_max_upload_part_size(self, size: int):
"""
Sets the maximum size of uploaded parts in MB (megabytes).

:param size: max. size of uploaded parts.
"""
self._check_upload_part_size(size)
self._max_upload_part_size = size

[docs]
def write_object(
self,
key: str,
path_or_data: Union[str, Path, bytes],
content_type: str = "application/octet-stream",
overwrite: bool = True,
upload_part_size: Optional[int] = None,
content_encoding: Optional[str] = None,
):
"""
Write an object to the object store layer.
If file/bytes size is larger than max. upload part size then the blob will
be written in multiple parts.

This functions adds a new object or overwrites an existing object.

:param key: key for the object to write.
:param path_or_data: data to be written.
:param content_type: the standard MIME type describing the format of the data.
:param overwrite: if True then this method will overwrite if the key exists,
and if False then this method will raise error if key exists.
:param upload_part_size: optional size of upload parts; if not specified the
class' default is used
:param content_encoding: Content-encoding of the object. This header is optional.
For more information, see https://tools.ietf.org/html/rfc2616#section-14.11
:return: None
:raises ValueError: in case the file does not exist or upload_part_size is out of range.
"""

self._validate_key(key)
headers =
if not overwrite:
if self.key_exists(key=key):
raise ValueError(
f"Key: {key} already exists, cannot overwrite with param overwrite=False."
)

if upload_part_size is None:
upload_part_size = self._max_upload_part_size
self._check_upload_part_size(upload_part_size)

if isinstance(path_or_data, bytes):
data_size = len(path_or_data)
if data_size > upload_part_size * ObjectStoreLayer._MB:
return self._multipart_upload(
key=key,
path_or_data=path_or_data,
content_type=content_type,
upload_part_size=upload_part_size,
content_encoding=content_encoding,
)
headers["Content-Length"] = str(data_size)
object_data = path_or_data
else:
if not Path(path_or_data).is_file():
raise ValueError(f"File: {path_or_data} does not exist.")
file_size = os.path.getsize(path_or_data)
if file_size > upload_part_size * ObjectStoreLayer._MB:
return self._multipart_upload(
key=key,
path_or_data=path_or_data,
content_type=content_type,
upload_part_size=upload_part_size,
content_encoding=content_encoding,
)
headers["Content-Length"] = str(file_size)
with open(path_or_data, "rb") as file_data:
object_data = file_data.read()
if content_type:
headers["Content-Type"] = content_type
if content_encoding:
headers["Content-Encoding"] = content_encoding
self._data_object_blob_api.put_object(
layer_id=self.id, key=key, object_data=object_data, headers=headers
)

[docs]
def delete_object(self, key: str, strict: bool = False):
"""
Delete an object from the object store layer.

:param key: key for the object to delete
:param strict: when True, raise a PlatformException if the object doesn't exist,
when False, no exception is raised
:raises PlatformException: if the platform responds with an HTTP error
"""
self._validate_key(key)
if strict and not self.key_exists(key):
resp = Response()
resp.status_code = 404
resp.reason = f"Object not found: {key}"
raise PlatformException(resp)

self._data_object_blob_api.delete_object(self.id, key=key)

[docs]
def list_keys(
self,
parent: Optional[str] = None,
deep: bool = False,
) -> List[str]:
"""
List the keys of the objects stored in the layer.

[docs]
def iter_keys(
self, parent: Optional[str] = None, deep: bool = False, limit: int = 1000
) -> Iterator[str]:
"""
Iterate over the keys of the objects stored in the layer.

:param parent: a string that tells what "directory" should be the root
for the returned content. When not set, the root is assumed
:param deep: if True, returns also keys from the subdirectories
:param limit: number of results to return per request call: a larger value
performs larger but less frequent requests to the service, a smaller
value performs shorter but more frequent requests to the service.
The overall content retrieved is independent of this value.
To limit the amount of keys returned, simply filter the iterator
or consume the iterator up to the number of elements wanted.
:return: an iterator of object keys
"""
return self._get_key_metadata(limit, parent, deep, True) # type: ignore

def _get_key_metadata(
self, limit, parent, deep, only_keys
) -> Union[Iterator[ObjectMetadata], Iterator[str]]:
"""
Internal method to get keys along with their metadata.

:param parent: a string that tells what "directory" should be the root
:param limit: number of results to return per request call
:param deep: a boolean if true, returns all the keys from its subdirectories
:param only_keys: if true returns only keys else :class:ObjectMetadata objects
:yields: yield keys or :class:ObjectMetadata objects
:raises NotImplementedError: It throws NotImplementedError, if adapter is not supported.
"""
if not isinstance(self._adapter, DefaultAdapter):
raise NotImplementedError("Adapter is not supported for object store layers.")

page_token = None
while True:
params = {"parent": parent, "pageToken": page_token, "limit": limit, "deep": deep}
resp = self._data_object_blob_api.list_keys(layer_id=self.id, params=params)
page_token = resp.get("pageToken")
for key_metadata in resp["items"]:
yield (
key_metadata.get("name")
if only_keys
else ObjectMetadata(
key=key_metadata.get("name"),
last_modified=key_metadata.get("lastModified"),
size=key_metadata.get("size"),
object_type=ObjectType(key_metadata.get("type")),
content_type=None,
content_encoding=None,
)
)

if page_token is None:
break

[docs]
def get_objects_metadata(
self, parent: Optional[str] = None, limit: int = 1000, deep: bool = False
) -> Iterator[ObjectMetadata]:
"""
Iterate over the metadata of the objects stored in the layer.

:param parent: a string that tells what "directory" should be the root
for the returned content. When not set, the root is assumed
:param deep: if True, returns also metadata from the subdirectories
:param limit: number of results to return per request call: a larger value
performs larger but less frequent requests to the service, a smaller
value performs shorter but more frequent requests to the service.
The overall content retrieved is independent of this value.
To limit the amount of metadata returned, simply filter the iterator
or consume the iterator up to the number of elements wanted.
:return: an iterator of :class:ObjectMetadata
"""
return self._get_key_metadata( # type: ignore
limit=limit, parent=parent, deep=deep, only_keys=False
)

def _multipart_upload(
self,
key: str,
path_or_data: Union[str, Path, bytes],
content_type: str,
upload_part_size: int,
content_encoding: Optional[str] = None,
):
"""
Publishes 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 500GB.

:param key: The key identifies a specific blob so that you can get that blob's contents.
:param path_or_data: Path of the file or data to be read and upload as a partition.
:param content_type: A standard MIME type describing the format of the blob data.
:param upload_part_size: size of upload parts
:param content_encoding: Content-encoding of the object. This header is optional.
For more information, see https://tools.ietf.org/html/rfc2616#section-14.11
:return: None

"""
content = self._data_object_blob_api.start_multipart_upload(
self.id, key=key, content_type=content_type, content_encoding=content_encoding
)
return self._upload_blob_part(
key=key,
token=content["multipartToken"],
path_or_data=path_or_data,
content_type=content_type,
upload_part_size=upload_part_size,
)

def _get_multipart_upload_status(self, token: str):
"""
Gets the status of a multipart upload.
The status can be received only when the upload has been completed.

:param token: The identifier of the multipart upload (token).
This token is returned when the multipart upload is initiated.
:return: returns status of multipart upload.
"""
content = self._data_object_blob_api.get_multipart_upload_status(
self.id, multipart_token=token
)
return content["status"]

def _complete_multipart_upload(self, token, data):
"""
This method is called when all parts have been uploaded.

:param 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
"""
self._data_object_blob_api.complete_multipart_upload(
self.id, multipart_token=token, data=data
)

def _cancel_multipart_upload(self, token: str):
"""
Cancels an entire multipart upload operation. You can only cancel a multipart upload before
it has been completed.

:param token: The identifier of the multipart upload (token).
This token is returned when the multipart upload is initiated.
"""
self._data_object_blob_api.cancel_multipart_upload(self.id, multipart_token=token)

def _upload_blob_part(self, key, token, path_or_data, content_type, upload_part_size):
"""
Read input file in chunks of 50Mb each and creates urls.

:param key: The key identifies a specific blob so that you can get that blob's contents.
:param token: The identifier of the multipart upload (token).
This token is returned when the multipart upload is initiated.
:param path_or_data: Path of the file or data to be read and upload as a partition.
:param content_type: A standard MIME type describing the format of the blob data.
:param upload_part_size: size of upload parts
:raises Exception: If platform responds with an HTTP error.

"""
part_ids = []
part_number = 1
loop = self.__get_event_loop()
try:
future = asyncio.ensure_future(
self._read_and_upload_parts(path_or_data, token, content_type, upload_part_size)
)
results = loop.run_until_complete(future)
for id in results:
part_id = json.loads(id.decode("utf-8"))
part_id["number"] = part_number
part_ids.append(part_id)
part_number += 1

data = {"parts": part_ids}
self._complete_multipart_upload(token, data)
self._get_multipart_upload_status(token)

except Exception:
self._cancel_multipart_upload(token)
raise

def __get_event_loop(self):
try:
loop = asyncio.get_event_loop() # event loop
except RuntimeError as e:
if e.args[0].startswith("There is no current event loop"):
logger.info("Trying with new event loop")
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
return loop

async def _read_and_upload_parts(
self,
path_or_data: Union[str, bytes, Path],
token: str,
content_type: str,
part_size: int,
):
"""
Read input file in chunks of 50Mb each and creates urls.

:param path_or_data: Path of the file or data to be read and upload as a partition.
:param token: The identifier of the multipart upload (token). This token is returned
when the multipart upload is initiated.
:param content_type: A standard MIME type describing the format of the blob data.
:param part_size: An int representing size in MB.
:return: A list of dict with key as id and value as token.
"""
tasks = []
completed_tasks = []
with Partition.get_data_handler(path_or_data) as data_handler:
data = data_handler.read(part_size 1024 1024) # reading in chunks of megabytes
part_num = 1
while data:
logger.debug(f"Starting part number: {part_num}")
task = self._data_object_blob_api.upload_blob_part(
self.id,
multipart_token=token,
part_number=part_num,
object_part=data,
content_type=content_type,
)
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]
def is_directory(self, key: str) -> bool:
"""
Check if given key is a directory.

:param key: key of the object.
:return: returns True if the given key is a directory.
"""
if key.endswith("/"):
key = key.rstrip("/")

object_metadata = next(self.get_objects_metadata(parent=key, limit=1), None)
return object_metadata is not None and object_metadata.key != key

[docs]
def delete_all_objects(self, parent_key: str = "/", strict: bool = False):
"""
Delete all objects which are associated with given key from the object store layer.

:param parent_key: parent key for the object to delete
:param strict: when True, raise a PlatformException if the object doesn't exist,
when False, no exception is raised
:raises PlatformException: if the platform responds with an HTTP error
"""
self._validate_key(parent_key)
if strict and not self.key_exists(parent_key):
resp = Response()
resp.status_code = 404
resp.reason = f"Object not found: {parent_key}"
raise PlatformException(resp)

keys = self.list_keys(parent=parent_key, deep=True)
for key in keys:
self._data_object_blob_api.delete_object(self.id, key=key)

[docs]
def copy_object(self, key: str, copy_from: str, replace: bool = False):
"""
Copy object using the source to copy from in the object store layer .

:param key: key for the object to created.
:param copy_from: key for the object to copy from.
:param replace: if true, will replace the object while copying if
the destination already exists. This replace is not atomic, if the delete
is succeeded and put object fails then the object is gone.
:raises ValueError: in case given key and copy_from are same or
destination already exists with replace=False.
"""
self._validate_key(key)
self._validate_key(copy_from)
if key == copy_from:
raise ValueError(
f"Given Key {key} is same as the one to copy from {copy_from}" # noqa: W604
)
assert self.key_exists(copy_from)
if self.key_exists(key):
if replace:
self.delete_object(key=key)
else:
raise ValueError(f"Destination key {key} already exists") # noqa: W604
self._data_object_blob_api.put_object(self.id, key=key, source=copy_from)