updates

.env

+MONGO_URI=mongodb://intern_23:intern%40123@192.168.0.220:2717/?authSource=interns_b2_23&authMechanism=SCRAM-SHA-256
\ No newline at end of file

conf/application.conf

+[SERVICE]
+port=7889
+host=0.0.0.0
+
+
+[MONGO-DB]
+mongo_uri=$MONGO_URI
+
+

main.py

+from dotenv import load_dotenv
+
+load_dotenv()
+
+import uvicorn
+from fastapi import FastAPI
+
+from script.config.app_config import Service
+from script.service.employee import router
+
+app = FastAPI(title="Employee")
+app.include_router(router)
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host=Service.host, port=Service.port)

mongo_file.py

+import pymongo
+
+client = pymongo.MongoClient(
+    "mongodb://intern_23:intern%40123@192.168.0.220:2717/?authSource=interns_b2_23&authMechanism=SCRAM-SHA-256")
+db = client["interns_b2_23"]
+col = db["employee"]
+
+r = col.find()
+print(list(r))

script/config/app_config.py

+import os
+from configparser import SafeConfigParser
+
+config = SafeConfigParser()
+config.read('conf/application.conf')
+
+
+class Service:
+    port = int(config.getint("SERVICE", "port"))
+    host = config.get("SERVICE", "host")
+
+
+class Mongo:
+    mongo_uri: str = os.environ.get("MONGO_URI")
+    print(mongo_uri)

script/constants/__init__.py

+class APIEndpoints:
+    delete = "/delete"
+    save = "/save"
+    employee_base = "/employee"
+    insert = "/insert"
+    update = "/update"
+    find = "/find"
\ No newline at end of file

script/core/db/mongo/interns2023/__init__.py

+from script.config.app_config import Mongo
+from script.utils.mongo_utils import MongoConnect
+
+mongo_client = MongoConnect(uri=Mongo.mongo_uri)()
+

script/core/db/mongo/interns2023/employee.py

+from script.utils.mongo_utils import MongoCollectionBaseClass
+
+
+class EmployeeDetails(MongoCollectionBaseClass):
+
+    def __init__(self, mongo_client):
+        super().__init__(mongo_client=mongo_client, database="interns_b2_23", collection="employee")
+
+    def insert_item(self, data):
+        self.insert_one(data)
+
+    def list_items(self, data):
+        self.find_one(data)
+
+    def delete_item(self, data):
+        self.delete_one(data)
+
+    def update_item(self, query, data):
+        self.update_one(query, data)

script/core/errors/__init__.py

+class NameDoesNotExist(Exception):
+    pass
\ No newline at end of file

script/core/handlers/employee.py

+from script.core.db.mongo.interns2023 import mongo_client
+from script.core.db.mongo.interns2023.employee import EmployeeDetails
+from script.core.schema.employee import Employee
+
+
+class EmployeeData:
+
+    def __init__(self):
+        self.employee_col = EmployeeDetails(mongo_client=mongo_client)
+
+    def delete_data(self, emp_id):
+        try:
+            self.employee_col.delete_one(query={"emp_id": emp_id})
+        except ValueError:
+            raise ValueError("ID")
+        except Exception as e:
+            print(e, "Error Detected in Deletion")
+
+    def update_data(self, request_data: Employee):
+        try:
+            d = {"emp_id": request_data.emp_id,
+                 "emp_name": request_data.emp_name,
+                 "emp_department": request_data.emp_department,
+                 "emp_sal": request_data.emp_sal}
+            self.employee_col.update_one(query={"emp_id": request_data.emp_id}, data=d)
+        except Exception as e:
+            print(e, "Error Detected in Updation")
+
+    def find_data(self, emp_id):
+        try:
+            data = self.employee_col.find_one(query={"emp_id": emp_id})
+            return data
+        except Exception as e:
+            print(e, "Error Detected in Finding")
+
+    def insert_data(self, request_data: Employee):
+        try:
+            d = {"emp_id": request_data.emp_id,
+                 "emp_name": request_data.emp_name,
+                 "emp_department": request_data.emp_department,
+                 "emp_sal": request_data.emp_sal}
+            self.employee_col.insert_one(d)
+        except Exception as e:
+            print(e, "Error Detected in inserting")
+

script/core/schema/employee.py

+from typing import Optional
+
+from pydantic import BaseModel
+
+
+# define a model for Item
+class Employee(BaseModel):
+    emp_id: Optional[int]
+    emp_name: str
+    emp_department: str
+    emp_sal: int
+

