pswamp.streaming.kafka_extras

Submodules

Classes

`KafkaConsumer`	Wrapper for Kafka consumer.
`KafkaProducer`	Wrapper for Kafka producer.

Functions

`consumer_seek_relative_offset`(consumer, relative_offset)
`create_topic`(name, kafka_kwargs, **kwargs)
`get_last_message_from_topic`(kafka_kwargs, topic)
`send_to_kafka_topic`(kafka_kwargs, topic, msg)

Package Contents

class pswamp.streaming.kafka_extras.KafkaConsumer(*args, use_nqkafka=False, **kwargs)

Wrapper for Kafka consumer.

Parameters:

use_nqkafka (bool, optional) – Determines whether self.instance is a NQKafkaConsumer (from the nqkafka package, can be thought of as a “mock”) or actual KafkaConsumer (from the kafka-python package), which connects to a Kafka server. Defaults to False.
*args – Variable length argument list.
**kwargs – Arbitrary keyword arguments.

instance

Determined from input parameters.

Type:: nqkafka.NQKafkaConsumer or kafka.KafkaConsumer

instance

__getattr__(name)

Inherits the same attributes as the instance.

Parameters:: name (_type_) – Name of the attribute.
Returns:: The attribute of the instance.

__iter__()

The iterator is the same as the iterator of the instance.

Returns:: The instance.

class pswamp.streaming.kafka_extras.KafkaProducer(*args, use_nqkafka=False, **kwargs)

Wrapper for Kafka producer.

Parameters:

use_nqkafka (bool, optional) – Determines whether self.instance is a NQKafkaProducer (from the nqkafka package, can be thought of as a “mock”) or actual KafkaProducer (from the kafka-python package), which connects to a Kafka server. Defaults to False.
*args – Variable length argument list.
**kwargs – Arbitrary keyword arguments.

instance

Determined from input parameters.

Type:: nqkafka.NQKafkaProducer or kafka.KafkaProducer

instance

__getattr__(name)

Inherits the same attributes as the instance.

Parameters:: name (_type_) – Name of the attribute.
Returns:: The attribute of the instance.

__iter__()

The iterator is the same as the iterator of the instance.

Returns:: The instance.

pswamp.streaming.kafka_extras.consumer_seek_relative_offset(consumer, relative_offset)

pswamp.streaming.kafka_extras.create_topic(name, kafka_kwargs, **kwargs)

pswamp.streaming.kafka_extras.get_last_message_from_topic(kafka_kwargs, topic)

pswamp.streaming.kafka_extras.send_to_kafka_topic(kafka_kwargs, topic, msg)