A wrapper for confluent-kafka producer and consumer
Project description
A wrapper for kafka producer and consumer that can be used as decorator for a function which can keep consuming data, process this data and broadcast it to next topics/queues.
This uses confluent-kafka python package to create prooducer, consumer and then wraps it. So, big thanks to them!
Installation
$ pip install kafka-client-decorator
Usage
Define your function how you want to process the data and then decorate it.
from kafka_client_decorator.kafka_client import KafkaClient
@KafkaClient(bootstrap_servers, security_protocol, sasl_username, sasl_password).consumer_producer(consumer_from_topic='my-topic-1', group_id='pdf', produce_to_topic=['my-topic-2'])
def process_data(data = None):
# Call your driver modules here to process the data
result = Driver(data)
return result
NOTE: If you want the your driver result to be pushed to next topic/queue, you can simply pass produce_to_topic as arg in decorator 'consumer_prodcuer' method.
To only produce to topic(s) -
from kafka_client_decorator.client_producer import ClientProducer
producer = ClientProducer(bootstrap_servers, security_protocol, sasl_username, sasl_password)
prodcuer.produce_to_broker(data, topics_list)
NOTE: If your kafka broker does not uses SASL or SSL protocol, no need to pass 'sasl_username' and 'sasl_password'.
NOTE: If you want to work with multiple partitions in kafka, you can use below method to produce (it provides custom unique key to be sent with the message)
prodcuer.produce_to_broker_with_key(data, topic_list)
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
File details
Details for the file kafka_client_decorator-1.8.tar.gz
.
File metadata
- Download URL: kafka_client_decorator-1.8.tar.gz
- Upload date:
- Size: 5.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/5.1.0 CPython/3.12.2
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | ad6be76cf21f057cda12b222a28b66c665061446ef7c9330a223584c21fa9c0e |
|
MD5 | 25fab7d468062f20eebf5ac11fe91b6a |
|
BLAKE2b-256 | 386c41d95d8be84c8bf74a48eba08a21480edd712b254a0284f2d2c9021ae07c |