Cloud s3 data storage

Author: GoJo-Rika

app.py

@@ -6,7 +6,6 @@
 
 import certifi
 import pandas as pd
-import pymongo
 from dotenv import load_dotenv
 from fastapi import FastAPI, File, Request, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
@@ -32,17 +31,12 @@
 
 
 load_dotenv()
-username = os.getenv("MONGO_DB_USERNAME")
-password = os.getenv("MONGO_DB_PASSWORD")
-
-username = quote_plus(username)
-password = quote_plus(password)
+username = quote_plus(os.getenv("MONGO_DB_USERNAME"))
+password = quote_plus(os.getenv("MONGO_DB_PASSWORD"))
 
 mongo_db_url: str = f"mongodb+srv://{username}:{password}@cluster0.l5ee6dv.mongodb.net/?retryWrites=true&w=majority&appName=Cluster0"
 
-client = MongoClient(mongo_db_url, server_api=ServerApi("1"))
-
-client = pymongo.MongoClient(mongo_db_url, tlsCAFile=ca)
+client = MongoClient(mongo_db_url, server_api=ServerApi("1"), tlsCAFile=ca)
 
 
 database = client[DATA_INGESTION_DATABASE_NAME]

final_model/model.pkl

@@ -0,0 +1,11 @@
+import os
+
+
+class S3Sync:
+    def sync_folder_to_s3(self, folder: str, aws_bucket_url: str) -> None:
+        command = f"aws s3 sync {folder} {aws_bucket_url} "
+        os.system(command)
+
+    def sync_folder_from_s3(self, folder: str, aws_bucket_url: str) -> None:
+        command = f"aws s3 sync  {aws_bucket_url} {folder} "
+        os.system(command)

final_model/preprocessor.pkl

@@ -66,3 +66,8 @@
 MODEL_TRAINER_TRAINED_MODEL_NAME: str = "model.pkl"
 MODEL_TRAINER_EXPECTED_SCORE: float = 0.6
 MODEL_TRAINER_OVER_FIITING_UNDER_FITTING_THRESHOLD: float = 0.05
+
+
+## The bucket name should be unique globally and cannot be changed
+## Make a s3 bucket and change it below.
+TRAINING_BUCKET_NAME = "demo-networksecurity"

network_security/cloud/s3_syncer.py

@@ -1,9 +1,11 @@
 import sys
 
+from network_security.cloud.s3_syncer import S3Sync
 from network_security.components.data_ingestion import DataIngestion
 from network_security.components.data_transformation import DataTransformation
 from network_security.components.data_validation import DataValidation
 from network_security.components.model_trainer import ModelTrainer
+from network_security.constant.training_pipeline import TRAINING_BUCKET_NAME
 from network_security.entity.artifact_entity import (
     DataIngestionArtifact,
     DataTransformationArtifact,
@@ -24,6 +26,7 @@
 class TrainingPipeline:
     def __init__(self) -> None:
         self.training_pipeline_config = TrainingPipelineConfig()
+        self.s3_sync = S3Sync()
 
     def start_data_ingestion(self) -> DataIngestionArtifact:
         try:
@@ -43,7 +46,9 @@ def start_data_ingestion(self) -> DataIngestionArtifact:
         except Exception as e:
             raise NetworkSecurityException(e, sys)
 
-    def start_data_validation(self, data_ingestion_artifact: DataIngestionArtifact) -> DataValidationArtifact:
+    def start_data_validation(
+        self, data_ingestion_artifact: DataIngestionArtifact,
+    ) -> DataValidationArtifact:
         try:
             self.data_validation_config = DataValidationConfig(
                 training_pipeline_config=self.training_pipeline_config,
@@ -61,7 +66,9 @@ def start_data_validation(self, data_ingestion_artifact: DataIngestionArtifact)
         except Exception as e:
             raise NetworkSecurityException(e, sys)
 
-    def start_data_transformation(self, data_validation_artifact: DataValidationArtifact) -> DataTransformationArtifact:
+    def start_data_transformation(
+        self, data_validation_artifact: DataValidationArtifact,
+    ) -> DataTransformationArtifact:
         try:
             self.data_transformation_config = DataTransformationConfig(
                 training_pipeline_config=self.training_pipeline_config,
@@ -82,7 +89,9 @@ def start_data_transformation(self, data_validation_artifact: DataValidationArti
         except Exception as e:
             raise NetworkSecurityException(e, sys)
 
-    def start_model_trainer(self, data_transformation_artifact: DataTransformationArtifact) -> ModelTrainerArtifact:
+    def start_model_trainer(
+        self, data_transformation_artifact: DataTransformationArtifact,
+    ) -> ModelTrainerArtifact:
         try:
             self.model_trainer_config: ModelTrainerConfig = ModelTrainerConfig(
                 training_pipeline_config=self.training_pipeline_config,
@@ -102,6 +111,28 @@ def start_model_trainer(self, data_transformation_artifact: DataTransformationAr
         except Exception as e:
             raise NetworkSecurityException(e, sys)
 
+    ## Local artifact -----> s3 bucket
+    def sync_artifact_dir_to_s3(self) -> None:
+        try:
+            aws_bucket_url = f"s3://{TRAINING_BUCKET_NAME}/artifact/{self.training_pipeline_config.timestamp}"
+            self.s3_sync.sync_folder_to_s3(
+                folder=self.training_pipeline_config.artifact_dir,
+                aws_bucket_url=aws_bucket_url,
+            )
+        except Exception as e:
+            raise NetworkSecurityException(e, sys)
+
+    ## Local final model -----> s3 bucket
+    def sync_saved_model_dir_to_s3(self) -> None:
+        try:
+            aws_bucket_url = f"s3://{TRAINING_BUCKET_NAME}/final_model/{self.training_pipeline_config.timestamp}"
+            self.s3_sync.sync_folder_to_s3(
+                folder=self.training_pipeline_config.model_dir,
+                aws_bucket_url=aws_bucket_url,
+            )
+        except Exception as e:
+            raise NetworkSecurityException(e, sys)
+
     def run_pipeline(self) -> ModelTrainerArtifact:
         try:
             data_ingestion_artifact = self.start_data_ingestion()
@@ -114,6 +145,8 @@ def run_pipeline(self) -> ModelTrainerArtifact:
             model_trainer_artifact = self.start_model_trainer(
                 data_transformation_artifact=data_transformation_artifact,
             )
+            self.sync_artifact_dir_to_s3()
+            self.sync_saved_model_dir_to_s3()
 
             return model_trainer_artifact
         except Exception as e: