Kafka 1.2.0

.github/workflows/tests.yml

@@ -30,6 +30,22 @@ jobs:
         image: nats
         ports:
           - 4222:4222
+
+      kafka:
+        image: bitnami/kafka
+        ports:
+          - 9092:9092
+        env:
+          KAFKA_ENABLE_KRAFT: "true"
+          KAFKA_CFG_NODE_ID: "1"
+          KAFKA_CFG_PROCESS_ROLES: "broker,controller"
+          KAFKA_CFG_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
+          KAFKA_CFG_LISTENERS: "PLAINTEXT://:9092,CONTROLLER://:9093"
+          KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
+          KAFKA_CFG_ADVERTISED_LISTENERS: "PLAINTEXT://127.0.0.1:9092"
+          KAFKA_BROKER_ID: "1"
+          KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: "1@kafka:9093"
+          ALLOW_PLAINTEXT_LISTENER: "true"
 
     steps:
       - uses: actions/checkout@v3

CONTRIBUTING.md

@@ -105,6 +105,22 @@ services:
     ports:
       - 4222:4222
       - 8222:8222  # management
+
+  kafka:
+    image: bitnami/kafka
+    ports:
+      - 9092:9092
+    environment:
+      - KAFKA_ENABLE_KRAFT=yes
+      - KAFKA_CFG_NODE_ID=1
+      - KAFKA_CFG_PROCESS_ROLES=broker,controller
+      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
+      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
+      - KAFKA_BROKER_ID=1
+      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
+      - ALLOW_PLAINTEXT_LISTENER=yes
 ```
 
 ```bash

README.md

@@ -49,7 +49,8 @@ It is a modern, high-level framework on top of popular specific Python brokers l
 * **MQ independent**: Single interface to popular MQ:
   * **Redis** (based on <a href="https://redis.readthedocs.io/en/stable/index.html" target="_blank">redis-py</a>)
   * **RabbitMQ** (based on <a href="https://aio-pika.readthedocs.io/en/latest/" target="_blank">aio-pika</a>)
-  * **NATS** (based on <a href="https://github.com/nats-io/nats.py" target="_blank">nats-py</a>)
+  * **Kafka** (based on <a href="https://aiokafka.readthedocs.io/en/stable/" target="_blank">aiokafka</a>)
+  * **Nats** (based on <a href="https://github.com/nats-io/nats.py" target="_blank">nats-py</a>)
 * <a href="https://lancetnik.github.io/Propan/getting_started/4_broker/5_rpc/" target="_blank">**RPC**</a>: The framework supports RPC requests over MQ, which will allow performing long operations on remote services asynchronously.
 * [**Great to develop**](#cli-power): CLI tool provides great development experience:
   * framework-independent way to manage the project environment
@@ -63,10 +64,10 @@ It is a modern, high-level framework on top of popular specific Python brokers l
 |-------------------|:-------------------------------------------------------:|:--------------------:|
 | **RabbitMQ**      | :heavy_check_mark: **stable** :heavy_check_mark:        | :mag: planning :mag: |
 | **Redis**         | :heavy_check_mark: **stable** :heavy_check_mark:        | :mag: planning :mag: |
+| **Kafka**         | :warning: **beta** :warning:                            | :mag: planning :mag: |
 | **Nats**          | :warning: **beta** :warning:                            | :mag: planning :mag: |
 | **NatsJS**        | :hammer_and_wrench: **in progress** :hammer_and_wrench: | :mag: planning :mag: |
 | **MQTT**          | :mag: planning :mag:                                    | :mag: planning :mag: |
-| **Kafka**         | :mag: planning :mag:                                    | :mag: planning :mag: |
 | **Redis Streams** | :mag: planning :mag:                                    | :mag: planning :mag: |
 | **Pulsar**        | :mag: planning :mag:                                    | :mag: planning :mag: |
 | **SQS**           | :mag: planning :mag:                                    | :mag: planning :mag: |
@@ -141,6 +142,8 @@ pip install "propan[async-rabbit]"
 pip install "propan[async-nats]"
 # or
 pip install "propan[async-redis]"
+# or
+pip install "propan[async-kafka]"
 ```
 
 ### Basic usage

docs/docs/en/CHANGELOG.md

@@ -1,8 +1,22 @@
 # CHANGELOG
 
+## 2023-05-23 **0.1.2.0** Kafka
+
+**Propan** added support for *Kafka* as a message broker. This functionality is full tested.
+
+*KafkaBroker* supports:
+
+* message delivery
+* test client, without the need to run *Kafka*
+* *FastAPI* Plugin
+
+*KafkaBroker* not supports **RPC** yet.
+
+Also, **Propan CLI** is able to generate templates to any supported broker
+
 ## 2023-05-18 **0.1.1.0** REDIS
 
-**Propan** added support for *Redis Pub/Sub* as a message broker. This functionality is fully tested and described in the documentation.
+**Propan** added support for *Redis Pub/Sub* as a message broker. This functionality is full tested and described in the documentation.
 
 *RedisBroker* supports:
 

docs/docs/en/contributing/2_contributing-index.md

@@ -105,6 +105,22 @@ services:
     ports:
       - 4222:4222
       - 8222:8222  # management
+
+  kafka:
+    image: bitnami/kafka
+    ports:
+      - 9092:9092
+    environment:
+      - KAFKA_ENABLE_KRAFT=yes
+      - KAFKA_CFG_NODE_ID=1
+      - KAFKA_CFG_PROCESS_ROLES=broker,controller
+      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
+      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
+      - KAFKA_BROKER_ID=1
+      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
+      - ALLOW_PLAINTEXT_LISTENER=yes
 ```
 
 ```bash

docs/docs/en/index.md

@@ -43,7 +43,8 @@ It is a modern, high-level framework on top of popular Python libraries for vari
 * **MQ independent**: Single interface to popular MQ:
     * **Redis** (based on [redis-py]("https://redis.readthedocs.io/en/stable/index.html"){target="_blank"})
     * **RabbitMQ** (based on [aio-pika](https://aio-pika.readthedocs.io/en/latest/){target="_blank"})
-    * **NATS** (based on [nats-py](https://github.com/nats-io/nats.py){target="_blank"})
+    * **Kafka** (based on [aiokafka](https://aiokafka.readthedocs.io/en/stable/){target="_blank"})
+    * **Nats** (based on [nats-py](https://github.com/nats-io/nats.py){target="_blank"})
 * [**RPC**](getting_started/4_broker/5_rpc/): The framework supports RPC requests on top of message brokers, which will allow performing long operations on remote services asynchronously.
 * [**Great to develop**](getting_started/2_cli/): CLI tool provides great development experience:
     * framework-independent way to manage the project environment
@@ -57,7 +58,7 @@ It is a modern, high-level framework on top of popular Python libraries for vari
 
 With declarative tools you can define **what you need to get**. With traditional imperative tools you must write **what you need to do**.
 
-Take a look at classic imperative tools, such as [aio-pika](https://aio-pika.readthedocs.io/en/latest/){target="_blank"}, [pika](https://pika.readthedocs.io/en/stable/){target="_blank"}, [redis-py]("https://redis.readthedocs.io/en/stable/index.html"){target="_blank"}, [nats-py](https://github.com/nats-io/nats.py){target="_blank"}, etc.
+Take a look at classic imperative tools, such as [aio-pika](https://aio-pika.readthedocs.io/en/latest/){target="_blank"}, [pika](https://pika.readthedocs.io/en/stable/){target="_blank"}, [redis-py]("https://redis.readthedocs.io/en/stable/index.html"){target="_blank"}, [nats-py](https://github.com/nats-io/nats.py){target="_blank"}, [aiokafka](https://aiokafka.readthedocs.io/en/stable/){target="_blank"}, etc.
 
 This is the **Quickstart** with the *aio-pika*:
 
@@ -114,10 +115,10 @@ This is the **Propan** declarative way to write the same code. That is so much e
 |-------------------|:-------------------------------------------------------:|:--------------------:|
 | **RabbitMQ**      | :heavy_check_mark: **stable** :heavy_check_mark:        | :mag: planning :mag: |
 | **Redis**         | :heavy_check_mark: **stable** :heavy_check_mark:        | :mag: planning :mag: |
+| **Kafka**         | :warning: **beta** :warning:                            | :mag: planning :mag: |
 | **Nats**          | :warning: **beta** :warning:                            | :mag: planning :mag: |
 | **NatsJS**        | :hammer_and_wrench: **in progress** :hammer_and_wrench: | :mag: planning :mag: |
 | **MQTT**          | :mag: planning :mag:                                    | :mag: planning :mag: |
-| **Kafka**         | :mag: planning :mag:                                    | :mag: planning :mag: |
 | **Redis Streams** | :mag: planning :mag:                                    | :mag: planning :mag: |
 | **Pulsar**        | :mag: planning :mag:                                    | :mag: planning :mag: |
 | **SQS**           | :mag: planning :mag:                                    | :mag: planning :mag: |

docs/docs/ru/CHANGELOG.md

@@ -1,5 +1,17 @@
 # CHANGELOG
 
+## 2023-05-23 **0.1.2.0** Kafka
+
+В **Propan** добавлена поддержка *Kafka* в качестве брокера сообщений. Данный функционал полностью протестирован.
+
+*KafkaBroker* поддерживает:
+
+* доставку сообщений
+* тестовый клиент, без необходимости запуска *Kafka*
+* В качестве плагина *FastAPI*
+
+*KafkaBroker* на данный момент не поддерживает **RPC** запросы.
+
 ## 2023-05-18 **0.1.1.0** REDIS
 
 В **Propan** добавлена поддержка *Redis Pub/Sub* в качестве брокера сообщений. Данный функционал полностью протестирован и описан в документации.

docs/docs/ru/contributing/2_contributing-index.md

@@ -114,6 +114,22 @@ services:
     ports:
       - 4222:4222
       - 8222:8222  # management
+
+  kafka:
+    image: bitnami/kafka
+    ports:
+      - 9092:9092
+    environment:
+      - KAFKA_ENABLE_KRAFT=yes
+      - KAFKA_CFG_NODE_ID=1
+      - KAFKA_CFG_PROCESS_ROLES=broker,controller
+      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
+      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
+      - KAFKA_BROKER_ID=1
+      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
+      - ALLOW_PLAINTEXT_LISTENER=yes
 ```
 
 #### Hatch

docs/docs/ru/index.md

@@ -42,8 +42,9 @@
 * [**Интeграция**](getting_started/1_quick-start/#http): Propan полностью совместим с [любыми HTTP фреймворками](integrations/1_integrations-index/)
 * **Независимость от брокеров**: Единый интерфейс для популярных брокеров:
     * **Redis** (основан на [redis-py]("https://redis.readthedocs.io/en/stable/index.html"){target="_blank"})
-    * **NATS** (основан на [nats-py](https://github.com/nats-io/nats.py){target="_blank"})
     * **RabbitMQ** (основан на [aio-pika](https://aio-pika.readthedocs.io/en/latest/){target="_blank"})
+    * **Kafka** (основан на [aiokafka](https://aiokafka.readthedocs.io/en/stable/){target="_blank"})
+    * **Nats** (основан на [nats-py](https://github.com/nats-io/nats.py){target="_blank"})
 * [**RPC**](getting_started/4_broker/5_rpc/): Фреймворк поддерживает RPC запросы поверх брокеров сообщений, что позволит выполнять длительные операции на удаленных сервисах асинхронно.
 * [**Скорость разработки**](getting_started/2_cli/): собственный *CLI* инструмент предоставляет отличный опыт разработки:
     * Полностью совместимый с любым фреймворком способ управлять окружением проекта
@@ -58,7 +59,7 @@
 Декларативные иснтрументы позволяют нам описывать **что мы хотим получить**, в то время как традиционные императивные инструменты
 заставляют нас писать **что мы хотим сделать**.
 
-К традиционным императивным библиотекам относятся [aio-pika](https://aio-pika.readthedocs.io/en/latest/){target="_blank"}, [pika](https://pika.readthedocs.io/en/stable/){target="_blank"}, [redis-py]("https://redis.readthedocs.io/en/stable/index.html"){target="_blank"}, [nats-py](https://github.com/nats-io/nats.py){target="_blank"} и подобные.
+К традиционным императивным библиотекам относятся [aio-pika](https://aio-pika.readthedocs.io/en/latest/){target="_blank"}, [pika](https://pika.readthedocs.io/en/stable/){target="_blank"}, [redis-py]("https://redis.readthedocs.io/en/stable/index.html"){target="_blank"}, [nats-py](https://github.com/nats-io/nats.py){target="_blank"}, [aiokafka](https://aiokafka.readthedocs.io/en/stable/){target="_blank"} и подобные.
 
 Например, это **Quickstart** из библиотеки *aio-pika*:
 
@@ -117,10 +118,10 @@ async def base_handler(body):
 |-------------------|:-------------------------------------------------------:|:--------------------:|
 | **RabbitMQ**      | :heavy_check_mark: **stable** :heavy_check_mark:        | :mag: planning :mag: |
 | **Redis**         | :heavy_check_mark: **stable** :heavy_check_mark:        | :mag: planning :mag: |
+| **Kafka**         | :warning: **beta** :warning:                            | :mag: planning :mag: |
 | **Nats**          | :warning: **beta** :warning:                            | :mag: planning :mag: |
 | **NatsJS**        | :hammer_and_wrench: **in progress** :hammer_and_wrench: | :mag: planning :mag: |
 | **MQTT**          | :mag: planning :mag:                                    | :mag: planning :mag: |
-| **Kafka**         | :mag: planning :mag:                                    | :mag: planning :mag: |
 | **Redis Streams** | :mag: planning :mag:                                    | :mag: planning :mag: |
 | **Pulsar**        | :mag: planning :mag:                                    | :mag: planning :mag: |
 | **SQS**           | :mag: planning :mag:                                    | :mag: planning :mag: |

docs/docs_src/contributing/adapter/parent.py

@@ -37,7 +37,7 @@ async def publish(
         message: SendableMessage,
         *args: Any,
         callback: bool = False,
-        callback_timeout: float | None = None,
+        callback_timeout: Optional[float] = None,
         raise_timeout: bool = False,
         **kwargs: Any,
     ) -> Any:

propan/__about__.py

@@ -1,3 +1,13 @@
 """Simple and fast framework to create message brokers based microservices"""
 
-__version__ = "0.1.1.4"
+__version__ = "0.1.2.0"
+
+
+INSTALL_MESSAGE = (
+    "You should specify using broker!\n"
+    "Install it using one of the following commands:\n"
+    'pip install "propan[async-rabbit]"\n'
+    'pip install "propan[async-nats]"\n'
+    'pip install "propan[async-redis]"\n'
+    'pip install "propan[async-kafka]"\n'
+)

propan/__init__.py

@@ -1,4 +1,5 @@
 # Imports to use at __all__
+from propan.__about__ import INSTALL_MESSAGE
 from propan.cli.app import *  # noqa: F403
 from propan.log import *  # noqa: F403
 from propan.utils import *  # noqa: F403
@@ -15,18 +16,15 @@
 
 try:
     from propan.brokers.redis import RedisBroker
-except Exception as e:
-    print(e)
+except Exception:
     RedisBroker = None  # type: ignore
 
-assert any((RabbitBroker, NatsBroker, RedisBroker)), (
-    "You should specify using broker!\n"
-    "Install it using one of the following commands:\n"
-    'pip install "propan[async-rabbit]"\n'
-    'pip install "propan[async-nats]"\n'
-    'pip install "propan[async-redis]"\n'
-)
+try:
+    from propan.brokers.kafka import KafkaBroker
+except Exception:
+    KafkaBroker = None  # type: ignore
 
+assert any((RabbitBroker, NatsBroker, RedisBroker)), INSTALL_MESSAGE
 
 __all__ = (  # noqa: F405
     # app
@@ -46,4 +44,5 @@
     "NatsBroker",
     "RabbitBroker",
     "RedisBroker",
+    "KafkaBroker",
 )

propan/annotations.py

@@ -2,6 +2,7 @@
 
 from typing_extensions import Annotated
 
+from propan.__about__ import INSTALL_MESSAGE
 from propan.cli.app import PropanApp
 from propan.utils.context import Context as ContextField
 from propan.utils.context import ContextRepo as CR
@@ -21,6 +22,7 @@
 except Exception:
     RabbitBroker = RabbitMessage = None  # type: ignore
 
+
 try:
     from nats.aio.msg import Msg
 
@@ -31,23 +33,31 @@
 except Exception:
     NatsBroker = NatsMessage = None  # type: ignore
 
+
 try:
     from propan.brokers.redis import RedisBroker as RedB
 
     RedisBroker = Annotated[RedB, ContextField("broker")]
 except Exception:
     RedisBroker = None  # type: ignore
 
+
+try:
+    from aiokafka.structs import ConsumerRecord
+
+    from propan.brokers.kafka import KafkaBroker as KB
+
+    KafkaBroker = Annotated[KB, ContextField("broker")]
+    KafkaMessage = Annotated[ConsumerRecord, ContextField("message")]
+except Exception:
+    KafkaBroker = KafkaMessage = None  # type: ignore
+
+
 assert any(
     (
         all((RabbitBroker, RabbitMessage)),
         all((NatsBroker, NatsMessage)),
         RedisBroker,
+        all((KafkaBroker, KafkaMessage)),
     )
-), (
-    "You should specify using broker!\n"
-    "Install it using one of the following commands:\n"
-    'pip install "propan[async-rabbit]"\n'
-    'pip install "propan[async-nats]"\n'
-    'pip install "propan[async-redis]"\n'
-)
+), INSTALL_MESSAGE

propan/brokers/kafka/__init__.py

@@ -0,0 +1,3 @@
+from propan.brokers.kafka.kafka_broker import KafkaBroker
+
+__all__ = ("KafkaBroker",)