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.

"""HERE platform catalog abstraction."""

import logging
import os
import time
import webbrowser
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Dict, Iterator, List, Mapping, Optional, Union, cast

from deprecated import deprecated
from here.platform.adapter import Adapter
from here.platform.api.aaa_authorization_api import AAAAuthorizationApi
from here.platform.api.artifact_api import ArtifactApi
from here.platform.api.base_api import BaseApi
from here.platform.api.data_blob_api import DataBlobApi
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.factory.apifactory import API, APIFactory
from here.platform.exceptions import CatalogConfigurationException, LayerConfigurationException
from here.platform.layer import (
IndexLayer,
InteractiveMapLayer,
Layer,
LayerConfiguration,
LayerType,
ObjectStoreLayer,
StreamIngestion,
StreamLayer,
VersionedLayer,
VolatileLayer,
)
from here.platform.model import Publication, PublicationState
from here.platform.model.version_dependency import VersionDependency
from here.platform.models import (
AutomaticVersionDeletion,
Coverage,
Notifications,
Owner,
Replication,
)
from here.platform.partition import Partition
from here.platform.utils import JsonDictDocument

if TYPE_CHECKING:
import pandas as pd
from here.platform.platform import Platform

logger = logging.getLogger(name)

[docs]
@deprecated(version="2.18", reason="Please use here.platform.model.VersionDependency")
class Dependency(VersionDependency):
"""
This class is deprecated. Please use here.platform.model.VersionDependency.
"""

pass

[docs]
class Version(JsonDictDocument):
"""
Represents a catalog version with its details.
"""

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

@property
def dependencies(self) -> List[VersionDependency]:
"""The list of dependencies on external catalogs and their versions"""
return [VersionDependency.from_dict(d) for d in self.json["dependencies"]]

@property
def shared_dependencies(self) -> List[VersionDependency]:
"""The list of dependencies on external catalogs and their versions"""
return [VersionDependency.from_dict(d) for d in self.json["sharedDependencies"]]

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

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

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

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

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

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

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

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

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

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

@property
def owner(self) -> Optional[Owner]:
"""The owner of the catalog"""
return Owner(self.json["owner"])

@property
def replication(self) -> Optional[Replication]:
"""The replication set for the catalog"""
return Replication(self.json["replication"])

@property
def automatic_version_deletion(self) -> Optional[AutomaticVersionDeletion]:
"""Specifies the number of versions to keep for the catalog"""
return AutomaticVersionDeletion(self.json["automaticVersionDeletion"])

@property
def layers(self) -> List[LayerConfiguration]:
"""The layers in the catalog"""
return [LayerConfiguration.from_dict(t) for t in self.json["layers"]]

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

@property
def notifications(self) -> Optional[Notifications]:
"""Indicates whether or not to notify each time the version of the catalog changes"""
return Notifications(self.json["notifications"])

[docs]
class Catalog:
"""HERE platform catalog abstraction."""

def init(
self,
hrn: str,
platform: "Platform",
adapter: Optional[Adapter] = None,
billing_tag: Optional[str] = None,
):
"""
Instantiate catalog given its HRN, API instances, and schema registry.

:param hrn: the HERE Resource Name of the catalog
:param platform: a Platform instance
:param adapter: the Adapter to transform data between different representations
None to use the default adapter of the platform.
:param billing_tag: A string to represent a grouping of billing records.
"""
self.hrn = hrn
self.platform = platform
self.adapter: Adapter = adapter or self.platform.adapter
self._data_config_api = self.platform.data_config_api # type: ignore
self._base_api: BaseApi = self.platform.base_api
self.billing_tag: Optional[str] = billing_tag

self._api_factory: APIFactory = self.platform.api_factory

These are lazy-loaded

self._details: Optional[dict] = None
self._layer_details: Optional[Dict[str, dict]] = None
self._configuration: Optional[CatalogConfiguration] = None
self._layer_configuration: Optional[Dict[str, LayerConfiguration]] = None

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

