Merge pull request #37 from salute-developers/feature/fix_kafka_messa…

…ge_key

DPNLPF-1729: add recovery of Kafka message key, refactor check_message_key, reformatting

core/mq/kafka/kafka_publisher.py

@@ -2,6 +2,7 @@
 import logging
 import os
 import time
+from typing import Union
 
 from confluent_kafka import Producer
 
@@ -21,11 +22,13 @@ def __init__(self, config):
         if internal_log_path:
             debug_logger = logging.getLogger("debug_publisher")
             timestamp = time.strftime("_%d%m%Y_")
-            debug_logger.addHandler(logging.FileHandler("{}/kafka_publisher_debug{}{}.log".format(internal_log_path, timestamp, os.getpid())))
+            debug_logger.addHandler(logging.FileHandler(
+                "{}/kafka_publisher_debug{}{}.log".format(internal_log_path, timestamp, os.getpid())
+            ))
             conf["logger"] = debug_logger
         self._producer = Producer(**conf)
 
-    def send(self, value, key=None, topic_key=None, headers=None):
+    def send(self, value: Union[str, bytes], key=None, topic_key=None, headers=None):
         try:
             topic = self._config["topic"]
             if topic_key is not None:

core/request/kafka_request.py

@@ -46,7 +46,7 @@ def send(self, data, publisher: KafkaPublisher, source_mq_message):
             }
             log("KafkaRequest: got no topic and no topic_key", params=log_params, level="ERROR")
 
-    def run(self, data, params=None):
+    def run(self, data, params):
         publishers = params["publishers"]
         publisher = publishers[self.kafka_key]
         self.send(data=data, publisher=publisher, source_mq_message=params["mq_message"])