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 details)

Uploaded Source

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

Hashes for kafka_client_decorator-1.8.tar.gz
Algorithm Hash digest
SHA256 ad6be76cf21f057cda12b222a28b66c665061446ef7c9330a223584c21fa9c0e
MD5 25fab7d468062f20eebf5ac11fe91b6a
BLAKE2b-256 386c41d95d8be84c8bf74a48eba08a21480edd712b254a0284f2d2c9021ae07c

See more details on using hashes here.

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