:return: DataBlobApi instance.
"""
return cast(DataBlobApi, self._api_factory.get_api(API.DATA_BLOB, self.hrn))

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

:return: DataVolatileBlobApi instance.
"""
return cast(
DataVolatileBlobApi, self._api_factory.get_api(API.DATA_VOLATILE_BLOB, self.hrn)
)

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

:return: DataMetadataApi instance.
"""
return cast(DataMetadataApi, self._api_factory.get_api(API.DATA_METADATA, self.hrn))

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

:return: DataPublishApi instance.
"""
return cast(DataPublishApi, self._api_factory.get_api(API.DATA_PUBLISH, self.hrn))

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

:return: DataQueryApi instance.
"""
return cast(DataQueryApi, self._api_factory.get_api(API.DATA_QUERY, self.hrn))

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

:return: DataIndexApi instance.
"""
return cast(DataIndexApi, self._api_factory.get_api(API.DATA_INDEX, self.hrn))

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

:return: DataStreamApi instance.
"""
return cast(DataStreamApi, self._api_factory.get_api(API.DATA_STREAM, self.hrn))

@property
def _artifact_api(self) -> ArtifactApi:
"""
Lazy loads Artifact API.

:return: ArtifactApi instance.
"""
return cast(ArtifactApi, self._api_factory.get_api(API.ARTIFACT, self.hrn))

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

This API is not implemented on the Local Data Service.

:return: DataIngestApi instance
"""
return cast(DataIngestApi, self._api_factory.get_api(API.DATA_INGEST, self.hrn))

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

This API is not implemented on the Local Data Service.

:return: DataInteractiveApi instance
"""
return cast(DataInteractiveApi, self._api_factory.get_api(API.DATA_INTERACTIVE, self.hrn))

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

This API is not implemented on the Local Data Service.

:return: DataObjectBlobApi instance
"""
return cast(DataObjectBlobApi, self._api_factory.get_api(API.DATA_OBJECT_BLOB, self.hrn))

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

This API is not implemented on the Local Data Service.

:return: DataStatisticsApi instance
"""
return cast(DataStatisticsApi, self._api_factory.get_api(API.DATA_STATISTICS, self.hrn))

@property
def _aaa_auth_api(self) -> AAAAuthorizationApi:
"""
Returns parent platform's AAAAuthorization instance.

This API is not implemented on the Local Data Service.

:return: AAAAuthorization instance
"""
return self.platform.aaa_auth_api # type: ignore

def _ensure_catalog_config_loaded(self):
if not self._details:
json_config = self._data_config_api.get_catalog_details(catalog_hrn=self.hrn)
self._details = json_config # TODO: remove
self._layer_details = {layer["id"]: layer for layer in json_config["layers"]} # TODO: remove
self._configuration = CatalogConfiguration.from_dict(json_config)
self._layer_configuration = {layer["id"]: LayerConfiguration.from_dict(layer) for layer in json_config["layers"]}

def _invalidate_catalog_config(self):
self._details = None
self._layer_details = None
self._configuration = None
self._layer_configuration = None

[docs]
def lookup_apis(self, region: Optional[str] = None) -> dict:
"""
Lookup implemented APIs for the given HRN.

:param region: an Optional param to look up a specific region for a given resource
:return: dictionary with API descriptions for the HRN
"""
res: dict = self.platform.lookup_api.get_resource_api_list(hrn=self.hrn, region=region)
return res

@property
def configuration(self) -> CatalogConfiguration:
"""The configuration of the catalog"""
self._ensure_catalog_config_loaded()
assert self._configuration is not None
return self._configuration

[docs]
def get_details(self) -> dict:
"""
Fetch and buffer catalog details from the platform.

:return: catalog details
"""
self._ensure_catalog_config_loaded()
assert self._details is not None
return self._details

[docs]
def has_layer(self, layer_id: str) -> bool:
"""
Check if a layer ID exists in the catalog.

:param layer_id: layer ID in the catalog.
:return: a boolean value indicating if a layer exists
"""
self._ensure_catalog_config_loaded()
assert self._layer_details is not None
return layer_id in self._layer_details

[docs]
def first_version(self) -> Optional[int]:
"""
Return first version of the catalog available on the platform.

