velikafkaclient 1.0.8

Veli Kafka Client

Kafka Client for VELI.STORE

Description

This module is used for producing kafka events, subscribing to kafka topics and consuming kafka events.

To remember:

Each kafka topic event has a predefined structure which you can see in velikafkaclient/eventregistration, producer and consumer both have built in validation, meaning that if incorrect object structure is sent to a topic then producer will raise an exception as well as consumer.

How to use:

• Initialize kafka producer:

BOOTSTRAP_SERVERS = "kafka bootstrap servers uri"
kafka_producer: AsyncKafkaEventProducer = AsyncKafkaEventProducer(BOOTSTRAP_SERVERS)
await kafka_producer.start()

• Produce events to a topic:

from events.base import KafkaEvent
from topics.base import BaseTopic

kafka_event = KafkaEvent()
await kafka_producer.produce_event(BaseTopic, kafka_event)

• Setup and start consumer

async def base_handler(kafka_event: KafkaEvent):
    print(str(kafka_event))
    
    
bootstrap_servers = BOOTSTRAP_SERVER
consumer = AsyncKafkaConsumer(bootstrap_servers, group_id=GROUP_ID)
consumer.subscribe(BaseTopic, base_handler)

await consumer.start()
await consumer.consume()

How to add topics:

To add a new topic you need three things:

1. Topic itself
2. Topic event structure
3. Register Event structure to a Topic

All changes are done in velikafkaclient library (in the veli_libs repo)

• To add a topic go to kafka-client/velikafkaclient/topics and add topic like this:

class SomeTopicCollection(KafkaTopic):
    
    USER_REGISTRATIONS = 'user_registrations'

• To add an event structure go to kafka-client/velikafkaclient/events and create pydantic model for event structure:

class UserRegistrationEvent(KafkaEvent):
    
    id: int
    username: str
    password: str
    username: str

• To register event structure to a topic go to kafka-client/velikafkaclient/eventregistration.py and add:

kafka_topic_events.register_topic_event_model(KafkaTopic.USER_REGISTRATIONS, UserRegistrationEvent)

Once done follow the instructions in the veli_libs readme to update the library on pyip