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

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

.. |stream_api_reference| raw:: html

Stream API Reference # noqa E501
"""

from typing import List, Optional

from here.platform.api.base_api import BaseApi
from here.platform.auth import Auth
from here.platform.config import ApplicationConfig, PlatformConfig
from here.platform.exceptions import PlatformException

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

The stream service provides the ability to consume data from a stream layer.
With this service you can subscribe to a stream layer and consume messages.
"""

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

:param base_url: a Stream API Url obtained from Lookup API for a given hrn.
:param platform_config: a mandatory :class:PlatformConfig object to provide
configuration information for the API.
:param application_config: a mandatory :class:ApplicationConfig object to provide
configuration information for the API.
:param auth: an Authentication instance.
:param proxies: an optional proxy configuration. Defaults to the environment proxy
configuration.
"""
super(DataStreamApi, self).init(
platform_config=platform_config,
application_config=application_config,
auth=auth,
proxies=proxies,
)
self.base_url = base_url

[docs]
def subscribe(
self,
layer_id: str,
subscription_id: Optional[str] = None,
mode: Optional[str] = "serial",
consumer_id: Optional[str] = None,
kafka_consumer_properties: Optional[dict] = None,
) -> dict:
"""
Enable message consumption for given layer ID.

:param layer_id: a string with the layer ID of stream layer.
:param subscription_id: subscription id returned from the subscribe call.
:param mode: The subscription mode of this subscriptionId.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: This is the same as Kafka Consumer properties with
some default values changed.
:return: dict containing node base url and subscription id.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/subscribe"
url = self.format_url(self.base_url, path)
params = {"mode": mode, "subscriptionId": subscription_id, "consumerId": consumer_id}
if kafka_consumer_properties:
resp = self.post(url, params=params, data=kafka_consumer_properties)
else:
resp = self.post(url, params=params)
if resp.status_code == 201:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def delete_subscription(
self,
node_base_url: str,
layer_id: str,
subscription_id: str,
mode: str = "serial",
strict: bool = False,
) -> bool:
"""Disable message consumption for given layer ID.

:param node_base_url: node base url returned from subscribe call.
:param layer_id: a string with the layer ID.
:param subscription_id: subscription id returned from the subscribe call.
:param strict: If the subscription doesn't exist, strict=True will raise a
PlatformException while strict=False will not.
:param mode: The subscription mode of this subscriptionId.By default value is serial.
:return: True if the subscription exists and was deleted, False if it doesn't exist.
:raises PlatformException: If platform responds with an HTTP error.
"""
success_codes = [200]
if not strict:
success_codes.append(404)

path = f"/layers/{layer_id}/subscribe"
url = self.format_url(node_base_url, path)
params = {"mode": mode, "subscriptionId": subscription_id}
resp = self.delete(url, params=params)
if resp.status_code in success_codes:
return resp.status_code != 404
else:
raise PlatformException(resp)

[docs]
def consume_data(
self, node_base_url: str, layer_id: str, subscription_id: str, mode: str = "serial"
):
"""
Consume data for given layer ID and subscription ID.

:param node_base_url: node base url returned from subscribe call.
:param layer_id: a string with the layer ID
:param subscription_id : subscription id returned from the subscribe call.
:param mode: The subscription mode of this subscriptionId. By default value is serial.
:return: messages from the stream layer.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/partitions"
url = self.format_url(node_base_url, path)
params = {"mode": mode, "subscriptionId": subscription_id}
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def endpoint(self, layer_id: str, internal: bool = False) -> dict:
"""
Get a list of Kafka Broker URL's, client ID,
and Group ID prefix. Type is set to a consumer.

:param layer_id: a string with the layer ID of stream layer.
:param internal: Specifies whether internal bootstrap servers
will be returned in the response.
:return: Dict containing Kafka Broker URL's, client ID,
and Group ID prefix.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/endpoint"
url = self.format_url(self.base_url, path)
params = {"internal": internal}
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def kafka_endpoint(self, layer_id: str, kafka_type: str) -> dict:
"""
Exposes the endpoint to use to produce and consume the data of the layer directly,
per type of user - consumer or producer. It returns a list of Kafka Broker URL's,
client ID, and Group ID prefix.

:param layer_id: a string with the layer ID of stream layer.
:param kafka_type: a string denoting type as a producer or consumer.
:return: Dict containing Kafka Broker URL's, client ID,
and Group ID prefix.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/{kafka_type}/endpoint"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def commit_offsets(
self,
node_base_url: str,
layer_id: str,
subscription_id: str,
offsets: List[dict],
mode: str = "serial",
):
"""
Commit offsets for a streaming layer and a given subscription.

After reading data, you should commit the offset of the last message read
from each partition so that your application can resume reading new messages
from the correct partition in the event that there is a disruption to the subscription,
such as an application crash. An offset can also be useful if you delete a subscription
then recreate a subscription for the same layer, because the new subscription
can start reading data from the offset.

Only one offset per partition is allowed.

:param node_base_url: node base url returned from subscribe call.
:param layer_id: a string with the layer ID
:param subscription_id : subscription id returned from the subscribe call.
:param offsets: list of offsets
[{"partition": <Partition ID>,"offset": <Offset Number>}, ...]
:param mode: The subscription mode of this subscriptionId. By default value is serial.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/layers/{layer_id}/offsets"
url = self.format_url(node_base_url, path)
params = {"mode": mode, "subscriptionId": subscription_id}
headers = self.headers
headers["content-type"] = "application/json"
data = dict(offsets=offsets)
resp = self.put(url, data=data, params=params, headers=headers)
if resp.status_code == 200:
return
else:
raise PlatformException(resp)

[docs]
def seek_to_offset(
self,
node_base_url: str,
layer_id: str,
subscription_id: str,
offsets: List[dict],
mode: str = "serial",
):
"""
Seek a stream offset for a streaming layer and a given subscription.

:param node_base_url: node base url returned from subscribe call.
:param layer_id: a string with the layer ID
:param subscription_id : subscription id returned from the subscribe call.
:param offsets: list of offsets [{"partition": <Partition ID>, "offset": <Offset Number>}, ...]
:param mode: The subscription mode of this subscriptionId. By default value is serial.
:raises PlatformException: If platform responds with an HTTP error.
"""
params = {"mode": mode, "subscriptionId": subscription_id}
headers = self.headers
headers["content-type"] = "application/json"
path = f"/layers/{layer_id}/seek"
url = self.format_url(node_base_url, path)
data = dict(offsets=offsets)
resp = self.put(url, data=data, params=params, headers=headers)
if resp.status_code == 200:
return
else:
raise PlatformException(resp)