:return: first version number of the catalog if present, None otherwise
"""
return self._data_metadata_api.get_minimum_version(billing_tag=self.billing_tag)

[docs]
def latest_version(self) -> Optional[int]:
"""
Return latest version of the catalog available on the platform.

:return: latest version number of the catalog if present, None otherwise
"""
return self._data_metadata_api.get_latest_version(billing_tag=self.billing_tag)

[docs]
def list_versions(
self, start: Optional[int] = None, end: Optional[int] = None
) -> List[Version]:
"""
Return list of all catalog versions available on the platform, with one dict each.

:param start: An int to represent start version of catalog, None if unspecified
:param end: An int to represent end version of catalog, None if unspecified
:return: list of all catalog versions beginning at start + 1and continuing
through (and including) end
"""
if start is None:
start = -1
if end is None:
end = self.latest_version()
if end is None or start >= end:
return []
limit = 1000
versions = []
for i in range(start, end, limit):
list_versions_resp = self._data_metadata_api.list_versions(
i, i + limit if i + limit < end else end, billing_tag=self.billing_tag
)
if "versions" in list_versions_resp:
versions.extend([Version(version) for version in list_versions_resp["versions"]])
return versions

[docs]
def get_version(self, version: int) -> Version:
"""
Return details of a single version which is specified.

:param version: An int to specify the specific version
for which catalog details are needed
:return: details of the version specified
:raises ValueError:The value of the version is not an integer
or when the value is greater than the latest catalog version or less than 0
"""
if type(version) == int and version >= 0:
start = version - 1
end = version
latest = self.latest_version()
else:
raise ValueError(
f"Invalid Version value -> {version}, expected positive integer value."
)
if latest is None:
latest = -1
if start >= latest:
raise ValueError(f"Version {version} does not exist.")
get_version_resp = self._data_metadata_api.list_versions(
start, end, billing_tag=self.billing_tag
)
return Version(get_version_resp["versions"][0])

[docs]
def get_layers_version(self, catalog_version: int) -> Dict[str, int]:
"""
Return list of all available layers and layer version
for a given catalog version. Layer version is always
less or equal than the catalog version and it represent
the version when the layer was last updated.

:param catalog_version: An int to represent a catalog version
:return: all available layers with their version
"""
layer_versions_resp = self._data_metadata_api.get_layers_version(
catalog_version, billing_tag=self.billing_tag
)
if "layerVersions" in layer_versions_resp:
return {lv["layer"]: lv["version"] for lv in layer_versions_resp["layerVersions"]}
else:
return

[docs]
def compatible_versions(
self, catalog_dependencies: dict, limit: Optional[int] = 1
) -> Iterator[List[Version]]:
"""
Return compatible versions iterator which returns list of Versions.
Each Version object contains version and shared_dependencies attribute.
:param catalog_dependencies: The catalog dependencies we want to search for
:param limit: number of records in api response.
:yields: List of class: Version objects.
"""

is_next = True
url = None

while is_next:
resp_json = self._data_metadata_api.compatible_versions(
catalog_dependencies, limit, next=url
)
url = resp_json.get("next")
is_next = True if url else False
yield [Version(version) for version in resp_json["versions"]]

def _update_layer_in_catalog(self, catalog_details, layer_id: str, layers: list) -> Layer:
"""
Updates the catalog with :param:catalog_details and :param:layers.
:param catalog_details: Details of catalog.
:param layer_id: id of layer that needs to be added or updated in the catalog
:param layers: list of layers
:returns: object of added layer
"""
assert any([layer["id"] == layer_id for layer in layers])
self.platform.update_catalog(
self.hrn,
layers=layers,
)
updated_catalog = self.platform.get_catalog(catalog_details["hrn"])
return updated_catalog.get_layer(layer_id)

[docs]
def add_layer(
self,
id: str,
layer_type: str,
name: str,
summary: str,
description: str,
content_type: str,
**details,
) -> Layer:
"""
Adds a layer to the catalog and returns the object of the created layer.

