Merge pull request #147 from conductor-sdk/add-kafka-queue-support

Added EventClient with support for Queue configuration (e.g. Kafka)

Author: Gustavo Gardusi

src/conductor/client/event/__init__.py

@@ -0,0 +1,30 @@
+from conductor.client.http.api_client import ApiClient
+from conductor.client.http.api.event_resource_api import EventResourceApi
+from conductor.client.event.queue.queue_configuration import QueueConfiguration
+
+
+class EventClient:
+    def __init__(self, api_client: ApiClient):
+        self.client = EventResourceApi(api_client)
+
+    def delete_queue_configuration(self, queue_configuration: QueueConfiguration) -> None:
+        return self.client.delete_queue_config(
+            queue_name=queue_configuration.queue_name,
+            queue_type=queue_configuration.queue_type,
+        )
+
+    def get_kafka_queue_configuration(self, queue_topic: str) -> QueueConfiguration:
+        return self.get_queue_configuration(
+            queue_type='kafka',
+            queue_name=queue_topic,
+        )
+
+    def get_queue_configuration(self, queue_type: str, queue_name: str):
+        return self.client.get_queue_config(queue_type, queue_name)
+
+    def put_queue_configuration(self, queue_configuration: QueueConfiguration):
+        return self.client.put_queue_config(
+            body=queue_configuration.get_worker_configuration(),
+            queue_name=queue_configuration.queue_name,
+            queue_type=queue_configuration.queue_type,
+        )

src/conductor/client/event/event_client.py

@@ -0,0 +1,35 @@
+from conductor.client.event.queue.queue_configuration import QueueConfiguration
+from conductor.client.event.queue.queue_worker_configuration import QueueWorkerConfiguration
+from typing import Any, Dict
+
+
+class KafkaQueueConfiguration(QueueConfiguration):
+    def __init__(self, queue_topic_name: str):
+        super().__init__(queue_topic_name, 'kafka')
+
+    def get_worker_configuration(self) -> Dict[str, Any]:
+        worker_configuration = {}
+        for required_key in ['consumer', 'producer']:
+            if required_key not in self.worker_configuration:
+                raise RuntimeError(f'required key not present: {required_key}')
+        for key, value in self.worker_configuration.items():
+            worker_configuration[key] = value.configuration
+        return worker_configuration
+
+
+class KafkaConsumerConfiguration(QueueWorkerConfiguration):
+    def __init__(self, bootstrap_servers_config: str):
+        super().__init__()
+        super().add_configuration(
+            key='bootstrap.servers',
+            value=bootstrap_servers_config
+        )
+
+
+class KafkaProducerConfiguration(QueueWorkerConfiguration):
+    def __init__(self, bootstrap_servers_config: str):
+        super().__init__()
+        super().add_configuration(
+            key='bootstrap.servers',
+            value=bootstrap_servers_config
+        )

src/conductor/client/event/queue/__init__.py

@@ -0,0 +1,23 @@
+from conductor.client.event.queue.queue_worker_configuration import QueueWorkerConfiguration
+
+from abc import ABC
+from typing import Any, Dict
+
+
+class QueueConfiguration(ABC):
+    WORKER_CONSUMER_KEY = "consumer"
+    WORKER_PRODUCER_KEY = "producer"
+
+    def __init__(self, queue_name: str, queue_type: str):
+        self.queue_name = queue_name
+        self.queue_type = queue_type
+        self.worker_configuration = {}
+
+    def add_consumer(self, worker_configuration: QueueWorkerConfiguration) -> None:
+        self.worker_configuration[self.WORKER_CONSUMER_KEY] = worker_configuration
+
+    def add_producer(self, worker_configuration: QueueWorkerConfiguration) -> None:
+        self.worker_configuration[self.WORKER_PRODUCER_KEY] = worker_configuration
+
+    def get_worker_configuration(self) -> Dict[str, Any]:
+        raise NotImplementedError

src/conductor/client/event/queue/kafka_queue_configuration.py

@@ -0,0 +1,6 @@
+class QueueWorkerConfiguration:
+    def __init__(self):
+        self.configuration = {}
+
+    def add_configuration(self, key: str, value: str) -> None:
+        self.configuration[key] = value