Skip to main content

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

kafka_client_decorator-1.8.tar.gz (5.2 kB view hashes)

Uploaded Source

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page