first

.idea/sonarlint/issuestore/index.pb

+
+S
+#.idea/sonarlint/issuestore/index.pb,9\f\9fe84ebb15faf917b7def6236dba604453cc61e0
+]
+-.idea/sonarlint/securityhotspotstore/index.pb,8\8\88a9255124c95bdc913197c120a8d560edc59c8e
+^
+..idea/inspectionProfiles/profiles_settings.xml,1\e\1e9075f5bf079c01ef2c910709e91a497d262080
\ No newline at end of file

.idea/sonarlint/securityhotspotstore/index.pb

+
+S
+#.idea/sonarlint/issuestore/index.pb,9\f\9fe84ebb15faf917b7def6236dba604453cc61e0
+]
+-.idea/sonarlint/securityhotspotstore/index.pb,8\8\88a9255124c95bdc913197c120a8d560edc59c8e
+^
+..idea/inspectionProfiles/profiles_settings.xml,1\e\1e9075f5bf079c01ef2c910709e91a497d262080
\ No newline at end of file

conf/application.conf

+[SERVICE]
+port=7998
+host=0.0.0.0
+
+
+[MONGO-DB]
+mongo_uri=mongodb://192.168.0.220:2717/
\ No newline at end of file

main.py

 import uvicorn
+from fastapi import FastAPI
 
-from script.core.engine.mongo_connection import get_database
+from script.config.app_confing import Service
+from script.service.inventory import router
+
+app = FastAPI(title="Inventor Mt")
+app.include_router(router)
 
 if __name__ == "__main__":
-    get_database()
-    uvicorn.run("script.service.fastapi_calls:app")
+    uvicorn.run("main:app", host=Service.host, port=Service.port)

script/config/app_confing.py

+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:
+    mon = config.get('MONGO-DB', 'mongo_uri')

script/constants/__init__.py

+class APIEndpoints:
+    delete = "/delete"
+    save = "/save"
+    inventory_base = "/inventory"
+    insert = "/insert"
+    update = "/update"
+    find = "/find"

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

+database_name = "interns2023"

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

+from script.core.db.mongo.interns2023 import database_name
+from script.utils.mongo_utils import MongoCollectionBaseClass
+
+
+class InventoryCollection(MongoCollectionBaseClass):
+    def __init__(self, mongo_client):
+        super().__init__(mongo_client=mongo_client, database="interns_b2_23", collection="inventory")
+        self.mongo_client = mongo_client
+        self.database_name = database_name
+        self.collection_name = "inventory"
+
+    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, data):
+        self.update_one(data)

script/core/engine/mongo_connection.py

-from pymongo import MongoClient
-
-
-def get_database():
-    uri = "mongodb://intern_23:intern%40123@192.168.0.220:2717/?serverSelectionTimeoutMS=5000&connectTimeoutMS=10000&authSource=interns_b2_23&authMechanism=SCRAM-SHA-256"
-    client = MongoClient(uri)
-    return client['interns_b2_23']
-
-
-dbname = get_database()
-inventory = dbname["Vipul-inventory"]

script/core/errors/__init__.py

+class NameDoesNotExist(Exception):
+    pass

script/core/handlers/CRUD_operations.py

-from script.core.engine.mongo_connection import inventory
-
-inven_tory5 = {
-    "order_id": "8",
-    "date": "13/05/2022",
-    "customer_name": "karun",
-    "status": "not active",
-    "sales_order": 90
-}
-
-
-def insert_one():
-    inventory.insert_one(inven_tory5)
-    return inven_tory5
-
-def update_data():
-    inventory.update_one({"customer_name": "Akul"}, {"$set": {"customer_name": "vijay"}})
-
-def delete_data():
-    inventory.delete_one({"order_id": "1"})
-
-def find_data():
-    inventory.delete_one({"order_id": "1"})

script/core/handlers/inventory.py

+# class for CRUD operations
+from script.core.db.mongo.interns2023.inventory import InventoryCollection
+from script.core.errors import NameDoesNotExist
+
+class InventoryData:
+
+    def __init__(self):
+        self.inventory_col = InventoryCollection(mongo_client="")
+
+    def delete_data(self, order_id, int):
+        try:
+            d = {"order_id": order_id}
+            if 3 not in d.values():
+                raise NameDoesNotExist
+        except ValueError:
+            raise ValueError("ID")
+        except Exception as e:
+            print(e, "Error Detected in Deleion")
+    @staticmethod
+    def update_data(self, order_id: int):
+        try:
+            d = {"order_id": order_id}
+            if 2 not in d.values():
+                raise NameDoesNotExist
+        except Exception as e:
+            print(e, "Error Detected in Updation")
+
+    @staticmethod
+    def find_data(self, order_id: int):
+         try:
+             d = {"order_id": order_id}
+             if 2 not in d.values():
+                raise NameDoesNotExist
+         except Exception as e:
+             print(e, "Error Detected in Finding")
+
+    @staticmethod
+    def insert_data(self, order_id: int):
+        try:
+            d = {"order_id": 5}
+            if 5 not in d.values():
+                raise NameDoesNotExist
+        except Exception as e:
+            print(e, "Error Detected in inserting")
\ No newline at end of file

script/core/schema/responses.py

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

script/service/fastapi_calls.py

-from fastapi import FastAPI
-
-from script.core.handlers.CRUD_operations import insert_one, update_data, delete_data, find_data
-
-app = FastAPI()
-
-
-@app.get("/insert")
-async def insert_data():
-    data = insert_one()
-    print(data)
-    return "inserted"
-
-@app.get("/update")
-async def update_file():
-    data = update_data()
-    print(data)
-    return "update"
-
-@app.get("/delete")
-async def delete_file():
-    data = delete_data()
-    print(data)
-    return "delete"
-
-@app.get("/find")
-async def find():
-    data = find_data()
-    print(data)
-    return "found"
\ No newline at end of file

script/service/inventory.py

+import logging
+
+from fastapi.routing import APIRouter
+
+from script.constants import APIEndpoints
+from script.core.handlers.inventory import InventoryData
+from script.core.schema.responses import DefaultResponse
+
+router = APIRouter(prefix=APIEndpoints.inventory_base)
+handler = InventoryData()
+
+
+
+@router.post(APIEndpoints.delete)
+async def delete_item(order_id: int):
+    """This API deletes an item"""
+    try:
+        handler.delete_data(order_id, int)
+        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(order_id: int):
+    try:
+        handler.update_data(order_id, int)
+        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(order_id: int):
+    try:
+        handler.find_data(order_id, int)
+        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")
+
+
+@router.post(APIEndpoints.insert)
+async def insert_item(order_id: int):
+    try:
+        handler.insert_data(order_id, int)
+        return  DefaultResponse(message="Successfully inserted", status="success")
+    except ValueError:
+        return DefaultResponse(message="Due to value error")
+    except Exception as e:
+        logging.exception(e)
+        return  DefaultResponse(message="inserted Failed due to server error")
+
+

script/utils/mongo_connector.py

+from fastapi import APIRouter
+from pymongo import MongoClient
+
+router = APIRouter()
+
+# Create a MongoClient instance
+client = MongoClient("Mongo_uri")
+
+# Connect to a database
+db = client.mydatabase

script/utils/mongo_utils.py

+from typing import Dict, Optional
+
+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
+
+
+class MongoCollectionBaseClass:
+    def __init__(self, mongo_client, database, collection):
+        self.client = mongo_client
+        self.database = database
+        self.collection = 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, data: Dict, filter_dict: Optional[Dict] = None):
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            if filter_dict is None:
+                filter_dict = {"_id": 0}
+                db = self.client[database_name]
+                collection = db[collection_name]
+                response = collection.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