diff --git a/creds/kafka/__init__.py b/creds/kafka/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/queue/kafka.py b/utils_v2/queue/kafka.py new file mode 100644 index 0000000..eb134d6 --- /dev/null +++ b/utils_v2/queue/kafka.py @@ -0,0 +1,546 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Friday, 20th Dec., 2024 + + OBJECTIVE: + + To work with Kafka in a synchronous way. This is a translation of the wrapper originally made for asynchronous + operation. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# To use Kafka: +from confluent_kafka import Producer +from confluent_kafka import Consumer + +# For working with JSON strings: +from utils_v2.string import json +from utils_v2.serialization.json_serializer import JSONSerializer + +# For debugging: +from icecream import IceCreamDebugger + +# To work with datatypes: +from typing import List, Literal + +# For SSL security: +import ssl + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +def create_config( + bootstrap_servers: str | List[str], + group_id: str | None = None, + security_protocol: Literal["PLAINTEXT", "SSL"] = "PLAINTEXT", + ca_file: str | None = None, + cert_file: str | None = None, + key_file: str | None = None, + client_id: str | None = None +) -> dict: + + """ + Creates the config required for Confluent-Kafka's library. + :param bootstrap_servers: The addresses of the Kafka brokers. + :param group_id: The identifier of the group of consumers. Irrelevant for producers. + :param security_protocol: What sort of security protocol to use. + :param ca_file: Needed for 'SSL' security protocol. + :param cert_file: Needed for 'SSL' security protocol. + :param key_file: Needed for 'SSL' security protocol. + :param client_id: An identifier for one producer/consumer. Useful for debugging later. + :return: The dictionary that needs to be passed as the 'conf' param when creating the producer/consumer. + """ + + # Start with the bare minimum: + if not isinstance(bootstrap_servers, list): bootstrap_servers = [bootstrap_servers] + config = { + "bootstrap.servers": ",".join(bootstrap_servers), + "security.protocol": security_protocol + } + + # Add the SSL security details: + if security_protocol == "SSL": + config["ssl.ca.location"] = ca_file + config["ssl.certificate.location"] = cert_file + config["ssl.key.location"] = key_file + + # Specific to consumers: + if group_id: config["group.id"] = group_id + + # Add an identifier for debugging: + if client_id: config["client.id"] = client_id + + # Done here: + return config + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +class ProducerKafka: + + def __init__( + self, + topic: str, + config: dict, + serializer = None, + debug: bool = True, + debug_prefix = "Kafka (P) | " + ): + + """ + Create a Kafka Producer. + :param topic: The topic to produce on. + :param config: The configuration as expected by Confluent-Kafka library. + :param serializer: The serializer to use. + :param debug: Whether, or not, you want to print the debug strings. + :param debug_prefix: The prefix to use while debugging. + """ + + # Initialize the debugger: + self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) + if not debug: self.__printer.disable() + + # initialize the Kafka producer: + self.__topic = topic + self.__config = config + self.__producer = None + self.__connected = False + self.__serializer = serializer or JSONSerializer() + + def enable_debug(self): + self.__printer.enable() + + def disable_debug(self): + self.__printer.disable() + + def connect(self) -> bool: + + """ + Connects to the Kafka server if not connected. + :return: True or False based on the success of the operation. + """ + + if not self.__connected: + try: + self.__producer = Producer(self.__config) + self.__connected = True + except Exception as exception: self.__printer(exception) + return self.__connected + + def ensure_connection(self): + + """ + Connects to the Kafka server if not connected. + :return: True or False based on the success of the operation. + """ + + if not self.__connected: self.connect() + return self.__connected + + def close(self): + + """ + Terminates the connection. + :return: None. + """ + + if self.__connected: + try: + self.__producer.flush() + self.__printer("Producer closed!") + self.__connected = False + except Exception as exception: self.__printer(exception) + return not self.__connected + + def produce( + self, + value, + key = None, + topic = None, + encoding = "utf-8", + callback = None + ) -> bool: + + """ + Sends one message to the Kafka server on the topic that has been set for this instance. + :param value: The message to send. + :param key: The key to use when you want the messages to follow an order. + :param topic: A custom topic for this message, else the topic defined during the creation of this instance will + be used by default. + :param encoding: The encoding format. + :param callback: The function to call for delivery reports. + :return: True or False based on the success of the operation. + """ + + # Ensure connectivity to the server. + # If not connected, return with failure immediately. + if not self.ensure_connection(): return False + + try: + + # Send the message: + self.__producer.produce( + topic = topic or self.__topic, + value = self.__serializer.serialize(data = value, encoding = encoding), + key = key, + callback = callback + ) + + # Return with success if no exception occurred: + return True + + # Return with failure if something went wrong: + except Exception as exception: + self.__printer(exception, self.__topic, type(value), value) + return False + + +# --------------------------------------------------------------------------------------------------------------------- + + +# class ConsumerKafka: +# +# def __init__( +# self, +# topic, +# serializer = None, +# debug = True, +# debug_prefix = "Kafka (C) | ", +# **kwargs +# ): +# +# """ +# Create a Kafka Consumer. +# :param topic: The topic to consumer on. +# :param serializer: The serializer to use. +# :param debug: Whether, or not, you want to print the debug strings. +# :param debug_prefix: The prefix to use while debugging. +# :param kwargs: Any configuration parameters for the Kafka instances. +# """ +# +# # Initialize the debugger: +# self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) +# if not debug: self.__printer.disable() +# +# # initialize the Kafka producer: +# self.__topic = topic +# self.__kwargs = kwargs +# self.__consumer = None +# self.__connected = False +# self.__serializer = serializer or JSONSerializer() +# +# # For establishing connection: +# self.__exclusive_semaphore = asyncio.Semaphore(1) +# +# def enable_debug(self): +# self.__printer.enable() +# +# def disable_debug(self): +# self.__printer.disable() +# +# async def connect(self): +# +# """ +# Connects to the Kafka server if not connected. +# :return: True or False based on the success of the operation. +# """ +# +# async with self.__exclusive_semaphore: +# if not self.__connected: +# try: +# self.__consumer = AIOKafkaConsumer(self.__topic, **self.__kwargs) +# await self.__consumer.start() +# self.__connected = True +# except Exception as exception: self.__printer(exception) +# return self.__connected +# +# async def ensure_connection(self): +# +# """ +# Connects to the Kafka server if not connected. +# :return: True or False based on the success of the operation. +# """ +# +# if not self.__connected: await self.connect() +# return self.__connected +# +# async def close(self): +# +# """ +# Terminates the connection. +# :return: None. +# """ +# +# if self.__connected: +# try: +# await self.__consumer.stop() +# self.__printer("Consumer closed!") +# self.__connected = False +# except Exception as exception: self.__printer(exception) +# +# async def consume(self, count = 1, timeout = 0.05, encoding = "utf-8"): +# +# """ +# Get messages from the Kafka server. +# :param count: The number of messages to get from the Kafka server. +# :param timeout: The time in seconds to wait for retrieval. +# :param encoding: The encoding to use. +# :return: The messages that were received. If no messages are available, an empty list will be returned. +# """ +# +# # Ensure connectivity to the server. +# # If not connected, return with failure immediately. +# if not await self.ensure_connection(): return [] +# +# # Make a variable that will hold the final results: +# messages = [] +# +# try: +# +# # Read some messages: +# results = await self.__consumer.getmany( +# max_records = max(1, count), +# timeout_ms = int(timeout * 1_000) +# ) +# +# # Format the received messages: +# if results: +# for topic_partition, records in results.items(): +# for record in records: +# record_dict = record.__dict__ +# record_dict["value"] = self.__serializer.deserialize( +# data = record_dict["value"], +# encoding = encoding +# ) +# messages.append(record_dict) +# +# # Debugging print if something went wrong: +# except Exception as exception: self.__printer(exception) +# +# # Done here: +# return messages + + +# --------------------------------------------------------------------------------------------------------------------- + + +# class BidirectionalKafka: +# +# # The 'roles' that the instance can take. +# # The master talks on the channel (topic) that the slave listens on and vice versa. +# # Master-Slave is only for deciding who talks on which channel and who listens on which. +# # In a two-party system, one must be the master, the other must be the slave. +# # There are no extra privileges that the master enjoys. The naming convention was borrowed from common protocols +# # used in electronics (like I2C). +# ROLE_MASTER = 1 +# ROLE_SLAVE = 0 +# +# def __init__( +# self, +# role, +# topic, +# ack_topic: str = None, +# group: str = None, +# serializer = None, +# debug = True, +# debug_prefix = "Kafka (B) | ", +# **kwargs +# ): +# +# """ +# Creates a walkie-talkie type setup to use Kafka in a bidirectional manner. Fo more information on all the +# individual methods, please read through the doc-strings of the component classes 'ProducerKafka', and +# 'ConsumerKafka'. +# :param role: Select from "ROLE_MASTER" and "ROLE_SLAVE". Between the two parties that are talking, one will be +# the master and the other will be the slave. The channel that the master uses to speak will the one the slave +# uses to listen, and vice versa. +# :param topic: The topic to communicate on. Will be the same between the master and the slave. +# :param ack_topic: Explicitly provide this for the second channel, or it will be created from the name of the +# topic itself. Will be the same between the master and the slave. +# :param group: The group to assign the instance to. +# :param debug: Whether, or not, you want to print the debug strings. +# :param debug_prefix: The prefix to use while debugging. +# :param kwargs: Any configuration parameters for the Kafka instances. +# """ +# +# # Not down the basic variables: +# self.__role = role +# self.__topic = topic +# self.__ack_topic = ack_topic or topic + "Ack" +# self.__group = group +# +# # In case the current instance is the master, +# # it will talk on "topic", and listen on "ack_topic": +# if self.__role == self.ROLE_MASTER: +# self.__producer_kwargs = kwargs.copy() +# self.__producer = ProducerKafka( +# topic = self.__topic, +# serializer = serializer, +# debug = debug, +# debug_prefix = debug_prefix.strip() + " (P) | ", +# **self.__producer_kwargs +# ) +# self.__consumer_kwargs = kwargs.copy() +# self.__consumer_kwargs["group_id"] = self.__group +# self.__consumer = ConsumerKafka( +# topic = self.__ack_topic, +# group = group, +# serializer = serializer, +# debug = debug, +# debug_prefix = debug_prefix.strip() + " (C) | ", +# **self.__consumer_kwargs +# ) +# +# # On the other hand, if the current instance is a slave, +# # It will listen on "topic", and talk on "ack_topic": +# else: +# self.__producer_kwargs = kwargs.copy() +# self.__producer = ProducerKafka( +# topic = self.__ack_topic, +# serializer = serializer, +# debug = debug, +# debug_prefix = debug_prefix.strip() + " (P) | ", +# **self.__producer_kwargs +# ) +# self.__consumer_kwargs = kwargs.copy() +# self.__consumer_kwargs["group_id"] = self.__group +# self.__consumer = ConsumerKafka( +# topic = self.__topic, +# group = group, +# serializer = serializer, +# debug = debug, +# debug_prefix = debug_prefix.strip() + " (C) | ", +# **self.__consumer_kwargs +# ) +# +# def enable_debug(self): +# self.__producer.enable_debug() +# self.__consumer.enable_debug() +# +# def disable_debug(self): +# self.__producer.disable_debug() +# self.__consumer.disable_debug() +# +# async def ensure_connection(self): +# await self.__producer.ensure_connection() +# await self.__consumer.ensure_connection() +# +# async def close(self): +# await self.__producer.close() +# await self.__consumer.close() +# +# async def produce(self, message, encoding = "utf-8"): +# return await self.__producer.produce(message, encoding = encoding) +# +# async def consume(self, count = 1, timeout = 0.05, encoding = "utf-8"): +# return await self.__consumer.consume(count = count, timeout = timeout, encoding = encoding) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + import datetime + import time + + def producer_test( + count: int = 10, + interval: float = 1.0 + ): + + # Create and connect the producer: + producer = ProducerKafka( + topic = "kft_file_upload", + config = create_config( + bootstrap_servers = "del.ditscentre.in:9092", + security_protocol = "SSL", + cert_file = r"../../creds/kafka/fullchain.pem", + key_file = r"../../creds/kafka/privkey.pem" + ) + ) + producer.connect() + print("PRODUCER READY!") + + # Send the message a number of times: + for message_no in range(count): + my_message = { + "ts": datetime.datetime.now().timestamp(), + "msgNo": message_no, + "payload": { + "name": "Bhopli", + "color": "orange", + "age": "just a baby", + } + } + success = producer.produce(my_message) + print("PRODUCED:", success) + time.sleep(interval) + + # Ensure a graceful close: + producer.close() + + + # Run the test code: + producer_test(count = 5)