script/core/schema/response.py

+from typing import Any, Optional
+
+from pydantic import BaseModel
+
+
+class DefaultResponse(BaseModel):
+    status: str = "failed"
+    message: str
+    data: Optional[Any]
+

script/service/employee.py

+import logging
+
+from fastapi.routing import APIRouter
+
+from script.constants import APIEndpoints
+from script.core.handlers.employee import EmployeeData
+from script.core.schema.employee import Employee
+from script.core.schema.response import DefaultResponse
+
+router = APIRouter(prefix=APIEndpoints.employee_base)
+handler = EmployeeData()
+
+
+@router.delete(APIEndpoints.delete)
+async def delete_item(emp_id: int):
+    """This API deletes an item"""
+    try:
+        handler.delete_data(emp_id)
+        return DefaultResponse(message="Successfully deleted", status="success")
+    except ValueError:
+        return DefaultResponse(message="Due to Value Error")
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Deletion failed due to server error")
+
+
+@router.post(APIEndpoints.update)
+async def update_item(request_data: Employee):
+    try:
+        handler.update_data(request_data)
+        return DefaultResponse(message="Successfully Updated", status="success")
+    except ValueError:
+        return DefaultResponse(message="Due to value error")
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Updation Failed due to server error")
+
+
+@router.get(APIEndpoints.find)
+async def find_item(emp_id: int):
+    try:
+        data = handler.find_data(emp_id)
+        return DefaultResponse(message="Successfully Found", status="success", data=data)
+    except ValueError:
+        return DefaultResponse(message="Due to value error")
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Finding Failed due to server error")
+
+
+@router.post(APIEndpoints.insert)
+def insert_item(request_data: Employee):
+    try:
+        handler.insert_data(request_data)
+        return DefaultResponse(message="Successfully Found", status="success")
+    except ValueError:
+        return DefaultResponse(message="Due to value error")
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Finding Failed due to server error")

script/utils/mongo_connection.py

+from fastapi import APIRouter
+from pymongo import MongoClient
+
+from script.config.app_config import Mongo
+
+router = APIRouter()
+
+# Create a MongoClient instance
+client = MongoClient(Mongo.mongo_uri)
+
+# Connect to a database
+db = client.mydatabase
+
+

script/utils/mongo_utils.py

+from typing import Dict
+
+from pymongo import MongoClient
+
+
+import logging
+
+
+class MongoConnect:
+    def __init__(self, uri):
+        try:
+            self.uri = uri
+            self.client = MongoClient(self.uri, connect=False)
+        except Exception as e:
+            logging.error(f"Exception in connection {(str(e))}")
+            raise e
+
+    def __call__(self, *args, **kwargs):
+        return self.client
+
+    def __repr__(self):
+        return f"Mongo Client(uri:{self.uri}, server_info={self.client.server_info()})"
+
+
+class MongoCollectionBaseClass:
+    def __init__(self, mongo_client, database, collection):
+        self.client = mongo_client
+        self.database = database
+        self.collection = collection
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}(database={self.database}, collection={self.collection}"
+
+    def insert_one(self, data: Dict):
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            db = self.client[database_name]
+            collection = db[collection_name]
+            response = collection.insert_one(data)
+            return response.inserted_id
+        except Exception as e:
+            logging.error(f"Error in inserting the data {str(e)}")
+            raise e
+
+    def find_one(self, query, filter_dict=None):
+        try:
+            database_name = self.client[self.database]
+            collection_name = database_name[self.collection]
+            if not filter_dict:
+                filter_dict = {"_id": 0}
+            response = collection_name.find_one(query, filter_dict)
+            return response
+        except Exception as e:
+            logging.error(f"Failed to fetch {str(e)}")
+            raise e
+
+
+    def update_one(self, query: Dict, data: Dict, upsert: bool = False, ):
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            db = self.client[database_name]
+            collection = db[collection_name]
+            response = collection.update_one(query, {"$set": data}, upsert=upsert)
+            return response.modified_count
+        except Exception as e:
+            logging.error(f"Failed to update one doc {str(e)}")
+            raise e
+
+
+    def delete_one(self, query: Dict):
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            db = self.client[database_name]
+            collection = db[collection_name]
+            response = collection.delete_one(query)
+            return response.deleted_count
+        except Exception as e:
+            logging.error(f"Failed to delete {str(e)}")
+            raise e