:param id: ID of the layer to be added
:param layer_type: layer_type of the layer to be added
:param name: name of the layer to be added
:param summary: summary of the layer to be added
:param description: description of the layer to be added
:param content_type: content_type of the layer to be added
:param details: other details of the layer to be added
:return: the object of added layer
:raises CatalogConfigurationException: if the catalog already
contains a layer with ID :param:id
"""
details["id"] = id
details["layerType"] = layer_type
details["name"] = name
details["summary"] = summary
details["description"] = description
details["contentType"] = content_type
if self.has_layer(details["id"]):
raise CatalogConfigurationException(
"Layer already exists in catalog ".format(
details["id"], self.configuration.json["name"]
)
)
catalog_details = self.configuration.json
layers = catalog_details["layers"]
layers.append(details)
return self._update_layer_in_catalog(catalog_details, details["id"], layers)

[docs]
def update_layer(
self,
layer_id: str,
layer_type: str,
name: str,
summary: str,
description: str,
content_type: str,
**details,
):
"""
Updates the layer with :param:layer_id with given layer details.

:param layer_id: ID of the layer to be updated
:param layer_type: layer_type of the layer to be updated
:param name: name of the layer to be updated
:param summary: summary of the layer to be updated
:param description: description of the layer to be updated
:param content_type: content_type of the layer to be updated
:param details: other details of the layer to be updated
:raises CatalogConfigurationException: if the catalog has no layer
with the given ID :param:layer_id
"""
details["id"] = id
details["layerType"] = layer_type
details["name"] = name
details["summary"] = summary
details["description"] = description
details["contentType"] = content_type
if not self.has_layer(layer_id):
raise CatalogConfigurationException(
f"Layer {layer_id} does not exist in catalog {self.configuration.json['name']}" # noqa: E501, E713
)
catalog_details = self.configuration.json
layers = catalog_details["layers"]
for i in range(0, len(layers)):
if layers[i]["id"] == layer_id:
details["id"] = layer_id
layers[i] = details
break
self._update_layer_in_catalog(catalog_details, layer_id, layers)

[docs]
def get_layer(self, layer_id: str) -> Layer:
"""
Create and return a :class:Layer object for the given layer ID.

:param layer_id: layer ID
:return: the layer requested
:raises ValueError: Unknown layer id
"""
self._ensure_catalog_config_loaded()
assert self._layer_details is not None
if layer_id in self._layer_details:
layer_type = self._layer_details[layer_id]["layerType"]
layer_type_lower = layer_type.lower()
if layer_type_lower == LayerType.VERSIONED.name.lower():
return VersionedLayer(layer_id=layer_id, catalog=self)
elif layer_type_lower == LayerType.VOLATILE.name.lower():
return VolatileLayer(layer_id=layer_id, catalog=self)
elif layer_type_lower == LayerType.STREAM.name.lower():
return StreamLayer(layer_id=layer_id, catalog=self)
elif layer_type_lower == LayerType.INDEX.name.lower():
return IndexLayer(layer_id=layer_id, catalog=self)
elif layer_type_lower == LayerType.INTERACTIVEMAP.name.lower():
return InteractiveMapLayer(layer_id=layer_id, catalog=self)
elif layer_type_lower == LayerType.OBJECTSTORE.name.lower():
return ObjectStoreLayer(layer_id=layer_id, catalog=self)
else:
return Layer(layer_id=layer_id, catalog=self)
raise ValueError(f"Unknown layer id {layer_id}")

[docs]
def list_layers(self) -> List[Layer]:
"""
Return a list of all the layers present in the catalog.

:return: a list of layers
"""
self._ensure_catalog_config_loaded()
assert self._layer_details is not None
return [self.get_layer(id) for id in self._layer_details]

[docs]
def open_in_portal(self):
"""Open the catalog 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.hrn}")

