Merge pull request #2 from TaskarCenterAtUW/develop

Develop to Main new version 0.0.2

Author: sujata-m

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Change log
 
+### 0.0.2
+- Added filtered exception handling
+- Removed unused packages
+- Fixed Topic subscription
+
 ### 0.0.1
 - Introduces Topic and subscription methods in Core.
 - Added methods

README.md

@@ -100,12 +100,17 @@ The configuration required by Queue and Topic is similar and will be handled via
 
 ### Accessing a specific topic
 
-Topic can be accessed by the core method `get_topic`. This method takes one parameter 
+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
-topic = Core.get_topic(topic_name='topicName')
+
+def callback(instance):
+    print(instance.get_messages())
+
+topic = Core.get_topic(topic_name='topicName', callback=callback)
 
 ```
 

freeze_version.py

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

requirements.txt

@@ -2,22 +2,6 @@ azure-common==1.1.28
 azure-core==1.26.1
 azure-servicebus==7.0.0
 azure-storage-blob==12.14.1
-certifi==2022.12.7
-cffi==1.15.1
-charset-normalizer==2.1.1
-cryptography==38.0.4
-gitdb==4.0.10
-GitPython==3.1.29
-idna==3.4
-isodate==0.6.1
-msrest==0.7.1
-oauthlib==3.2.2
-pycparser==2.21
 python-dotenv==0.21.0
 requests==2.28.1
-requests-oauthlib==1.3.1
-six==1.16.0
-smmap==5.0.0
-typing_extensions==4.4.0
-uamqp==1.6.3
-urllib3==1.26.13
+GitPython==3.1.29

src/example.py

@@ -1,39 +1,24 @@
 # Testing code
 
+import sys
 import uuid
 import datetime
 from io import BytesIO, StringIO
-from dotenv import load_dotenv
+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
 
-load_dotenv()
-
 Core.initialize()
 print('Hello')
 
 topic = 'gtfs-flex-upload'
-subscription = 'uploadprocessor'
+subscription = 'upload-validation-processor-test'
 some_other_sub = 'usdufs'
 
 topic_config = azure_queue_config.AzureQueueConfig()
 
-topicObject1 = Core.get_topic(topic_name=topic)
-topicObject2 = Core.get_topic(topic_name='tipic')
-
-print()
-
-queue_message = QueueMessage.data_from({
-    'message': str(uuid.uuid4().hex),
-    'data': {'a': 1}
-})
-
-topicObject1.publish(data=queue_message)
-
-msg = topicObject1.subscribe(subscription=subscription)
-print(f'Received message: {msg}')
-
 azure_client = Core.get_storage_client()
 
 container = azure_client.get_container(container_name='tdei-storage-test')
@@ -61,3 +46,26 @@
 
 logger = Core.get_logger(provider='Local')
 logger.record_metric(name='test', value='test')
+
+
+def on_connect_callback(instance):
+    print('Connected with result code {}'.format(instance.get_messages()))
+    time.sleep(30)
+    sys.exit()
+
+
+topicObject1 = Core.get_topic(topic_name=topic, callback=on_connect_callback)
+topicObject2 = Core.get_topic(topic_name='tipic')
+try:
+    topicObject2.subscribe(subscription=subscription)
+except Exception as e:
+    print(e)
+
+queue_message = QueueMessage.data_from({
+    'message': str(uuid.uuid4().hex),
+    'data': {'a': 1}
+})
+
+topicObject1.publish(data=queue_message)
+
+topicObject1.subscribe(subscription=subscription)

src/python_ms_core/__init__.py

@@ -34,9 +34,9 @@ def get_logger(provider=None):
     # def get_custom_queue():
 
     @staticmethod
-    def get_topic(topic_name=None):
+    def get_topic(topic_name=None, callback=None):
         if topic_name is not None:
-            return Topic(config=CoreConfig.default(), topic_name=topic_name)
+            return Topic(config=CoreConfig.default(), topic_name=topic_name, callback=callback)
         else:
             logging.error(f'Unimplemented initialization for core, Topic name required!')
             return

src/python_ms_core/core/queue/models/queue_message.py

@@ -1,8 +1,8 @@
 import json
 from datetime import datetime
 from typing import Union, List
-from dataclasses import dataclass, asdict
-from typing import Optional
+from dataclasses import dataclass
+from ...resource_errors import ExceptionHandler
 
 
 class Validations:
@@ -42,6 +42,7 @@ def send(self):
     def get_items(self):
         return self.queue
 
+    @ExceptionHandler.decorated
     def data_from(self):
         data = self
         if isinstance(data, str):
@@ -56,7 +57,8 @@ def data_from(self):
                 return QueueMessage(**kwargs)
             except Exception as e:
                 error = str(e).replace('QueueMessage', 'Invalid parameter,')
-                raise TypeError(error)
+                error = error.replace('__init__()', 'QueueMessage')
+                raise TypeError(f'{error} \nStatus code: 400')
 
     def to_dict(self):
         if isinstance(self, QueueMessage):

src/python_ms_core/core/queue/providers/azure_service_bus_topic.py

@@ -16,5 +16,5 @@ def __init__(self, topic_name=None, queue_name=None):
         self.logger_queue_name = queue_name or os.environ.get('LOGGERQUEUE', 'tdei-ms-log')
         self.provider = config.provider
         self.connection_string = config.connection_string
-        self.client = ServiceBusClient.from_connection_string(conn_str=self.connection_string)
+        self.client = ServiceBusClient.from_connection_string(conn_str=self.connection_string, retry_total=0)
         self.sender = ServiceBusMessage

src/python_ms_core/core/queue/queue.py

@@ -29,11 +29,8 @@ def send(self, data=None):
             message = QueueMessage.to_dict(data)
             with self.azure.client:
                 sender = self.azure.client.get_queue_sender(queue_name=self.azure.queue_name)
-                try:
-                    with sender:
-                        sender.send_messages(self.azure.sender(json.dumps(message)))
-                except Exception:
-                    print(message)
+                with sender:
+                    sender.send_messages(self.azure.sender(json.dumps(message)))
         self.queue = list()
 
     def send_local(self, data=None):

src/python_ms_core/core/queue/topic.py

@@ -6,8 +6,10 @@
 
 
 class Topic:
-    def __init__(self, config=None, topic_name=None):
+    def __init__(self, config=None, topic_name=None, callback=None):
         self.topic = topic_name
+        self._callback = callback
+        self._messages = []
         if config.provider == 'Azure':
             try:
                 self.azure = AzureServiceBusTopic(topic_name=topic_name)
@@ -19,16 +21,22 @@ def __init__(self, config=None, topic_name=None):
     @ExceptionHandler.decorated
     def subscribe(self, subscription=None):
         if subscription is not None:
-            messages = []
-            with self.azure.client.get_subscription_receiver(topic_name=self.azure.topic, subscription_name=subscription) as receiver:
-                received_msgs = receiver.receive_messages(max_message_count=10, max_wait_time=5)
-                for message in received_msgs:
-                    queue_message = QueueMessage.data_from(str(message))
-                    messages.append(queue_message)
-                    receiver.complete_message(message)
-            return messages
+            with self.azure.client:
+                topic_receiver = self.azure.client.get_subscription_receiver(self.azure.topic, subscription_name=subscription)
+                with topic_receiver:
+                    for message in topic_receiver:
+                        queue_message = QueueMessage.data_from(str(message))
+                        self._messages = queue_message
+                        if self._callback and callable(self._callback):
+                            self._callback(self)
+                        topic_receiver.complete_message(message)
+
         else:
-            logging.error(f'Unimplemented initialization for core {self.config.provider}, Subscription name is required!')
+            logging.error(
+                f'Unimplemented initialization for core {self.config.provider}, Subscription name is required!')
+
+    def get_messages(self):
+        return self._messages
 
     @ExceptionHandler.decorated
     def publish(self, data=None):