resistant-kafka-avataa 0.2

No project description provided

Resistant Kafka

Resistant Kafka is a Python library designed to simplify and stabilize interactions with Apache Kafka, both as a producer and a consumer.

Features

🔌 Easy integration into any Python service

To connect a consumer or producer, you just need to create one instance of the corresponding class: ConsumerInitializer or ProducerInitializer.

💾 Redis integration

We provide an easy way to connect to Redis. It is used for storing error messages and retrieving full information about each message.

🧾 Built-in logging of errors and events

Output of basic logs of connection of topic handlers in one consumer, as well as output of messages about successful sending of a message from the producer to certain topics.

🛡️ Resilience against consumer-side crashes

If an exception raises in the processor when reading a specific topic, by default, a detailed log about the dropped message will be issued and the consumer will continue its work.

In case you need to stop reading topic and raise exception - this option has also been added.

🧩 Handler creation for each topic in your service (Asynchronous)

One of the problems of working in the consumer is the case where the service reads several topics at the same time and this happens synchronously and in one handler.

We solved this problem!

By adding asynchronous reading of topics and adding the ability to read topics independently of each other. Even if one of them crashes (a crash will occur if you set the raise_error=True attribute in the kafka_processor) - the other handler will continue its work.

Also in this case it is very easy to separate the logic of processing messages of different topics if their keys, message type differ from each other.

Consumer Initializer

First Step. Add enviroments

Using the ConsumerConfig scheme you can configure the message reading handler in your service.

If reading of several topics is expected, then a more convenient way is to assemble common settings for connecting to Kafka and add them to the handler class (for example, to KafkaMessageProcessor) by **kwargs .

EXAMPLE:

from resistant_kafka.consumer_schemas import ConsumerConfig

common_consumer_config = dict(
        bootstrap_servers='HOST:PORT',
        group_id='CONSUMER_NAME',
        auto_offset_reset='latest',
        enable_auto_commit=False,
)

process_task_1 = KafkaMessageProcessor(
    config=ConsumerConfig(
        topic_to_subscribe='TOPIC_NAME_1',
        processor_name='KafkaMessageProcessor1',
        **common_consumer_config
    )
)

process_task_2 = KafkaMessageProcessor(
    config=ConsumerConfig(
        topic_to_subscribe='TOPIC_NAME_2',
        processor_name='KafkaMessageProcessor2',
        **common_consumer_config
    )
)

Second Step. Add processor

Processor is a class-handler of a specific topic. It allows to perform CRUD operations on received messages from a given topic independently of other processors.

⚠️ The name of the main method "process" is reserved and is required for installation.⚠️

⚠️The decorator "kafka_processor" is also required ⚠️, which is responsible for the operation of the message stream and the stable operation of the main method "process". It has the attribute raise_error, which allows to raise an error, while the work of a specific handler will be stopped.

EXAMPLE:

from resistant_kafka.consumer import ConsumerInitializer, kafka_processor

class KafkaMessageProcessor1(ConsumerInitializer):
    def __init__(
            self, config: ConsumerConfig
    ):
        super().__init__(config=config)
        self._config = config

    # required decorator
    # to raise error, instead logging @kafka_processor(raise_error=True)
    @kafka_processor() 
    async def process(self):  # required service name 
        message = await self.get_message(consumer=self._consumer)
        if self.message_is_empty(message=message, consumer=self._consumer):
            return
        
        message_key = message.key().decode("utf-8")
        message_value = message.value().decode("utf-8")
        
        # here you can process your message
        print('-----------------------------')
        print('KEY', message_key)
        print('VALUE', message_value)
        print('CONSUMER', self._config.topic_to_subscribe)
        print('-----------------------------')

Third Step. Initialization

In order to start topic processors, you should use the "init_kafka_connection" method, to which you need to pass a list of instances of the processor-classes.

EXAMPLE:

from resistant_kafka.consumer import init_kafka_connection

process_task_1 = KafkaMessageProcessor1(
    config=ConsumerConfig(
        topic_to_subscribe='TOPIC_NAME_1',
        processor_name='KafkaMessageProcessor1',
        **consumer_config
    )
)

process_task_2 = KafkaMessageProcessor2(
    config=ConsumerConfig(
        topic_to_subscribe='TOPIC_NAME_2',
        processor_name='KafkaMessageProcessor2',
        **consumer_config
    )
)

init_kafka_connection(
    tasks=[process_task_1, process_task_2]
)

️⚠️In the way, where you have already created loop - use method "process_kafka_connection" ⚠️

import asyncio
from resistant_kafka.consumer import process_kafka_connection

asyncio.create_task(process_kafka_connection([inventory_changes_processor]))

Additional Step. Add security

To add security, you should set attribute security_config using class KafkaSecurityConfig

from resistant_kafka.common_schemas import KafkaSecurityConfig
from resistant_kafka.consumer_schemas import ConsumerConfig

security_config = KafkaSecurityConfig(
    oauth_cb=method_to_get_token,
    security_protocol='SASL_PLAINTEXT',
    sasl_mechanisms='OAUTHBEARER'
)

consumer_config = ConsumerConfig(
        bootstrap_servers='HOST:PORT',
        group_id='CONSUMER_NAME',
        auto_offset_reset='latest',
        enable_auto_commit=False,
        
        security_config=consumer_config
)

Additional Step. Integrate Redis

    pip install redis=4.5.4

from resistant_kafka_avataa.common_schemas import RedisStoreConfig

redis_store_config = RedisStoreConfig(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,
)

