this is complete crud fastapi + mongodb

.idea/.gitignore

+# Default ignored files
+/shelf/
+/workspace.xml

.idea/inspectionProfiles/profiles_settings.xml

+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>
\ No newline at end of file

.idea/misc.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.10 (task4)" project-jdk-type="Python SDK" />
+</project>
\ No newline at end of file

.idea/modules.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/task4.iml" filepath="$PROJECT_DIR$/.idea/task4.iml" />
+    </modules>
+  </component>
+</project>
\ No newline at end of file

.idea/task4.iml

+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$">
+      <excludeFolder url="file://$MODULE_DIR$/venv" />
+    </content>
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>
\ No newline at end of file

.idea/vcs.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>
\ No newline at end of file

conf/database_mongo.conf

+CONN_URI=mongodb://localhost:27017
\ No newline at end of file

main.py

+import uvicorn
+
+if __name__ == "__main__":
+    uvicorn.run("scripts.core.services.app:app", reload=True)
\ No newline at end of file

scripts/config/student_model.py

+from typing import Optional
+
+from pydantic import BaseModel, EmailStr, Field
+
+
+class StudentSchema(BaseModel):
+    fullname: str = Field(...)
+    email: EmailStr = Field(...)
+    course_of_study: str = Field(...)
+    year: int = Field(..., gt=0, lt=9)
+    gpa: float = Field(..., le=4.0)
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "fullname": "John Doe",
+                "email": "jdoe@x.edu.ng",
+                "course_of_study": "Water resources engineering",
+                "year": 2,
+                "gpa": "3.0",
+            }
+        }
+
+
+class UpdateStudentModel(BaseModel):
+    fullname: Optional[str]
+    email: Optional[EmailStr]
+    course_of_study: Optional[str]
+    year: Optional[int]
+    gpa: Optional[float]
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "fullname": "John Doe",
+                "email": "jdoe@x.edu.ng",
+                "course_of_study": "Water resources and environmental engineering",
+                "year": 4,
+                "gpa": "4.0",
+            }
+        }
+
+
+def ResponseModel(data, message):
+    return {
+        "data": [data],
+        "code": 200,
+        "message": message,
+    }
+
+
+def ErrorResponseModel(error, code, message):
+    return {"error": error, "code": code, "message": message}

scripts/core/handlers/delete_data_from_db.py

+from bson import ObjectId
+
+from scripts.database_connection.mongo_connect import student_collection
+
+
+async def delete_student(id: str):
+    student = await student_collection.find_one({"_id": ObjectId(id)})
+    if student:
+        await student_collection.delete_one({"_id": ObjectId(id)})
+        return True
\ No newline at end of file

scripts/core/handlers/get_all_data_from_db.py

+# Retrieve all students present in the database
+from scripts.database_connection.mongo_connect import student_collection, student_helper
+
+
+async def retrieve_students():
+    students = []
+    async for student in student_collection.find():
+        students.append(student_helper(student))
+    return students
\ No newline at end of file

scripts/core/handlers/get_data_by_id_from_db.py

+# Retrieve a student with a matching ID
+from bson import ObjectId
+
+from scripts.database_connection.mongo_connect import student_collection, student_helper
+
+
+async def retrieve_student(id: str) -> dict:
+    student = await student_collection.find_one({"_id": ObjectId(id)})
+    if student:
+        return student_helper(student)
\ No newline at end of file

scripts/core/handlers/insert_data_into_db.py

+# Add a new student into to the database
+from scripts.database_connection.mongo_connect import student_collection, student_helper
+
+
+async def add_student(student_data: dict) -> dict:
+    student = await student_collection.insert_one(student_data)
+    new_student = await student_collection.find_one({"_id": student.inserted_id})
+    return student_helper(new_student)
\ No newline at end of file

scripts/core/handlers/update_data_from_db.py

+# Update a student with a matching ID
+from bson import ObjectId
+
+from scripts.database_connection.mongo_connect import student_collection
+
+
+async def update_student(id: str, data: dict):
+    # Return false if an empty request body is sent.
+    if len(data) < 1:
+        return False
+    student = await student_collection.find_one({"_id": ObjectId(id)})
+    if student:
+        updated_student = await student_collection.update_one(
+            {"_id": ObjectId(id)}, {"$set": data}
+        )
+        if updated_student:
+            return True
+        return False
\ No newline at end of file

scripts/core/services/app.py

+from fastapi import FastAPI, APIRouter, Body
+from fastapi.encoders import jsonable_encoder
+
+from scripts.core.handlers.get_data_by_id_from_db import retrieve_student
+from scripts.core.handlers.delete_data_from_db import delete_student
+from scripts.core.handlers.update_data_from_db import update_student
+from scripts.core.handlers.get_all_data_from_db import retrieve_students
+from scripts.core.handlers.insert_data_into_db import add_student
+
+from scripts.config.student_model import ErrorResponseModel, ResponseModel, StudentSchema, UpdateStudentModel
+
+app = FastAPI()
+router = APIRouter()
+
+
+@app.get("/", tags=["Root"])
+async def read_root():
+    return {"message": "Welcome to this  crud mongo db app!"}
+
+
+# below will route to different crud api
+
+# this is for post : adding new data
+@app.post("/", response_description="Student data added into the database")
+async def add_student_data(student: StudentSchema = Body(...)):
+    student = jsonable_encoder(student)
+    new_student = await add_student(student)
+    return ResponseModel(new_student, "Student added successfully.")
+
+
+# this is for getting student data
+@app.get("/", response_description="Students retrieved")
+async def get_students():
+    students = await retrieve_students()
+    if students:
+        return ResponseModel(students, "Students data retrieved successfully")
+    return ResponseModel(students, "Empty list returned")
+
+
+@app.get("/{id}", response_description="Student data retrieved")
+async def get_student_data(id):
+    student = await retrieve_student(id)
+    if student:
+        return ResponseModel(student, "Student data retrieved successfully")
+    return ErrorResponseModel("An error occurred.", 404, "Student doesn't exist.")
+
+
+# this is for put method
+# for updating student model
+@app.put("/{id}")
+async def update_student_data(id: str, req: UpdateStudentModel = Body(...)):
+    req = {k: v for k, v in req.dict().items() if v is not None}
+    updated_student = await update_student(id, req)
+    if updated_student:
+        return ResponseModel(
+            "Student with ID: {} name update is successful".format(id),
+            "Student name updated successfully",
+        )
+    return ErrorResponseModel(
+        "An error occurred",
+        404,
+        "There was an error updating the student data.",
+    )
+
+
+# this is for delete method
+@app.delete("/{id}", response_description="Student data deleted from the database")
+async def delete_student_data(id: str):
+    deleted_student = await delete_student(id)
+    if deleted_student:
+        return ResponseModel(
+            "Student with ID: {} removed".format(id), "Student deleted successfully"
+        )
+    return ErrorResponseModel(
+        "An error occurred", 404, "Student with id {0} doesn't exist".format(id)
+    )

scripts/database_connection/mongo_connect.py

+import motor.motor_asyncio
+
+
+MONGO_DETAILS = "mongodb://localhost:27017"
+
+client = motor.motor_asyncio.AsyncIOMotorClient(MONGO_DETAILS)
+
+database = client.students
+
+student_collection = database.get_collection("students_collection")
+
+
+# hellper function
+def student_helper(student) -> dict:
+    return {
+        "id": str(student["_id"]),
+        "fullname": student["fullname"],
+        "email": student["email"],
+        "course_of_study": student["course_of_study"],
+        "year": student["year"],
+        "GPA": student["gpa"],
+    }
\ No newline at end of file