Merge pull request #3 from TaskarCenterAtUW/fix_configuration

Fix Multiple Issue

Author: sujata-m

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Change log
 
+### 0.0.7
+- Fixed configuration issues
+- Fixed all the review comments
+- Refactor code
+- Removed unused code
+- Added local dev env using TDEI-local-server Package
+- Added local env support to -
+  - Logger
+  - Queue
+  - Topic
+  - Storage
+
 ### 0.0.6
 - Added get_remote_url function in FileEntity class which will return the uploaded file url
 

README.md

@@ -24,7 +24,7 @@ All the cloud connections are initialized with `initialize` function of core whi
 Eg.
 ```python
 from python_ms_core import Core
-Core.initialize()
+core = Core() or Core(config='Local')
 ```
 The method analyzes the `.env` variables and does a health check on what components are available
 
@@ -56,12 +56,13 @@ This file will have to be generated or shared offline as per the developer requi
 Offers helper classes to help log the information. It is also used to record the audit messages
 as well as the analytics information required.
 
-Use `Core.get_logger()`
+Use `core.get_logger()`
 Eg.
 ```python
 from python_ms_core import Core
 
-logger = Core.get_logger()
+core = Core()
+logger = core.get_logger()
 
 # Record a metric
 logger.record_metric(name='test', value='test') # Metric and value
@@ -74,7 +75,8 @@ Eg.
 ```python
 from python_ms_core import Core
 
-logger = Core.get_logger()
+core = Core()
+logger = core.get_logger()
 
 # Record a metric
 logger.debug('Debug Message')
@@ -102,15 +104,12 @@ The configuration required by Queue and Topic is similar and will be handled via
 
 Topic can be accessed by the core method `get_topic`. This method takes two parameters
 1. topic name (required)
-2. callback function where messages will be received
 
 ```python
 from python_ms_core import Core
 
-def callback(instance):
-    print(instance.get_messages())
-
-topic = Core.get_topic(topic_name='topicName', callback=callback)
+core = Core()
+topic = core.get_topic(topic_name='topicName')
 
 ```
 
@@ -121,7 +120,8 @@ Once the topic object is got, use `publish` method to publish the message to top
 from python_ms_core import Core
 from python_ms_core.core.queue.models.queue_message import QueueMessage
 