process_task = KafkaMessageProcessor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer1',
        processor_name='KafkaProcessor1',
        bootstrap_servers='kafka.avataa.dev:9093',
        group_id='LocalTester1',
        auto_offset_reset='latest',
        enable_auto_commit=False,

        redis_store_config=redis_store_config,
    )
)

class KafkaMessageProcessor(ConsumerInitializer):
    def __init__(
            self,
            config: ConsumerConfig
    ):
        super().__init__(config=config)
        self._config = config
    
    # IF "TRUE", ERROR MESSAGES DATA ARE STORED IN REDIS
    @kafka_processor(store_error_messages=True)
    async def process(self, message):
        
        # HERE YOU PROCESS KAFKA MESSAGES
        pass

Producer Initializer

First Step. Add enviroment variables

To configure a producer you will need only 2 fields: URL for connecting Kafka and the producer name.

producer_config = ProducerConfig(
        producer_name='KafkaTesterProducer1',
        bootstrap_servers='HOST:PORT',
)

Second Step. Add processor

The send_message method of ProducerInitializer class allows to send a message to a topic

Also an optional parameter of DataSend scheme is "headers" which allows to send additional information in a message without changing the structure of this message.

EXAMPLE:

task = ProducerInitializer(
    config=producer_config
)

task.send_message(
    data_to_send=DataSend(
        key='KEY1',
        value='VALUE1',
    )
)

# with headers
task.send_message(
    data_to_send=DataSend(
        key='KEY2',
        value='VALUE12,
        headers=[('additinal_key', 'additinal_value')]
    )
)

Additional Step. Add security

To add security, you should set attribute security_config using class KafkaSecurityConfig

from resistant_kafka import ProducerConfig
from resistant_kafka.common_schemas import KafkaSecurityConfig

security_config = KafkaSecurityConfig(
    oauth_cb=method_to_get_token,
    security_protocol='SASL_PLAINTEXT',
    sasl_mechanisms='OAUTHBEARER'
)

producer_config = ProducerConfig(
        producer_name='KafkaTesterProducer1',
        bootstrap_servers='HOST:PORT',

        security_config=consumer_config
)

Installation

    pip install resistant-kafka

CONSUMER CODE EXAMPLE

from custom_utils import custom_token_method
from resistant_kafka.common_schemas import KafkaSecurityConfig
from resistant_kafka.consumer_schemas import ConsumerConfig
from resistant_kafka.consumer import ConsumerInitializer, kafka_processor, init_kafka_connection

consumer_config = KafkaSecurityConfig(
    oauth_cb=custom_token_method,
    security_protocol='SASL_PLAINTEXT',
    sasl_mechanisms='OAUTHBEARER'
)


class KafkaMessage1Processor(ConsumerInitializer):
    def __init__(
            self, config: ConsumerConfig
    ):
        super().__init__(config=config)
        self._config = config

    # This way empty messages will not be process
    @kafka_processor()
    async def process(self, message):
        message_key = message.key().decode("utf-8")
        message_value = message.value().decode("utf-8")

        if message_value in ['WRONG_VALUE']:
            raise ValueError('You catch wrong value')

        print('-----------------------------')
        print('KEY', message_key)
        print('VALUE', message_value)
        print('CONSUMER', self._config.topic_to_subscribe)
        print('-----------------------------')


class KafkaMessage2Processor(ConsumerInitializer):
    def __init__(
            self,
            config: ConsumerConfig
    ):
        super().__init__(config=config)
        self._config = config
    
    # This way every empty message = None
    @kafka_processor(read_empty_messages=True)
    async def process(self, message):
        message_key = message.key().decode("utf-8")
        message_value = message.value().decode("utf-8")

        print('-----------------------------')
        print('KEY', message_key)
        print('VALUE', message_value)
        print('PRODUCER', self._config.topic_to_subscribe)
        print('-----------------------------')


process_task_1 = KafkaMessage1Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer1',
        processor_name='KafkaProcessor1',
        bootstrap_servers='HOST:PORT',
        group_id='CONSUMER NAME',
        auto_offset_reset='latest',
        enable_auto_commit=False,
        security_config=consumer_config
    )
)

process_task_2 = KafkaMessage2Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer2',
        processor_name='KafkaProcessor2',
        bootstrap_servers='HOST:PORT',
        group_id='CONSUMER NAME',
        auto_offset_reset='latest',
        enable_auto_commit=False,
        security_config=consumer_config
    )
)

init_kafka_connection(
    tasks=[process_task_1, process_task_2]
)

PRODUCER CODE EXAMPLE

from custom_utils import custom_token_method
from resistant_kafka import ProducerInitializer, ProducerConfig, DataSend
from resistant_kafka.common_schemas import KafkaSecurityConfig

security_config = KafkaSecurityConfig(
    oauth_cb=custom_token_method,
    security_protocol='SASL_PLAINTEXT',
    sasl_mechanisms='OAUTHBEARER'
)

task = ProducerInitializer(
    config=ProducerConfig(
        producer_name='KafkaTesterProducer1',
        bootstrap_servers='HOST:PORT',
        security_config=security_config
    )
)
task.send_message(
    data_to_send=DataSend(
        key='KEY1',
        value='VALUE1',
    )
)
task.send_message(
    data_to_send=DataSend(
        key='KEY1',
        value='WRONG_VALUE'
    )
)

task = ProducerInitializer(
    config=ProducerConfig(
        producer_name='KafkaTesterProducer2',
        bootstrap_servers='HOST:PORT',
        security_config=security_config
    ),

)
task.send_message(
    data_to_send=DataSend(
        key='KEY2',
        value='VALUE2',
        headers=[
            ('key_1', 'value_1'),
            ('key_2', 'value_2'),
        ]
    ))