Metadata-Version: 2.4
Name: resistant_kafka_avataa
Version: 0.9.8b5
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: confluent-kafka[protobuf,schemaregistry]<3.0.0,>=2.10.0
Requires-Dist: pydantic<3.0.0,>=1.10.21
Dynamic: description
Dynamic: description-content-type
Dynamic: license-file
Dynamic: requires-dist

# 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.
###

### 🔁 Serialisation | Deserialisation

The library allows you to serialize and deserialize data that you send using .proto formats. It also supports 
Schema Registry, which allows you to make sure that the data arrives in the correct format.
### 

### 💾 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:

```python
from resistant_kafka.consumer_schemas import ConsumerConfig

process_task_1 = KafkaMessage1Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer1',
        processor_name='KafkaProcessor1',
        bootstrap_servers='localhost:9093',
        group_id='LocalTester1',
        auto_offset_reset='latest',
        enable_auto_commit=False,
)

process_task_2 = KafkaMessage2Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer2',
        processor_name='KafkaProcessor2',
        bootstrap_servers='localhost:9093',
        group_id='LocalTester1',
        auto_offset_reset='latest',
        enable_auto_commit=False,
)

```
##
## 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**.⚠️

⚠️ **Attribute "_message_" in main method "_process_" is required** ⚠️

⚠️**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:

```python

from resistant_kafka.consumer import ConsumerInitializer, kafka_processor
from resistant_kafka_avataa.message_desirializers import MessageDeserializer

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

    # required decorator
    # to raise error, instead logging @kafka_processor(raise_error=True)
    @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')
        
        # here your message proccessing

```
## 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:

```python
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"_** ⚠️
```python
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

```python
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. Add deserializers
In Kafka, messages are stored as bytes, so we need to get them in the format we expect. The library provides
the ability to **_convert messages from bytes to objects using your .proto files_** or, if no .proto files are
available, **_we will convert them to strings for your future processing of this data_**.

### _PROTO FILES_
In case you have proto files that can help you format your messages - we can convert them from bytes to protobuf
structure.

This can be done with Kafka Schema Registry, if your project doesn't have Kafka Schema Registry - we will convert
bytes to strings.

You should use **_MessageDeserializer_** to registry your proto files with which you expect messages from the topic

### _REGISTRY .proto FILES_
```python
from resistant_kafka_avataa.message_desirializers import MessageDeserializer

deserializers_producer = MessageDeserializer(
    schema_registry_url="https://localhost:8081",
    topic='KafkaTesterProducer1'
)
deserializers_producer.register_protobuf_deserializer(ProtoFileWithData_1)
deserializers_producer.register_protobuf_deserializer(ProtoFileWithData_2)


process_task_1 = KafkaMessage1Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer1',
        processor_name='KafkaProcessor1',
        **consumer_config
    ),
    deserializers=deserializers_producer_1
)
```

### _SIMPLE MESSAGES_
```python
from resistant_kafka_avataa.message_desirializers import MessageDeserializer

deserializers_producer_2 = MessageDeserializer(
    topic='KafkaTesterProducer2'
)
```

### _READ DESERIALIZED MESSAGES_
Using the **_deserialize_** method you can convert the byte format of the message value from the format of objects
described in .proto and end up with an object instead of bytes

```python
class KafkaMessage1Processor(ConsumerInitializer):
    def __init__(
            self,
            config: ConsumerConfig,
            deserializers: MessageDeserializer = None
    ):
        super().__init__(config=config, deserializers=deserializers)
        self._config = config
        self._deserializers = deserializers

    @kafka_processor(store_error_messages=True)
    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')
        
        deserialized_message = self._deserializers.deserialize(message=message)
```

## Additional Step. Integrate Redis
```shell
pip install redis=4.5.4
```

```python
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='localhost: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.
```python
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:

```python
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

```python
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
)
```

## Additional Step. Add serializers
In Kafka, messages are stored as bytes, so we need to get them in the format we expect. The library provides 
the ability to **_convert messages from bytes to objects using your .proto files_** or, if no .proto files are available, 
**_we will convert them to strings for your future processing of this data_**.

### _PROTO FILES_
In case you have proto files that can help you format your messages - we can convert them from bytes to protobuf structure.

This can be done with Kafka Schema Registry, if your project doesn't have Kafka Schema Registry - we will convert bytes to strings.

You should use **_MessageSerializer_** to registry your proto files with which you expect messages from the topic

### REGISTRY .proto FILES
```python
from resistant_kafka_avataa.message_serializers import MessageSerializer

serializer_task = MessageSerializer(
    schema_registry_url="https://localhost:8081",
    topic='KafkaTesterProducer1'
)
serializer_task.register_protobuf_deserializer(ProtoFile1)
serializer_task.register_protobuf_deserializer(ProtoFile2)

_producer_manager.send_message(
    data_to_send=DataSend(
        key=key,
        value=serializer_task.serialize(
            message_to_send=message_to_send,
            class_name=ProtoFile1
        ),
    ),
)
```

### SIMPLE MESSAGES
```python
from resistant_kafka_avataa.message_serializers import MessageSerializer

serializer_task = MessageSerializer(
    topic='KafkaTesterProducer1'
)
_producer_manager.send_message(
    data_to_send=DataSend(
        key=key,
        value=serializer_task.serialize(
            message_to_send=message_to_send
        ),
    ),
)
```

#
## Installation

```shell
    pip install resistant-kafka-avataa
```

# CONSUMER CODE EXAMPLE

```python
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,
            deserializers: MessageDeserializer = None
    ):
        super().__init__(config=config, deserializers=deserializers)
        self._config = config
        self._deserializers = deserializers

    @kafka_processor(store_error_messages=True)
    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,
            deserializers: MessageDeserializer = None
    ):
        super().__init__(config=config, deserializers=deserializers)
        self._config = config

    @kafka_processor()
    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('-----------------------------')


deserializers_producer_1 = MessageDeserializer(
    schema_registry_url='http://localhost:8081',
    topic='KafkaTesterProducer1'
)
deserializers_producer_1.register_protobuf_deserializer(ProtoFileToDeserialize)

deserializers_producer_2 = MessageDeserializer(
    topic='KafkaTesterProducer2'
)

process_task_1 = KafkaMessage1Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer1',
        processor_name='KafkaProcessor1',
        bootstrap_servers='localhost:9093',
        group_id='LocalTester1',
        auto_offset_reset='latest',
        enable_auto_commit=False,
        redis_store_config=redis_store_config,
        security_config=consumer_config
    ),
    deserializers=deserializers_producer_1
)

process_task_2 = KafkaMessage2Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer2',
        processor_name='KafkaProcessor2',
        bootstrap_servers='localhost:9093',
        group_id='LocalTester1',
        auto_offset_reset='latest',
        enable_auto_commit=False,
        redis_store_config=redis_store_config,
        security_config=consumer_config
    ),
    deserializers=deserializers_producer_2
)

init_kafka_connection(
    tasks=[process_task_1, process_task_2]
)
```

# PRODUCER CODE EXAMPLE
```python
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'),
        ]
    ))

```