[docs]
def modify_layer(self, layer_id: str, **details):
"""
Modify the catalog layer details.

:param layer_id: the identifier of the layer
:param details: Details of the catalog to be modified

The documentation for the API endpoint used can be found here:
|update_layer|

.. |update_layer| raw:: html

Update a catalog layer # noqa E501

Usage::

from here.platform import Platform
platform = Platform()
catalog = platform.get_catalog("hrn:here:data::olp-here:example") # doctest: +SKIP
catalog.modify_layer('layer-id', name='updated-name', description='updated-description') # doctest: +SKIP
"""
if not details:
return
response = self._data_config_api.patch_layer(self.hrn, layer_id=layer_id, data=details)
status_response, complete = self._data_config_api.get_catalog_status(response["href"])
while not complete:
logger.debug(f"status polling wait {self.platform._polling_wait} sec.")
time.sleep(self.platform._polling_wait)
status_response, complete = self._data_config_api.get_catalog_status(response["href"])
logger.info(f"Layer update for hrn: {self.hrn} finished successfully.")
self._invalidate_catalog_config()

[docs]
def delete_layer(self, layer_id: str):
"""Delete the catalog layer. Supported layer-types: volatile, stream, index.

:param layer_id: the identifier of the layer
:raises CatalogConfigurationException: in case the catalog configuration is not accepted
:raises LayerConfigurationException: in case the layer configuration is not accepted

Usage::

from here.platform import Platform
platform = Platform()
catalog = platform.get_catalog("hrn:here:data::olp-here:example") # doctest: +SKIP
catalog.delete_layer("layer-id") # doctest: +SKIP
"""
response = self._data_config_api.delete_layer(self.hrn, layer_id)
while True:
logger.debug(f"status polling wait {self.platform._polling_wait} sec.")
time.sleep(self.platform._polling_wait)
status_response, complete = self._data_config_api.get_catalog_status(
response["href"], billing_tag=self.billing_tag
)
if "status" in status_response and status_response["status"] == "pending":
continue
if "status" in status_response and status_response["status"] == "failure":
raise CatalogConfigurationException(status_response)
if "layers" in status_response:
for layer in status_response["layers"]:
if layer["id"] == layer_id:
raise LayerConfigurationException(status_response)
break
if complete:
raise LayerConfigurationException("Completed without layers response.")
logger.info(f"Layer deletion for hrn: {self.hrn} finished successfully.")
self._invalidate_catalog_config()

[docs]
def init_publication(
self,
layers: Union[List[str], List[Layer]],
dependencies: Optional[List[VersionDependency]] = [],
) -> Publication:
"""
Initialize a new publication, indicating which layers will be affected.

Pass the returned Publication to the functions that write to the layers.
Call Publication.complete() to complete the work.

Dependencies with other catalogs and relative versions can be published together with
the data. Submitting a new publication that affects one or more versioned layer
determines the creation of a new catalog version with the dependencies specified.

:param layers: which layers are affected by the publication
:param dependencies: list of dependencies to be use in case a new version is created
:return: a new publication valid for the layers specified # noqa
"""
layers_list = [self.get_layer(x) if isinstance(x, str) else x for x in layers]
publication = Publication(self, layers=layers_list, dependencies=dependencies)
return publication

[docs]
def retrieve_publication(self, publication_id: str) -> Publication:
"""
Retrieves an existing publication based on its ID.

:param publication_id: the ID for an already active publication.
:return: a publication for the ID.
:raises ValueError: the publication isn't in an initialized (or open) state or a layer from
the publication isn't present in the current catalog.
"""
publication_details = self._data_publish_api.get_publication(
publication_id, self.billing_tag
)
state = PublicationState(publication_details["details"]["state"])
if state != PublicationState.INITIALIZED:
raise ValueError(
f"Publication ID {publication_id} has an invalid state: {state.value}"
)

layer_ids: List[str] = publication_details.get("layerIds", [])
layers = [self.get_layer(layer_id) for layer_id in layer_ids]

version_dependencies: List[Dict[str, Union[bool, int, str]]] = publication_details.get(
"versionDependencies", []
)
if version_dependencies:
dependencies: Optional[List[VersionDependency]] = [
VersionDependency.from_dict(depend) for depend in version_dependencies
]
else:
dependencies = None

