basic CRUD with FASTAPI + MONGODB + beanie
main.py
from typing import Union
from fastapi import FastAPI
from routes import crud
from database.connection import init_db
app = FastAPI()
@app.on_event("startup")
async def connect():
await init_db()
app.include_router(crud.router)
-----------------------------------------------
routes/crud.py
from fastapi import APIRouter, status
from database.models import Student
from database.schemas import Students, UpdateStudent
from bson.objectid import ObjectId
router = APIRouter()
@router.post("/student/", status_code=status.HTTP_200_OK, tags=['student'])
async def create_student(users: Students):
try:
stu_obj = Student(name=users.name, phone=users.phone, email=users.email, address=users.address)
await stu_obj.insert()
return {"status":"record inserted"}
except Exception as e:
print(e)
return {"status":"error occured"}
@router.get("/student/", status_code=status.HTTP_200_OK, tags=['student'])
async def read_students():
students = await Student.find_all().to_list()
return students
@router.put("/student/", status_code=status.HTTP_200_OK, tags=['student'])
async def update_student(userid, users: UpdateStudent):
objInstance = ObjectId(userid)
try:
stu_obj = await Student.find_one({"_id": objInstance})
if users.name:
await stu_obj.set({Student.name : users.name})
if users.address:
await stu_obj.set({Student.address : users.address})
if users.phone:
await stu_obj.set({Student.phone : users.phone})
if users.email:
await stu_obj.set({Student.email : users.email})
return {"status":"record updated"}
except Exception as e:
print(e)
return {"status":"user id not exist!"}
@router.delete("/student/", status_code=status.HTTP_200_OK, tags=['student'])
async def delete_user(userid : str):
objInstance = ObjectId(userid)
try:
stu_obj = await Student.find_one({"_id": objInstance})
await stu_obj.delete()
return {"status":"collection deleted"}
except Exception as e:
print(e)
return {"status":"something error occured"}
----------------------------------------------------------------------------------
database/connection.py
from motor.motor_asyncio import AsyncIOMotorClient
from beanie import init_beanie
from .models import Student
async def init_db():
client = AsyncIOMotorClient("mongodb+srv://iamshimantadas:9icJV6nfGTlFB0v5@cluster0.nyxj0fc.mongodb.net/")
await init_beanie(database=client.db_name, document_models=[Student])
------------------------------------------------------------------------------------
database/models.py
from beanie import Document
class Student(Document):
name : str
phone : int
address : str
email : str
------------------------------------------------------------------------------------
database/schemas,py
from pydantic import BaseModel
class Students(BaseModel):
name : str
phone : int
address : str
email : str
class UpdateStudent(BaseModel):
name : str | None = None
phone : int | None = None
address : str | None = None
email : str | None = None
-----------------------------------------------------------------------------------