fix: file path -> import loop, linting

Author: Abhinavexists

app.py

@@ -1,44 +1,41 @@
-
 import os
+from pathlib import Path
+
 import certifi
 import pandas as pd
-from pathlib import Path
 from dotenv import load_dotenv
-
-from fastapi import FastAPI, Request, UploadFile, File
-from fastapi.responses import Response
+from fastapi import FastAPI, File, Request, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import Response
 from fastapi.templating import Jinja2Templates
-from starlette.responses import RedirectResponse
-from uvicorn import run
-
 from pymongo.mongo_client import MongoClient
 from pymongo.server_api import ServerApi
+from starlette.responses import RedirectResponse
+from uvicorn import run
 
-from src.utils.utils import load_object
-from src.utils.estimator import NetworkModel
-from src.pipeline.training_pipeline import TrainingPipeline
 from src.constant.training_pipeline import (
-    DATA_INGESTION_DATABASE_NAME,
     DATA_INGESTION_COLLECTION_NAME,
+    DATA_INGESTION_DATABASE_NAME,
 )
-
 from src.exception.exception import CustomException
-from src.logging.logging import logging
+from src.logger.logger import logging
+from src.pipeline.training_pipeline import TrainingPipeline
+from src.utils.estimator import NetworkModel
+from src.utils.utils import load_object
 
 load_dotenv()
 
-uri = os.getenv('MONGO_URL')
-logging.info('MONGODB is set and ready to use')
+uri = os.getenv("MONGO_URL")
+logging.info("MONGODB is set and ready to use")
 
 client = MongoClient(
     uri,
     tls=True,
     tlsCAFile=certifi.where(),
     tlsAllowInvalidCertificates=False,
-    server_api=ServerApi('1'),
-    serverSelectionTimeoutMS=10000
-    )
+    server_api=ServerApi("1"),
+    serverSelectionTimeoutMS=10000,
+)
 
 database = client[DATA_INGESTION_DATABASE_NAME]
 collection = database[DATA_INGESTION_COLLECTION_NAME]
@@ -47,28 +44,31 @@
 origin = ["*"]
 
 app.add_middleware(
-    CORSMiddleware,
-    allow_origins = origin,
-    allow_methods = ["*"],
-    allow_headers = ["*"],
+    CORSMiddleware,  # type: ignore[arg-type]
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
 )
 
 templates = Jinja2Templates(directory="./templates")
 
+
 @app.get("/", tags=["authentication"])
 async def index():
     return RedirectResponse(url="/docs")
 
+
 @app.get("/train")
 async def train():
     try:
         train_pipeline = TrainingPipeline()
         train_pipeline.run_pipeline()
         return Response("Training is Successful")
-    
+
     except Exception as e:
         raise CustomException(e) from e
-    
+
+
 @app.get("/predict")
 async def predict_route(request: Request, file: UploadFile = File(...)):
     try:
@@ -83,15 +83,18 @@ async def predict_route(request: Request, file: UploadFile = File(...)):
         y_pred = Network_model.predict(df)
         print(y_pred)
 
-        df['predicted_column'] = y_pred
-        print(df['predicted_column'])
+        df["predicted_column"] = y_pred
+        print(df["predicted_column"])
 
         df.to_csv("prediction_output/output.csv")
-        index_html = df.to_html(classes='table table-stripped')
-        return templates.TemplateResponse("index.html", {"request": request, "table": index_html})
+        index_html = df.to_html(classes="table table-stripped")
+        return templates.TemplateResponse(
+            "index.html", {"request": request, "table": index_html}
+        )
 
     except Exception as e:
         raise CustomException(e) from e
 
+
 if __name__ == "__main__":
-    run(app, host='localhost', port=8000)
+    run(app, host="localhost", port=8000)

main.py

@@ -1,45 +1,51 @@
 from src.components.data_ingestion import DataIngestion
 from src.components.data_transformation import DataTransformation
-from src.exception.exception import CustomException
-from src.logging.logging import logging
 from src.components.data_validation import DataValidation
 from src.components.model_trainer import ModelTrainer, ModelTrainerConfig
 from src.entity.config import (
     DataIngestionConfig,
-    TrainingPipelineConfig,
+    DataTransformationConfig,
     DataValidationConfig,
-    DataTransformationConfig
+    TrainingPipelineConfig,
 )
