Fixed review comments

Author: sujata-m

src/example.py

@@ -13,7 +13,7 @@
 core = Core(config='Local')
 print('Hello')
 
-topic = 'gtfs-flex-upload'
+topic = 'gtfspathways'
 subscription = 'upload-validation-processor-test'
 some_other_sub = 'usdufs'
 
@@ -46,11 +46,12 @@ def process(message):
 container = azure_client.get_container(container_name='gtfspathways')
 
 list_of_files = container.list_files()
-for single in list_of_files:
-    print(single.path)
-firstFile = list_of_files[0]
-# print(firstFile.name+'<><>')
-file_content = firstFile.get_body_text()
+if len(list_of_files) > 0:
+    for single in list_of_files:
+        print(single.path)
+    firstFile = list_of_files[0]
+    # print(firstFile.name+'<><>')
+    file_content = firstFile.get_body_text()
 
 # Creating a text stream
 txt = 'foo\nbar\nbaz'

src/python_ms_core/__init__.py

@@ -1,3 +1,4 @@
+import logging
 from .core.logger.logger import Logger
 from .core.logger.local_logger import LocalLogger
 from .core.topic.topic import Topic
@@ -7,6 +8,8 @@
 from .core.config.config import CoreConfig, LocalConfig
 
 LOCAL_ENV = 'LOCAL'
+AZURE_ENV = 'AZURE'
+
 
 class Core:
     def __init__(self, config=None):
@@ -20,22 +23,28 @@ def get_logger(self):
         logger_config = self.config.logger()
         if logger_config.provider.upper() == LOCAL_ENV:
             return LocalLogger(config=logger_config)
-        else:
+        elif logger_config.provider.upper() == AZURE_ENV:
             return Logger(config=logger_config)
+        else:
+            logging.error(f'Unimplemented initialization for core {logger_config.provider}')
 
     def get_topic(self, topic_name: str):
         topic_config = self.config.topic()
         if topic_config.provider.upper() == LOCAL_ENV:
             return LocalTopic(config=topic_config, topic_name=topic_name)
-        else:
+        elif topic_config.provider.upper() == AZURE_ENV:
             return Topic(config=topic_config, topic_name=topic_name)
+        else:
+            logging.error(f'Unimplemented initialization for core {topic_config.provider}')
 
     def get_storage_client(self):
         storage_config = self.config.storage()
         if storage_config.provider.upper() == LOCAL_ENV:
             return LocalStorageClient(storage_config)
-        else:
+        elif storage_config.provider.upper() == AZURE_ENV:
             return AzureStorageClient(storage_config)
+        else:
+            logging.error(f'Unimplemented initialization for core {storage_config.provider}')
 
     def __check_health(self):
         print('\x1b[32m ------------------------- \x1b[0m')

src/python_ms_core/core/config/config.py

@@ -67,30 +67,34 @@ def storage(self):
 
 class LocalConfig:
     def __init__(self):
-        self.provider = 'local'
+        self.provider = os.environ.get('PROVIDER', 'local')
+        self.queue_connection = os.environ.get('QUEUECONNECTION', 'amqp://guest:guest@localhost:5672')
+        self.queue_name = os.environ.get('LOGGERQUEUE', 'http://localhost:8100')
+        self.topic_connection = os.environ.get('QUEUECONNECTION', 'amqp://guest:guest@localhost:5672')
+        self.storage_connection = os.environ.get('STORAGECONNECTION', 'http://localhost:8100')
 
     def logger(self):
         return LogerConfig(
             provider=self.provider,
-            con_string='amqp://guest:guest@rabbitmq:5672/',
-            queue_name='http://localhost:8100'
+            con_string=self.queue_connection,
+            queue_name=self.queue_name
         )
 
     def queue(self):
         return QueueConfig(
             provider=self.provider,
-            con_string='amqp://guest:guest@rabbitmq:5672/',
-            queue_name='http://localhost:8100'
+            con_string=self.queue_connection,
+            queue_name=self.queue_name
         )
 
     def topic(self):
         return TopicConfig(
             provider=self.provider,
-            con_string='amqp://guest:guest@localhost:5672'
+            con_string=self.topic_connection
         )
 
     def storage(self):
         return StorageConfig(
             provider=self.provider,
-            con_string='http://localhost:8100'
+            con_string=self.storage_connection
         )

