kafka_rpc_pubsub

Pydantic-typed Kafka pub/sub с RPC-семантикой. Тонкий слой поверх kfk_mng — даёт типизированный KafkaProducer[T] / KafkaConsumer[T] для inter-service async events.

Установка

pip install kafka_rpc_pubsub

Возможности

• ✅ Generic над Pydantic v2: KafkaProducer[MyMessage], KafkaConsumer[MyMessage]
• ✅ ABC + concrete subclasses: Producer (только publish), Consumer (только subscribe)
• ✅ JSON сериализация через model_dump_json() / model_validate_json()
• ✅ Manual offset commit
• ✅ Consumer groups для параллельной обработки
• ✅ Layered API: KafkaProtocol → PubSub[T] → BaseKafkaService[T] → Producer/Consumer

Быстрый старт

import kfk_mng
from pydantic import BaseModel
from kafka_rpc_pubsub import KafkaProducer, KafkaConsumer

# 1. Регистрация подключения через kfk_mng
kfk_mng.register(
    alias="default",
    bootstrap_servers="kafka:9092",
    enable_producer=True,
    enable_consumer=True,
    consumer_topics=["events"],
    consumer_group_id="my-workers",
)

# 2. Описать сообщение через Pydantic
class EventMessage(BaseModel):
    id: str
    payload: dict


# 3. Создать роль (Producer / Consumer)
class EventProducer(KafkaProducer[EventMessage]):
    model = EventMessage


class EventConsumer(KafkaConsumer[EventMessage]):
    model = EventMessage


# 4. Использовать
producer = EventProducer(topic="events")
partition, offset = await producer.call(EventMessage(id="x", payload={"a": 1}))

consumer = EventConsumer(topic="events", group_id="my-workers")
while True:
    result = await consumer.process()
    if result.message:
        await handle(result.message)
        await consumer.commit(result)

API

Layered архитектура

kfk_mng.KafkaComponents       (raw aiokafka)
    ↓
KafkaProtocol                 (raw send / consume / commit)
    ↓
PubSub[T]                     (Pydantic-typed publish / subscribe / commit)
    ↓
BaseKafkaService[T]           (ABC, integration с alias, group_id, model)
    ↓
KafkaProducer[T] / KafkaConsumer[T]   (роли)

Классы

Partitioning key

По умолчанию get_key(message) возвращает str(message.id) или str(message.oid). Override:

class OrderProducer(KafkaProducer[OrderMessage]):
    model = OrderMessage

    def get_key(self, message: OrderMessage) -> str:
        return message.customer_id  # партиционирование по клиенту

Связанные библиотеки

• kfk_mng — connection manager (обязательная зависимость)
• Inspired by redis_rpc_pubsub (тот же API, но через Redis)

License

MIT