catalog_version = cast(Optional[int], publication_details.get("catalogVersion"))
return Publication(self, layers, dependencies, (publication_id, catalog_version))

[docs]
def write_versioned_layers(
self,
layers_write_info: Mapping[str, Mapping[Union[str, int], Union[str, Path, bytes]]],
version_dependencies: Optional[List[VersionDependency]] = None,
encode: bool = False,
adapter: Optional[Adapter] = None,
):
"""
Write data to versioned layers of a catalog, creating one single new version.

Data for each layer must be encoded according to the content type configured for the layer.

:param layers_write_info: a nested dict containing layer_ids, partition_ids and
encoded data in the form of bytes or file names or paths which will be uploaded
:param version_dependencies: A list of version dependencies
:param encode: whether to encode the data or upload raw bytes
:param adapter: the Adapter to transform the input data.
None to use the default adapter of the catalog.
:raises ValueError: in case one the layers specified is of the wrong type
"""

layer_ids: List[str] = list(layers_write_info.keys())
publication = self.init_publication(
layers=layer_ids,
dependencies=version_dependencies,
)
for layer_id, data in layers_write_info.items():
layer = self.get_layer(layer_id=layer_id)
if not isinstance(layer, VersionedLayer):
raise ValueError(f"Layer {layer.id} is not a versioned layer")
layer.write_partitions(
publication=publication, data=data, encode=encode, adapter=adapter
)
publication.complete()

[docs]
def write_stream_layers(
self,
layers_write_info: Mapping[str, Mapping[Union[str, int], Union[str, Path, bytes]]],
timestamp: Optional[int] = None,
encode: bool = False,
inline_data_limit: int = 819200,
adapter: Optional[Adapter] = None,
):
"""
Write data to stream layers of a catalog.

Data for each layer must be encoded according to the content type configured for the layer.

:param layers_write_info: a nested dict containing layer_ids, partition_ids and
encoded data in the form of bytes or file names or paths which will be uploaded.
:param timestamp: optional timestamp for all the messages:
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 encode: whether to encode the data or upload raw bytes
:param adapter: the Adapter to transform the input data.
None to use the default adapter of the catalog.
:raises ValueError: in case one the layers specified is of the wrong type
"""

publication is not needed anymore for writing stream layer

publication = self.init_publication(layers=layer_ids)

for layer_id, data in layers_write_info.items():
layer = self.get_layer(layer_id=layer_id)
if not isinstance(layer, StreamLayer):
raise ValueError(f"Layer {layer.id} is not a stream layer")
layer.write_stream(
data=data,
encode=encode,
timestamp=timestamp,
inline_data_limit=inline_data_limit,
adapter=adapter,
)

publication.complete()

[docs]
def write_volatile_layers(
self,
layers_write_info: Mapping[str, Mapping[Union[str, int], Union[str, Path, bytes]]],
):
"""
Write data to volatile layers of a catalog.

Data for each layer must be encoded according to the content type configured for the layer.

:param layers_write_info: a nested dict containing layer_ids, partition_ids and
encoded data in the form of bytes or file names or paths which will be uploaded.
:raises ValueError: in case one the layers specified is of the wrong type
"""

layer_ids: List[str] = list(layers_write_info.keys())
publication = self.init_publication(layers=layer_ids)
for layer_id, data in layers_write_info.items():
layer = self.get_layer(layer_id=layer_id)
if not isinstance(layer, VolatileLayer):
raise ValueError(f"Layer {layer.id} is not a volatile layer")
layer.write_partitions(publication=publication, data=data, encode=False)
publication.complete()