+from src.exception.exception import CustomException
+from src.logger.logger import logging
 
 if __name__ == "__main__":
     try:
         training_pipeline_config = TrainingPipelineConfig()
         data_ingestion_config = DataIngestionConfig(training_pipeline_config)
         data_ingestion = DataIngestion(data_ingestion_config)
-        logging.info('Initiate the data Ingestion')
+        logging.info("Initiate the data Ingestion")
 
         data_ingestion_artifact = data_ingestion.initiate_data_ingestion()
-        logging.info('Data Initialization completed')
+        logging.info("Data Initialization completed")
         print(data_ingestion_artifact)
 
         data_validation_config = DataValidationConfig(training_pipeline_config)
-        data_validation = DataValidation(data_ingestion_artifact, data_validation_config)
-        logging.info('Initiate the data validation')
+        data_validation = DataValidation(
+            data_ingestion_artifact, data_validation_config
+        )
+        logging.info("Initiate the data validation")
 
         data_validation_artifact = data_validation.initiate_data_validation()
-        logging.info('Data validation completed')
+        logging.info("Data validation completed")
         print(data_validation_artifact)
 
         data_transformation_config = DataTransformationConfig(training_pipeline_config)
-        data_transformation = DataTransformation(data_validation_artifact, data_transformation_config)
-        data_transformation_artifact = data_transformation.initiate_data_transformation()
-        logging.info('Data Transformation Complete')
+        data_transformation = DataTransformation(
+            data_validation_artifact, data_transformation_config
+        )
+        data_transformation_artifact = (
+            data_transformation.initiate_data_transformation()
+        )
+        logging.info("Data Transformation Complete")
         print(data_transformation_artifact)
 
         model_trainer_config = ModelTrainerConfig(training_pipeline_config)
         model_trainer = ModelTrainer(model_trainer_config, data_transformation_artifact)
         model_trainer_artifact = model_trainer.initiate_model_trainer()
-        logging.info('Model Training Complete')
+        logging.info("Model Training Complete")
         print(model_trainer_artifact)
 
     except Exception as e:

mongo.py

@@ -1,71 +1,75 @@
-
-import os
 import json
+import os
+from pathlib import Path
+
 import certifi
 import pandas as pd
-from pathlib import Path
+from dotenv import load_dotenv
 from pymongo.mongo_client import MongoClient
 from pymongo.server_api import ServerApi
-from dotenv import load_dotenv
 
 from src.exception.exception import CustomException
-from src.logging.logging import logging
+from src.logger.logger import logging
 
 load_dotenv()
 
 # Create a new client and connect to the server
 # client = MongoClient(uri, server_api=ServerApi('1'))
 
-uri = os.getenv('MONGO_URL')
-logging.info('MONGODB is set and ready to use')
+uri = os.getenv("MONGO_URL")
+logging.info("MONGODB is set and ready to use")
 
-class NetworkDataExtract():
+
+class NetworkDataExtract:
     def __init__(self):
         try:
             pass
-        
+
         except Exception as e:
             raise CustomException(e)
-        
+
     def convert_csv_to_json(self, file_path: Path):
         try:
             data = pd.read_csv(file_path)
             data.reset_index(drop=True, inplace=True)
             records = list(json.loads(data.T.to_json()).values())
-            logging.info('records are ready')
+            logging.info("records are ready")
             return records
-        
+
         except Exception as e:
             raise CustomException(e)
-    
+
     def insert_data_to_mongo(self, records, database, collections):
         try:
             self.records = records
             self.database = database
             self.collections = collections
             self.mongo_client = MongoClient(
-                                            uri,
-                                            tls=True,
-                                            tlsCAFile=certifi.where(),
-                                            tlsAllowInvalidCertificates=False,
-                                            server_api=ServerApi('1'),
-                                            serverSelectionTimeoutMS=10000
-                                        )
+                uri,
+                tls=True,
+                tlsCAFile=certifi.where(),
+                tlsAllowInvalidCertificates=False,
+                server_api=ServerApi("1"),
+                serverSelectionTimeoutMS=10000,
+            )
             self.database = self.mongo_client[self.database]
             self.collections = self.database[self.collections]
             self.collections.insert_many(records)