src/python_ms_core/core/logger/local_logger.py

@@ -1,49 +1,49 @@
 from .abstracts.logger_abstract import LoggerAbstract
 from ..queue.models.queue_message import QueueMessage
-from ..queue.queue import Queue
+from ..queue.local_queue import LocalQueue
 
 
 class LocalLogger(LoggerAbstract):
 
     def __init__(self, config=None):
         super().__init__(config=config)
-        self.queue_client = Queue(config)
+        self.queue_client = LocalQueue(config)
 
     def add_request(self, request_data):
         message = QueueMessage.data_from({
             'message': 'Add Request',
             'messageType': 'addRequest',
             'data': request_data
         })
-        self.queue_client.send_local(message)
+        self.queue_client.send(message)
 
     def info(self, message: str):
         msg = QueueMessage.data_from({
             'message': message,
             'messageType': 'info',
         })
-        self.queue_client.send_local(msg)
+        self.queue_client.send(msg)
 
     def debug(self, message: str):
         msg = QueueMessage.data_from({
             'message': message,
             'messageType': 'debug',
         })
-        self.queue_client.send_local(msg)
+        self.queue_client.send(msg)
 
     def warn(self, message: str):
         msg = QueueMessage.data_from({
             'message': message,
             'messageType': 'warn',
         })
-        self.queue_client.send_local(msg)
+        self.queue_client.send(msg)
 
     def error(self, message: str):
         msg = QueueMessage.data_from({
             'message': message,
             'messageType': 'error',
         })
-        self.queue_client.send_local(msg)
+        self.queue_client.send(msg)
 
     def record_metric(self, name: str, value: str):
         message = QueueMessage.data_from({
@@ -54,4 +54,4 @@ def record_metric(self, name: str, value: str):
                 'value': value
             }
         })
-        self.queue_client.send_local(message)
+        self.queue_client.send(message)

src/python_ms_core/core/queue/abstracts/queue_abstract.py

@@ -8,9 +8,6 @@ def __init__(self, config): pass
     @abstractmethod
     def send(self, data=None): pass
 
-    @abstractmethod
-    def send_local(self, data=None): pass
-
     @abstractmethod
     def add(self, data): pass
 

src/python_ms_core/core/queue/local_queue.py

@@ -0,0 +1,38 @@
+import json
+import requests
+from .config.queue_config import Config
+from .models.queue_message import QueueMessage
+from .abstracts.queue_abstract import QueueAbstract
+from ..resource_errors import ExceptionHandler
+
+
+class LocalQueue(QueueAbstract):
+    queue = list()
+
+    def __init__(self, config):
+        self.provider = Config(config=config)
+
+    @ExceptionHandler.decorated
+    def send(self, data=None):
+        if data:
+            message = QueueMessage.to_dict(data)
+            url = f'{self.provider.queue_name}/log'
+            try:
+                resp = requests.post(url, json=message)
+                print(resp.status_code)
+            except Exception as e:
+                print(e)
+                print(message)
+
+        self.queue = list()
+
+    def add(self, data):
+        if data is not None:
+            self.queue.insert(0, json.dumps(data))
+
+    def remove(self):
+        if len(self.queue) > 0:
+            self.queue.pop()
+
+    def get_items(self):
+        return self.queue

src/python_ms_core/core/storage/providers/local/local_file_entity.py

@@ -29,7 +29,7 @@ def upload(self, upload_stream):
         upload_path = f'{self.path}/{self.name}'
         upload_relative_path = f'{self.config.connection_string}{self.upload_path}{upload_path}'
         requests.post(upload_relative_path, files={'uploadFile': upload_stream})
-        self._get_remote_url = f'{upload_relative_path}/uploadFile'
+        self._get_remote_url = upload_relative_path
 
     @ExceptionHandler.decorated
     def get_remote_url(self):

src/python_ms_core/core/topic/local_topic.py

@@ -1,5 +1,3 @@
-import json
-
 import logging
 import threading
 import time
@@ -28,6 +26,7 @@ def __init__(self, config=None, topic_name=None):
         self.topic = topic_name
         self.config = config
         self.client = Config(config=config, topic_name=topic_name)
+        print(self.client.connection_string)
         params = pika.URLParameters(self.client.connection_string)
         self.connection = pika.BlockingConnection(params)
         self.channel = self.connection.channel()