This lirary provides you with aiokafka broker for taskiq.
Usage:
from taskiq_aio_kafka import AioKafkaBroker
broker = AioKafkaBroker(bootstrap_servers="localhost")
@broker.task
async def test() -> None:
print("The best task ever!")You can configure kafka producer and consumer with special methods configure_producer and configure_consumer.
Example:
from taskiq_aio_kafka import AioKafkaBroker
broker = AioKafkaBroker(bootstrap_servers="localhost")
# configure producer, you can set any parameter from
# base AIOKafkaProducer, except `loop` and `bootstrap_servers`
broker.configure_producer(request_timeout_ms=100000)
# configure consumer, you can set any parameter from
# base AIOKafkaConsumer, except `loop` and `bootstrap_servers`
broker.configure_consumer(group_id="the best group ever.")AioKafkaBroker parameters:
bootstrap_servers- url to kafka nodes. Can be either string or list of strings.kafka_topic- custom topic in kafka.result_backend- custom result backend.task_id_generator- custom task_id genertaor.kafka_admin_client- customkafkaadmin client.delete_topic_on_shutdown- flag to delete topic on broker shutdown.