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    

-----------------------------------------------------------------------------------



Popular posts from this blog

MCSL 216 MCA NEW Practical ~ common questions suggestions

dev chaeatsheet

STRAPI