Adding SQS trigger for the lambda to convert stored DB data to CSV files (#76)

Author: NishanthMuruganantham

aws/mens_t20i_dataset_stack.py

@@ -7,7 +7,9 @@
     Stack,
     RemovalPolicy,
     aws_events as events,
+    aws_lambda_event_sources as lambda_event_sources,
     aws_events_targets as events_targets,
+    aws_sqs as sqs,
     aws_s3_notifications as s3_notifications,
 )
 import boto3
@@ -49,6 +51,14 @@ def __init__(
             billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST
         )
 
+        ########################################  SQS Configurations #####################################################
+        sqs_queue_to_send_delayed_message_when_new_file_is_downloaded = sqs.Queue(
+            self,
+            f"{stack_name}-sqs_queue_to_send_delayed_message_when_new_file_is_downloaded",
+            visibility_timeout=Duration.minutes(15),
+            delivery_delay=Duration.minutes(5),
+        )
+
         ########################################  SECRET MANAGER Configurations ##########################################
         __db_secrets = get_secret_from_secrets_manager(self._secret_manager_client, "db_secret")
         __kaggle_secrets = get_secret_from_secrets_manager(self._secret_manager_client, "kaggle_credentials")
@@ -76,6 +86,7 @@ def __init__(
             handler="download_from_cricsheet_lambda_function.handler",
             runtime=_lambda.Runtime.PYTHON_3_11,
             environment={
+                "DELAYED_SQS_QUEUE_URL": sqs_queue_to_send_delayed_message_when_new_file_is_downloaded.queue_url,
                 "DOWNLOAD_BUCKET_NAME": cricsheet_data_downloading_bucket.bucket_name,
                 "DYNAMODB_TABLE_NAME": dynamodb_to_store_file_status_data.table_name,
                 "THRESHOLD_FOR_NUMBER_OF_FILES_TO_BE_SENT_FOR_PROCESSING": THRESHOLD_FOR_NUMBER_OF_FILES_TO_BE_SENT_FOR_PROCESSING,
@@ -101,6 +112,8 @@ def __init__(
                 resources=["*"],
             )
         )
+        # Granting permissions to the Lambda function to send messages to the SQS queue
+        sqs_queue_to_send_delayed_message_when_new_file_is_downloaded.grant_send_messages(cricsheet_data_downloading_lambda)
         # EventBridge Rule to trigger the Lambda every Monday at 12:00 AM UTC
         event_bridge_rule_to_trigger_cricsheet_data_downloading_lambda = events.Rule(
             self,
@@ -270,6 +283,13 @@ def __init__(
                 resources=["*"],
             )
         )
+        # Trgerring the convert_mongodb_data_to_csv lambda function with delayed SQS message
+        convert_mongodb_data_to_csv_lambda.add_event_source(
+            lambda_event_sources.SqsEventSource(
+                sqs_queue_to_send_delayed_message_when_new_file_is_downloaded,
+                batch_size=1,
+            )
+        )
 
         # Lambda function to upload the dataset to KAGGLE and create a new version of dataset
         upload_dataset_to_kaggle_lambda = _lambda.Function(

src/mens_t20i_data_collector/_lambdas/download_from_cricsheet/download_from_cricsheet_lambda_function.py

@@ -69,7 +69,9 @@ def upload_new_json_data_files_for_data_processing(self, downloaded_zip_file_pat
             raise
 
         new_files = self._seggregate_new_files_from_downloaded_zip()
-        self._upload_new_json_files_to_s3_and_send_sns_notification(new_files=new_files)
+        self._upload_new_json_files_to_s3(new_files=new_files)
+        if new_files:
+            self._trigger_an_sqs_message_whenever_new_file_is_downloaded(new_files=new_files)
 
     def _list_all_files_from_dynamo_db(self) -> Set:
         response = self._dynamo_db_to_store_file_data_extraction_status.scan(ProjectionExpression="file_name")
@@ -87,7 +89,25 @@ def _seggregate_new_files_from_downloaded_zip(self) -> List:
         logger.info(f"Total newly downloaded files: {len(new_files)}")
         return new_files
 
-    def _upload_new_json_files_to_s3_and_send_sns_notification(self, new_files: List):
+    def _trigger_an_sqs_message_whenever_new_file_is_downloaded(self, new_files: List[str]):
+        """
+        This function will trigger an SQS message whenever a new file is downloaded
+        :param new_files: List of new files downloaded
+        :return: None
+        """
+        sqs_client = boto3.client("sqs")
+        queue_url = get_environmental_variable_value("DELAYED_SQS_QUEUE_URL")
+        message_body = {
+            "message": "New files downloaded from Cricsheet",
+            "new_files": new_files,
+        }
+        response = sqs_client.send_message(
+            QueueUrl=queue_url,
+            MessageBody=str(message_body)
+        )
+        logger.info(f"Message sent to SQS: {response['MessageId']}")
+
+    def _upload_new_json_files_to_s3(self, new_files: List):
         for file in new_files[:self._threshold_for_number_of_files_to_be_sent_for_processing]:
             file_path = f"{self._extraction_directory}/{file}"
             key = f"{self._s3_folder_to_store_cricsheet_data}/{self._s3_folder_to_store_processed_json_files_zip}/{file}"

src/mens_t20i_data_collector/_lambdas/upload_dataset_to_kaggle/upload_dataset_to_kaggle_lambda.py

@@ -38,7 +38,6 @@ def upload_dataset_to_kaggle(self):
         Uploads the dataset to Kaggle.
         """
         os.makedirs(self._folder_to_keep_the_files_to_upload, exist_ok=True)
-        self._create_metadata_json_file()
         self._download_dataset_files_from_s3()
         self._authenticate_to_kaggle_and_upload_dataset()
 
@@ -58,10 +57,11 @@ def _authenticate_to_kaggle_and_upload_dataset(self):
             team_1 = last_match_details["team_1"]
             team_2 = last_match_details["team_2"]
             date = last_match_details["date"]
+            self._create_metadata_json_file(date)
             api.dataset_create_version(
                 delete_old_versions=True,
                 folder=self._folder_to_keep_the_files_to_upload,
-                version_notes=f"Dataset updated till the match between {team_1} and {team_2} on {date}",
+                version_notes=f"Updated till the match between {team_1} and {team_2} on {date}",
             )
             logger.info("Dataset uploaded to Kaggle successfully")
         except Exception as e:
@@ -83,13 +83,14 @@ def _create_kaggle_json_file(self):
         os.environ["KAGGLE_CONFIG_DIR"] = self._temporary_directory
         logger.info(f"kaggle.json file created at the temporary path {kaggle_json_file_path}")
 
-    def _create_metadata_json_file(self):
+    def _create_metadata_json_file(self, date):
         """
         Creates a metadata.json file with the dataset metadata for Kaggle API.
         """
         logger.info("Creating metadata.json file...")
         metadata = {
             "id": f"{self._kaggle_username}/{get_environmental_variable_value('KAGGLE_DATASET_SLUG')}",
+            "subtitle": f"Complete T20I data updated till {date} for ML & match analysis"
         }
         metadata_file_path = os.path.join(self._folder_to_keep_the_files_to_upload, "dataset-metadata.json")
         with open(metadata_file_path, "w", encoding="utf-8") as metadata_file: