Merge pull request #30 from TaskarCenterAtUW/stage

Author: MashB

Dockerfile

@@ -4,4 +4,4 @@ COPY ./requirements.txt /code/requirements.txt
 RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
 COPY ./src /code/src
 EXPOSE 8080
-CMD ["uvicorn", "src.main:app", "--reload", "--host", "0.0.0.0", "--port", "8080"]
+CMD ["uvicorn", "src.main:app","--host", "0.0.0.0", "--port", "8080"]

README.md

@@ -30,12 +30,14 @@ VALIDATION_REQ_SUB=xxxx
 VALIDATION_RES_TOPIC=xxxx
 CONTAINER_NAME=xxxx
 AUTH_PERMISSION_URL=xxx
+MAX_CONCURRENT_MESSAGES=xxx
 
 ```
 
 The application connect with the `STORAGECONNECTION` string provided in `.env` file and validates downloaded zipfile using `python-osw-validation` package.
 `QUEUECONNECTION` is used to send out the messages and listen to messages.
 
+`MAX_CONCURRENT_MESSAGES` is the maximum number of concurrent messages that the service can handle. If not provided, defaults to 2
 
 ### How to Set up and Build
 Follow the steps to install the python packages required for both building and running the application

requirements.txt

@@ -1,10 +1,6 @@
-psutil==5.9.5
 fastapi==0.88.0
-python-dotenv==0.21.0
 pydantic==1.10.4
-python-ms-core==0.0.18
+python-ms-core==0.0.22
 uvicorn==0.20.0
-coverage==7.2.7
 html_testRunner==1.2.1
-httpx==0.24.1
-python-osw-validation==0.2.3
+python-osw-validation==0.2.4

src/assets/osw-upload.json

@@ -2,7 +2,7 @@
   "messageId": "c8c76e89f30944d2b2abd2491bd95337",
   "messageType": "workflow_identifier",
   "data": {
-    "file_upload_path": "https://tdeisamplestorage.blob.core.windows.net/osw/test_upload/valid.zip",
+    "file_upload_path": "https://tdeisamplestorage.blob.core.windows.net/tdei-storage-test/Archivew.zip",
     "user_id": "c59d29b6-a063-4249-943f-d320d15ac9ab",
     "tdei_project_group_id": "0b41ebc5-350c-42d3-90af-3af4ad3628fb"
   }

src/config.py

@@ -17,6 +17,7 @@ class Settings(BaseSettings):
     app_name: str = 'python-osw-validation'
     event_bus = EventBusSettings()
     auth_permission_url: str = os.environ.get('AUTH_PERMISSION_URL', None)
+    max_concurrent_messages: int = os.environ.get('MAX_CONCURRENT_MESSAGES', 2)
 
     @property
     def auth_provider(self) -> str:

src/main.py

@@ -9,6 +9,8 @@
 
 prefix_router = APIRouter(prefix='/health')
 
+# Have a reference to validator in the app object
+app.validator = None
 
 @lru_cache()
 def get_settings():
@@ -18,7 +20,8 @@ def get_settings():
 @app.on_event('startup')
 async def startup_event(settings: Settings = Depends(get_settings)) -> None:
     try:
-        OSWValidator()
+        # OSWValidator()
+        app.validator = OSWValidator()
     except:
         print('\n\n\x1b[31m Application startup failed due to missing or invalid .env file \x1b[0m')
         print('\x1b[31m Please provide the valid .env file and .env file should contains following parameters\x1b[0m')
@@ -34,6 +37,11 @@ async def startup_event(settings: Settings = Depends(get_settings)) -> None:
             child.kill()
         parent.kill()
 
+@app.on_event('shutdown')
+async def shutdown_event() -> None:
+    print('Shutting down the application')
+    if app.validator:
+        app.validator.stop_listening()
 
 @app.get('/', status_code=status.HTTP_200_OK)
 @prefix_router.get('/', status_code=status.HTTP_200_OK)

src/osw_validator.py

@@ -20,30 +20,26 @@ class OSWValidator:
     _settings = Settings()
 
     def __init__(self):
-        core = Core()
+        self.core = Core()
         options = {
             'provider': self._settings.auth_provider,
             'api_url': self._settings.auth_permission_url
         }
         listening_topic_name = self._settings.event_bus.upload_topic or ''
-        publishing_topic_name = self._settings.event_bus.validation_topic or ''
         self.subscription_name = self._settings.event_bus.upload_subscription or ''
-        self.listening_topic = core.get_topic(topic_name=listening_topic_name)
-        self.publishing_topic = core.get_topic(topic_name=publishing_topic_name)
-        self.logger = core.get_logger()
-        self.storage_client = core.get_storage_client()
-        self.auth = core.get_authorizer(config=options)
-        self.container_name = self._settings.event_bus.container_name
-        self.start_listening()
+        self.listening_topic = self.core.get_topic(topic_name=listening_topic_name, max_concurrent_messages=self._settings.max_concurrent_messages)
+        self.logger = self.core.get_logger()
+        self.storage_client = self.core.get_storage_client()
+        self.auth = self.core.get_authorizer(config=options)
+        self.listener_thread = threading.Thread(target=self.start_listening)
+        self.listener_thread.start()
 
     def start_listening(self):
         def process(message) -> None:
             if message is not None:
                 queue_message = QueueMessage.to_dict(message)
                 upload_message = Upload.data_from(queue_message)
-                process_thread = threading.Thread(target=self.validate, args=[upload_message])
-                process_thread.start()
-                # self.validate(upload_message)
+                self.validate(received_message=upload_message)
 
         self.listening_topic.subscribe(subscription=self.subscription_name, callback=process)
 
@@ -89,10 +85,11 @@ def send_status(self, result: ValidationResult, upload_message: Upload):
             'data': upload_message.data.to_json()
         })
         try:
-            self.publishing_topic.publish(data=data)
+            self.core.get_topic(topic_name=self._settings.event_bus.validation_topic).publish(data=data)
+            logger.info(f'Publishing message for : {upload_message.message_id}')
         except Exception as e:
-            print(e)
-        logger.info(f'Publishing message for : {upload_message.message_id}')
+            logger.error(f'Error occurred while publishing message for : {upload_message.message_id} with error: {e}')
+
 
     def has_permission(self, roles: List[str], queue_message: Upload) -> bool:
         try:
@@ -107,3 +104,6 @@ def has_permission(self, roles: List[str], queue_message: Upload) -> bool:
         except Exception as error:
             print('Error validating the request authorization:', error)
             return False
+
+    def stop_listening(self):
+        self.listener_thread.join(timeout=0) # Stop the thread during shutdown.Its still an attempt. Not sure if this will work.