This module contains a DataStreamApi class to perform API operations.

The HERE API reference documentation used in this module can be found here: Stream API Reference # noqa E501

class here.platform.api.data_stream_api.DataStreamApi(base_url: str, auth: Auth | None, platform_config: PlatformConfig, application_config: ApplicationConfig, proxies: dict | None = None)[source]#

Bases: 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.

commit_offsets(node_base_url: str, layer_id: str, subscription_id: str, offsets: List[dict], mode: str = 'serial')[source]#

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.

Parameters:

  • node_base_url – node base url returned from subscribe call.

  • 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>}, …]

Parameters:

mode – The subscription mode of this subscriptionId. By default value is serial.

Raises:

PlatformException – If platform responds with an HTTP error.

consume_data(node_base_url: str, layer_id: str, subscription_id: str, mode: str = 'serial')[source]#

Consume data for given layer ID and subscription ID.

Parameters:

  • node_base_url – node base url returned from subscribe call.

  • 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.

delete_subscription(node_base_url: str, layer_id: str, subscription_id: str, mode: str = 'serial', strict: bool = False) → bool[source]#

Disable message consumption for given layer ID.

Parameters:

  • node_base_url – node base url returned from subscribe call.

  • layer_id – a string with the layer ID.

  • subscription_id – subscription id returned from the subscribe call.

  • strict – If the subscription doesn’t exist, strict=True will raise a PlatformException while strict=False will not.

  • mode – The subscription mode of this subscriptionId.By default value is serial.

Returns:

True if the subscription exists and was deleted, False if it doesn’t exist.

Raises:

PlatformException – If platform responds with an HTTP error.

endpoint(layer_id: str, internal: bool = False) → dict[source]#

Get a list of Kafka Broker URL’s, client ID, and Group ID prefix. Type is set to a consumer.

Parameters:

  • layer_id – a string with the layer ID of stream layer.

  • internal – Specifies whether internal bootstrap servers will be returned in the response.

Returns:

Dict containing Kafka Broker URL’s, client ID, and Group ID prefix.

Raises:

PlatformException – If platform responds with an HTTP error.

kafka_endpoint(layer_id: str, kafka_type: str) → dict[source]#

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.

Parameters:

  • layer_id – a string with the layer ID of stream layer.

  • kafka_type – a string denoting type as a producer or consumer.

Returns:

Dict containing Kafka Broker URL’s, client ID, and Group ID prefix.

Raises:

PlatformException – If platform responds with an HTTP error.

seek_to_offset(node_base_url: str, layer_id: str, subscription_id: str, offsets: List[dict], mode: str = 'serial')[source]#

Seek a stream offset for a streaming layer and a given subscription.

Parameters:

  • node_base_url – node base url returned from subscribe call.

  • 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>}, …]

Parameters:

mode – The subscription mode of this subscriptionId. By default value is serial.

Raises:

PlatformException – If platform responds with an HTTP error.

subscribe(layer_id: str, subscription_id: str | None = None, mode: str | None = 'serial', consumer_id: str | None = None, kafka_consumer_properties: dict | None = None) → dict[source]#

Enable message consumption for given layer ID.

Parameters:

  • layer_id – a string with the layer ID of stream layer.

  • subscription_id – subscription id returned from the subscribe call.

  • mode – The subscription mode of this subscriptionId.By default value is serial.

  • 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.

  • kafka_consumer_properties – This is the same as Kafka Consumer properties with some default values changed.

Returns:

dict containing node base url and subscription id.

Raises:

PlatformException – If platform responds with an HTTP error.