crud operation in mongo

.env

+#mongodb
+MONGO_URI=mongodb://intern_23:intern%40123@192.168.0.220:2717/?authSource=interns_b2_23&authMechanism=SCRAM-SHA-256
+
+
+

conf/application.conf

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

main.py

+from dotenv import load_dotenv
+load_dotenv()
 import uvicorn
 from fastapi import FastAPI
 

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["Vipul-inventory"]
+
+r = col.find()
+print(list(r))

script/config/app_confing.py

+import os
 from configparser import SafeConfigParser
 
+
 config = SafeConfigParser()
 config.read('conf/application.conf')
 
@@ -10,4 +12,5 @@ class Service:
 
 
 class Mongo:
-    mon = config.get('MONGO-DB', 'mongo_uri')
+    mongo_uri: str = os.environ.get("MONGO_URI")
+    print(mongo_uri)

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

-database_name = "interns2023"
+from script.config.app_confing import Mongo
+from script.utils.mongo_utils import MongoConnect
+
+mongo_client = MongoConnect(uri=Mongo.mongo_uri)()

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"
+        super().__init__(mongo_client=mongo_client, database="interns_b2_23", collection="Vipul-inventory")
 
     def insert_item(self, data):
         self.insert_one(data)
@@ -18,5 +14,5 @@ class InventoryCollection(MongoCollectionBaseClass):
     def delete_item(self, data):
         self.delete_one(data)
 
-    def update_item(self, data):
-        self.update_one(data)
+    def update_item(self, query, data):
+        self.update_one(query, data)

script/core/handlers/inventory.py

 # class for CRUD operations
+from script.core.db.mongo.interns2023 import mongo_client
 from script.core.db.mongo.interns2023.inventory import InventoryCollection
-from script.core.errors import NameDoesNotExist
 from script.core.schema.inventory import Inventory
 
 
 class InventoryData:
 
     def __init__(self):
-        self.inventory_col = InventoryCollection(mongo_client="")
+        self.inventory_col = InventoryCollection(mongo_client=mongo_client)
 
-    def delete_data(self, order_id, int):
+    def delete_data(self, order_id):
         try:
-            d = {"order_id": order_id}
-            if 3 not in d.values():
-                raise NameDoesNotExist
+            self.inventory_col.delete_one(query={"order_id": order_id})
         except ValueError:
             raise ValueError("ID")
         except Exception as e:
-            print(e, "Error Detected in Deleion")
+            print(e, "Error Detected in Deletion")
 
-    @staticmethod
-    def update_data(self, order_id: int):
+    def update_data(self, request_data: Inventory):
         try:
-            d = {"order_id": order_id}
-            if 2 not in d.values():
-                raise NameDoesNotExist
+            d = {"order_id": request_data.order_id,
+                 "customer_name": request_data.customer_name,
+                 "status": request_data.status,
+                 "sales_order": request_data.sales_order}
+            self.inventory_col.update_one(query={"order_id": request_data.order_id}, data=d)
         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
+            data = self.inventory_col.find_one(query={"order_id": order_id})
+            return data
         except Exception as e:
             print(e, "Error Detected in Finding")
 
     def insert_data(self, request_data: Inventory):
         try:
             d = {"order_id": request_data.order_id,
-                 "data": request_data.date,
                  "customer_name": request_data.customer_name,
                  "status": request_data.status,
                  "sales_order": request_data.sales_order}
-            self.inventory_col.insert_one(data)
-            if 5 not in d.values():
-                raise NameDoesNotExist
+            self.inventory_col.insert_one(d)
         except Exception as e:
             print(e, "Error Detected in inserting")

script/core/schema/inventory.py

+import datetime
 from typing import Optional
 
 from pydantic import BaseModel
@@ -5,8 +6,8 @@ from pydantic import BaseModel
 
 # define a model for Item
 class Inventory(BaseModel):
-    order_id: Optional[int]
-    date: int
+    order_id: int
     customer_name: str
-    status: str
-    sales_order: int
+    status: Optional[str]
+    date: Optional[datetime.date]
+    sales_order: Optional[int]

script/service/inventory.py

@@ -15,7 +15,7 @@ handler = InventoryData()
 async def delete_item(order_id: int):
     """This API deletes an item"""
     try:
-        handler.delete_data(order_id, int)
+        handler.delete_data(order_id)
         return DefaultResponse(message="Successfully deleted", status="success")
     except ValueError:
         return DefaultResponse(message="Due to Value Error")
@@ -25,9 +25,9 @@ async def delete_item(order_id: int):
 
 
 @router.post(APIEndpoints.update)
-async def update_item(order_id: int):
+async def update_item(request_data: Inventory):
     try:
-        handler.update_data(order_id, int)
+        handler.update_data(request_data)
         return DefaultResponse(message="Successfully Updated", status="success")
     except ValueError:
         return DefaultResponse(message="Due to value error")
@@ -39,8 +39,8 @@ async def update_item(order_id: int):
 @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")
+        data = handler.find_data(order_id)
+        return DefaultResponse(message="Successfully Found", status="success", data=data)
     except ValueError:
         return DefaultResponse(message="Due to value error")
     except Exception as e:

script/utils/mongo_connector.py

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

script/utils/mongo_utils.py

+import logging
 from typing import Dict, Optional
 
 from pymongo import MongoClient
 
-import logging
-
 
 class MongoConnect:
     def __init__(self, uri):
@@ -14,6 +13,11 @@ class MongoConnect:
             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):
@@ -21,6 +25,9 @@ class MongoCollectionBaseClass:
         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:
@@ -34,16 +41,12 @@ class MongoCollectionBaseClass:
             logging.error(f"Error in inserting the data {str(e)}")
             raise e
 
-    def find_one(self, data: Dict, filter_dict: Optional[Dict] = None):
+    def find_one(self, query, 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
+            database_name = self.client[self.database]
+            collection_name = database_name[self.collection]
+            response = collection_name.find_one(query)
+            return response
         except Exception as e:
             logging.error(f"Failed to fetch {str(e)}")
             raise e