[docs]
def ingest_sdii(
self,
layer_id: str,
partition_id: Union[str, int],
path_or_data: Union[str, Path, bytes],
checksum: Optional[str] = None,
trace_id: Optional[str] = None,
) -> StreamIngestion:
"""
Ingest sensor data in SDII format in a stream layer.

:param layer_id: the id of the stream layer
:param partition_id: A unique id to specify the partition
:param path_or_data: Data to be uploaded in the partition. It must be less than 20 MB
:param checksum: A base64 encoded SHA-256 hash of the data
:param trace_id: A unique id to track your request and identify the message in a catalog
:returns: Trace and message IDs returned by the service
:raises ValueError: Data size is greater than the permissible limit of 20 MB

Example::

from here.platform import Platform
platform = Platform()
catalog = platform.get_catalog("catalog_hrn") # doctest: +SKIP
layer_id = "layer_id" # doctest: ´+SKIP
partition_id = "partition_id" # doctest: +SKIP
data = b'Some valid data in SDII format' # doctest: +SKIP
catalog.ingest_sdii(layer_id, partition_id, data) # doctest: +SKIP
"""
if isinstance(path_or_data, bytes):
data_size = len(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 > 20 1024 1024:
raise ValueError(
"ingest_sdii function supports writing SDII data to"
" stream layer up to 20 MB. For larger data sets, use"
" write_stream_layer functionality."
)

partition_id = str(partition_id)
if isinstance(path_or_data, bytes):
data = path_or_data
else:
with Partition.get_data_handler(path_or_data) as data_handler:
data = data_handler.read()

response = self._data_ingest_api.ingest_sdii(
layer_id=layer_id,
body=data,
checksum_header=checksum,
traceid_header=trace_id,
message_key_header=partition_id,
billing_tag=self.billing_tag,
)

return StreamIngestion(response)

[docs]
def write_index_layer(
self,
layer: Union[str, Layer],
path_or_data: Union[str, Path, bytes, "pd.DataFrame"],
timestamp: Optional[int] = None,
fields: Dict[str, Union[str, int, bool]] = ,
additional_metadata: Dict[str, str] = ,
encode: bool = True,
adapter: Optional[Adapter] = None,
**kwargs,
):
"""
Write data to an index layer.

:param layer: the layer or ID of the layer
:param path_or_data: a file name or path or bytes with encoded content.
Alternatively, DataFrame is supported only for layers with content-type
application/x-parquet when using a GeoPandasAdapter.
: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 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
:raises ValueError: in case the specified layer is of the wrong type

Example::

from here.platform import Platform
platform = Platform()
catalog = platform.get_catalog(hrn="hrn:here:data::olp-here:write-function-test")
layer_id = "sample_index_layer"
file_path = "~/sample_data.parquet"
fields = {"ingestionTime": 1114, "partition_id": 343433}
catalog.write_index_layer(layer_id, path_or_data=file_path, # doctest: +SKIP
fields=fields, encode=False) # doctest: +SKIP
"""

layer = layer if isinstance(layer, Layer) else self.get_layer(layer)
if not isinstance(layer, IndexLayer):
raise ValueError(f"Layer {layer.id} is not an index layer")

layer.write_single_partition(
data=path_or_data,
timestamp=timestamp,
fields=fields,
additional_metadata=additional_metadata,
encode=encode,
adapter=adapter,
**kwargs,
)

[docs]
def grant_access(self, entity_type: str, entity_id: str, action: str):
"""
Grants access to a catalog to an entity.

:param entity_type: Type of entity app, user or group.
:param entity_id: a unique identifier for the given entity type.
:param action: an action type to grant access - read, write, manage.
"""
self._aaa_auth_api.add_grant(
resource_hrn=self.hrn,
entity_id=entity_id,
entity_type=entity_type,
action_id=f"{action}Resource",
)

[docs]
def revoke_access(self, entity_type, entity_id, action):
"""
Revokes access to a catalog from an entity.

:param entity_type: Type of entity app, user or group.
:param entity_id: a unique identifier for the given entity type.
:param action: an action type to grant access - read, write, manage.
"""
self._aaa_auth_api.remove_grant(
resource_hrn=self.hrn,
entity_id=entity_id,
entity_type=entity_type,
action_id=f"{action}Resource",
)

[docs]
def share(self, entity_type: str, entity_id: str):
"""
Share a catalog with an entity.

:param entity_type: Type of entity app, user or a group.
:param entity_id: a unique identifier for the given entity type.
"""
self._aaa_auth_api.share_authorization(
resource_hrn=self.hrn, entity_type=entity_type, entity_id=entity_id
)