-            logging.info('collections is ready')
+            logging.info("collections is ready")
 
             return len(self.records)
-        
+
         except Exception as e:
             raise CustomException(e)
-        
+
+
 if __name__ == "__main__":
-    FILE_PATH = Path('data/phisingData.csv')
-    DATABASE = 'NetworkDatabase'
-    Collections = 'NetworkData'
+    FILE_PATH = Path("data/phisingData.csv")
+    DATABASE = "NetworkDatabase"
+    Collections = "NetworkData"
     nde = NetworkDataExtract()
     records = nde.convert_csv_to_json(FILE_PATH)
-    no_of_records = nde.insert_data_to_mongo(records=records, database=DATABASE, collections=Collections)
-    print(no_of_records)
+    no_of_records = nde.insert_data_to_mongo(
+        records=records, database=DATABASE, collections=Collections
+    )
+    print(no_of_records)

src/components/data_ingestion.py

@@ -1,27 +1,29 @@
 import os
+from pathlib import Path
+
 import numpy as np
 import pandas as pd
-from pathlib import Path
 from dotenv import load_dotenv
 from pymongo.mongo_client import MongoClient
 from sklearn.model_selection import train_test_split
 
-from src.entity.config import DataIngestionConfig
 from src.entity.artifact import DataIngestionArtifact
+from src.entity.config import DataIngestionConfig
 from src.exception.exception import CustomException
-from src.logging.logging import logging
+from src.logger.logger import logging
 
 load_dotenv()
 
-MONGO_URL = os.getenv('MONGO_URL')
+MONGO_URL = os.getenv("MONGO_URL")
+
 
 class DataIngestion:
     def __init__(self, data_ingestion_config: DataIngestionConfig):
         try:
             self.data_ingestion_config = data_ingestion_config
         except Exception as e:
             raise CustomException(e)
-        
+
     def export_collection_as_dataframe(self) -> pd.DataFrame:
         """Fetch collection data from MongoDB and return it as a DataFrame."""
         try:
@@ -38,24 +40,24 @@ def export_collection_as_dataframe(self) -> pd.DataFrame:
 
             df.replace({"na": np.nan}, inplace=True)
             return df
-        
+
         except Exception as e:
             raise CustomException(e)
-        
+
     def export_data_into_feature_store(self, dataframe: pd.DataFrame) -> pd.DataFrame:
         """Save the fetched data into a local feature store CSV."""
         try:
             feature_store_file_path = self.data_ingestion_config.feature_store_file_path
             feature_store_file_path.parent.mkdir(parents=True, exist_ok=True)
-            
+
             dataframe.to_csv(feature_store_file_path, index=False, header=True)
             logging.info(f"Feature store saved at: {feature_store_file_path}")
-            
+
             return dataframe
-        
+
         except Exception as e:
             raise CustomException(e)
-        
+
     def split_data_into_test_train(self, dataframe: pd.DataFrame) -> None:
         """Split the dataset into train and test sets and store them."""
         try:
@@ -76,23 +78,25 @@ def split_data_into_test_train(self, dataframe: pd.DataFrame) -> None:
             train_set.to_csv(train_path, index=False, header=True)
             test_set.to_csv(test_path, index=False, header=True)
 
-            logging.info(f"Train and test data exported:\nTrain → {train_path}\nTest → {test_path}")
+            logging.info(
+                f"Train and test data exported:\nTrain → {train_path}\nTest → {test_path}"
+            )
 
         except Exception as e:
             raise CustomException(e)
-        
+
     def initiate_data_ingestion(self) -> DataIngestionArtifact:
         """Main pipeline that runs data fetching, storing, and splitting."""
         try:
             dataframe = self.export_collection_as_dataframe()
             dataframe = self.export_data_into_feature_store(dataframe)
             self.split_data_into_test_train(dataframe)
-            
+
             artifact = DataIngestionArtifact(
                 train_file_path=self.data_ingestion_config.training_file_path,
                 test_file_path=self.data_ingestion_config.testing_file_path,
             )
-            
+
             logging.info("Data ingestion completed successfully.")
             return artifact