-topic = Core.get_topic(topic_name='topicName')
+core = Core()
+topic = core.get_topic(topic_name='topicName')
 topic.publish(QueueMessage.data_from(
     {
         'message': 'Test message'
@@ -139,18 +139,26 @@ It takes one parameter
 ```python
 from python_ms_core import Core
 
-topic = Core.get_topic(topic_name='topicName')
-msg = topic.subscribe(subscription='subscriptionName')
-print(msg)
+core = Core()
+topic_object = topic = Core.get_topic(topic_name='topicName')
+
+def process(message):
+    print(f'Message Received: {message}')
+
+try:
+    topic_object.subscribe(subscription='subscriptionName')
+except Exception as e:
+    print(e)
 ```
 
 ### Storage
 For all the azure blobs and other storages, storage components will offer simple ways to upload/download and read the existing data.
 ```python
 from python_ms_core import Core
 
+core = Core()
 # Create storage client
-azure_client = Core.get_storage_client()
+azure_client = core.get_storage_client()
 
 # Get a container in the storage client
 container = azure_client.get_container(container_name='tdei-storage-test')
@@ -168,7 +176,8 @@ File upload is done only through read stream.
 from python_ms_core import Core
 from io import BytesIO, StringIO
 
-azure_client = Core.get_storage_client()
+core = Core()
+azure_client = core.get_storage_client()
 container = azure_client.get_container(container_name='tdei-storage-test')
 # Create an instance of `AzureFileEntity` with name and mime-type
 txt = 'foo\nbar\nbaz'

freeze_version.py

@@ -11,7 +11,7 @@
 
 build_date = date.today().strftime('%Y-%m-%d')
 
-version = '0.0.6'
+version = '0.0.7'
 
 with open(version_file_path, 'w+') as version_file:
     version_file.write("version = '{}'\n".format(version))

requirements.txt

@@ -18,3 +18,4 @@ six==1.16.0
 typing_extensions==4.4.0
 uamqp==1.6.3
 urllib3==1.26.14
+pika==1.3.1

src/example.py

@@ -1,58 +1,57 @@
 # Testing code
-
 import sys
+import os
+import time
 import uuid
+import random
 import datetime
 from io import BytesIO, StringIO
-import time
 
 from python_ms_core import Core
-from python_ms_core.core.queue.providers import azure_queue_config
 from python_ms_core.core.queue.models.queue_message import QueueMessage
 
-Core.initialize()
+core = Core(config='Local')
 print('Hello')
 
-topic = 'gtfs-flex-upload'
+topic = 'gtfspathways'
 subscription = 'upload-validation-processor-test'
 some_other_sub = 'usdufs'
 
-topic_config = azure_queue_config.AzureQueueConfig()
-
 
 def publish_messages(topic_name):
-    topic_object = Core.get_topic(topic_name=topic_name)
+    topic_object = core.get_topic(topic_name=topic_name)
     queue_message = QueueMessage.data_from({
         'message': str(uuid.uuid4().hex),
-        'data': {'a': 1}
+        'data': {'a': random.randint(0, 1000)}
     })
     topic_object.publish(data=queue_message)
+    print('Message Published')
 
 
 def subscribe(topic_name, subscription_name):
     def process(message):
-        print(message)
+        print(f'Message Received: {message}')
 
-    topic_object = Core.get_topic(topic_name=topic_name)
+    topic_object = core.get_topic(topic_name=topic_name)
     try:
         topic_object.subscribe(subscription=subscription_name, callback=process)
     except Exception as e:
         print(e)
 
 
-publish_messages(topic)
 subscribe(topic, subscription)
 
-azure_client = Core.get_storage_client()
+azure_client = core.get_storage_client()
 
-container = azure_client.get_container(container_name='tdei-storage-test')
+container = azure_client.get_container(container_name='gtfspathways')
 
 list_of_files = container.list_files()
-for single in list_of_files:
-    print(single.name)
-firstFile = list_of_files[2]
-# 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'
@@ -69,8 +68,12 @@ def process(message):
 print(test_file.get_remote_url())
 print('Uploaded Successfully')
 
-logger = Core.get_logger()
+logger = core.get_logger()
 logger.record_metric(name='test', value='test')
 
-logger = Core.get_logger(provider='Local')
-logger.record_metric(name='test', value='test')
+publish_messages(topic)
+time.sleep(2)
+
+# logger = core.get_logger()
+# logger.record_metric(name='test', value='test')
+os._exit(os.EX_OK)

src/python_ms_core/__init__.py

@@ -1,55 +1,50 @@
 import logging
-import os
-from dotenv import load_dotenv
-from .models.config import CoreConfig
 from .core.logger.logger import Logger
 from .core.logger.local_logger import LocalLogger
-from .core.queue.topic import Topic
-from .core.storage.providers.azure import azure_storage_client, azure_file_entity
-from .core.storage.providers.azure.azure_storage_config import AzureStorageConfig
+from .core.topic.topic import Topic
+from .core.topic.local_topic import LocalTopic
+from .core.storage.providers.azure.azure_storage_client import AzureStorageClient
+from .core.storage.providers.local.local_storage_client import LocalStorageClient
+from .core.config.config import CoreConfig, LocalConfig
 
-load_dotenv()
+LOCAL_ENV = 'LOCAL'
+AZURE_ENV = 'AZURE'
 
 
 class Core:
-
-    # FIXME: Fix the configuration
     def __init__(self, config=None):
-
-        if config is not None:
-            self.config = config
+        if config is not None and config.upper() == LOCAL_ENV:
+            self.config = LocalConfig()
         else:
-            self.config = CoreConfig.default()
-
-        # FIXME: remove local from here and accept only Azure for now
-        if self.config.provider == 'Local':
-            logging.error(f'Unimplemented initialization for core {self.config.provider}')
-
-    @staticmethod
-    def initialize():
-        return Core().__check_health()
-
-    @staticmethod
-    def get_logger(provider=None):
-        if provider and provider.lower() == 'local':
-            return LocalLogger(provider_config=provider)
+            self.config = CoreConfig()
+        self.__check_health()
+
+    def get_logger(self):
+        logger_config = self.config.logger()
+        if logger_config.provider.upper() == LOCAL_ENV:
+            return LocalLogger(config=logger_config)
+        elif logger_config.provider.upper() == AZURE_ENV:
+            return Logger(config=logger_config)
         else:
-            return Logger()
-
-    # @staticmethod
-    # def get_custom_queue():
-
-    @staticmethod
-    def get_topic(topic_name: str):
-        if topic_name is not None:
-            return Topic(config=CoreConfig.default(), topic_name=topic_name)
+            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)
+        elif topic_config.provider.upper() == AZURE_ENV:
+            return Topic(config=topic_config, topic_name=topic_name)
         else:
-            logging.error(f'Error: Topic name is required as function parameter')
-            return
-
-    @staticmethod
-    def get_storage_client():
-        return azure_storage_client.AzureStorageClient(AzureStorageConfig())
+            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)
+        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')
@@ -63,34 +58,34 @@ def __check_health(self):
 
         print(f'Configured for \x1b[32m {self.config.provider} \x1b[0m \n')
 
-        if self.config.provider == 'Azure':
-            logger_queue_name = os.environ.get('LOGGERQUEUE', None)
-            queue_connection = os.environ.get('QUEUECONNECTION', None)
-            storage_connection = os.environ.get('STORAGECONNECTION', None)
-
-            print('\x1b[31m > Checking Queue Connections\x1b[0m')
-            if queue_connection is None:
-                print('\x1b[33m Queue connection not available by default \x1b[0m')
-                print('\x1b[33m Please configure QUEUECONNECTION in .env file to ensure queue communication \x1b[0m')
-                print('\x1b[33m Note: All the logger functionality will be restricted to console \x1b[0m')
-            else:
-                print('\x1b[32m\x1b[40m Connected to Queues \x1b[0m')
+        logger_config = self.config.logger()
+        queue_config = self.config.queue()
+        storage_config = self.config.storage()
+        logger_queue_name = logger_config.queue_name
+        queue_connection = queue_config.connection_string
+        storage_connection = storage_config.connection_string
+        print('\x1b[31m > Checking Queue Connections\x1b[0m')
+        if queue_connection is None:
+            print('\x1b[33m Queue connection not available by default \x1b[0m')
+            print('\x1b[33m Please configure QUEUECONNECTION in .env file to ensure queue communication \x1b[0m')
+            print('\x1b[33m Note: All the logger functionality will be restricted to console \x1b[0m')
+        else:
+            print('\x1b[32m\x1b[40m Connected to Queues \x1b[0m')
 
-            print('\x1b[31m\n > Checking Storage Connections\x1b[0m')
+        print('\x1b[31m\n > Checking Storage Connections\x1b[0m')
 
-            if storage_connection is None:
-                print('\x1b[31m Storage connection not available \x1b[0m')
-                print('\x1b[31m Storage related functionalities will be unavailable \x1b[0m')
-                print('\x1b[31m Please configure STORAGECONNECTION in .env for storage functions \x1b[0m')
-            else:
-                print('\x1b[32m\x1b[40m Connected to Storage \x1b[0m')
+        if storage_connection is None:
+            print('\x1b[31m Storage connection not available \x1b[0m')
+            print('\x1b[31m Storage related functionalities will be unavailable \x1b[0m')
+            print('\x1b[31m Please configure STORAGECONNECTION in .env for storage functions \x1b[0m')
+        else:
+            print('\x1b[32m\x1b[40m Connected to Storage \x1b[0m')
 
-            print('\x1b[31m\n > Checking Logger Queue \x1b[0m')
-            if logger_queue_name is None:
-                print('\x1b[33m Logger queue is not configured. App will write to  \x1b[0m')
-                print('\x1b[32m tdei-ms-log queue \x1b[0m')
-            else:
-                print('\x1b[32m Logger configured \x1b[0m')
-            print('\x1b[32m ------------------------- \x1b[0m')
-            return True
+        print('\x1b[31m\n > Checking Logger Queue \x1b[0m')
+        if logger_queue_name is None:
+            print('\x1b[33m Logger queue is not configured. App will write to  \x1b[0m')
+            print('\x1b[32m tdei-ms-log queue \x1b[0m')
+        else:
+            print('\x1b[32m Logger configured \x1b[0m')
+        print('\x1b[32m ------------------------- \x1b[0m')
         return True