kfk_mng

Atomic Kafka connection manager. Регистрирует Kafka producers / consumers по alias, скрывает aiokafka boilerplate.

Установка

pip install kfk_mng

Возможности

• ✅ Регистрация Kafka подключений по alias
• ✅ Producer + Consumer пара (KafkaComponents) под одним alias
• ✅ Поддержка SSL / SASL
• ✅ Lazy start adapters (нет старта пока не вызвал send/getmany)
• ✅ Async API через aiokafka
• ✅ Manual offset commit по умолчанию (надёжность)

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

import 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",
)

# Получение
components = kfk_mng.get("default")
producer = components.producer
consumer = components.consumer

# Использование
metadata = await producer.send(topic="events", value=b"hello")
records = await consumer.getmany(timeout_ms=1000, max_records=10)
await consumer.commit()

# Закрытие
await kfk_mng.close("default")

С SSL и SASL

kfk_mng.register(
    alias="prod",
    bootstrap_servers="kafka.prod:9093",
    security_protocol="SASL_SSL",
    ssl_conf={"cafile": "/path/ca.pem"},
    sasl_conf={
        "mechanism": "PLAIN",
        "username": "user",
        "password": "secret",
    },
    enable_producer=True,
)

Multi-instance

kfk_mng.register(alias="default", bootstrap_servers="kafka:9092", enable_producer=True)
kfk_mng.register(alias="dlq", bootstrap_servers="dlq-kafka:9092", enable_producer=True)

await kfk_mng.get("default").producer.send("events", b"x")
await kfk_mng.get("dlq").producer.send("dead-letters", b"y")

Одноразовое подключение (без регистрации)

components = kfk_mng.connect(
    bootstrap_servers="kafka:9092",
    enable_producer=True,
)
await components.producer.start()
# ... use ...
await components.producer.stop()

API

Сопутствующие библиотеки

• kafka_rpc_pubsub — Pydantic-typed Kafka pub/sub поверх kfk_mng

License

MIT