MongoDB এর জন্য Pydantic Model ব্যবহার

FastAPI তে MongoDB ব্যবহার করার জন্য আমরা Pydantic Models এবং Motor (MongoDB-এর জন্য অ্যাসিঙ্ক্রোনাস ড্রাইভার) ব্যবহার করে সহজেই ডাটাবেসের সাথে ইন্টিগ্রেশন করতে পারি। এখানে আমরা দেখব কিভাবে FastAPI তে MongoDB সংযোগ স্থাপন করতে হয় এবং MongoDB ডকুমেন্টগুলি Pydantic Model এর মাধ্যমে কিভাবে পরিচালনা করা যায়।

Step 1: প্রয়োজনীয় লাইব্রেরি ইনস্টল করা

MongoDB এর সাথে কাজ করার জন্য আপনাকে Motor এবং Pydantic লাইব্রেরি ইনস্টল করতে হবে।

pip install motor pydantic fastapi

Motor MongoDB এর জন্য একটি অ্যাসিঙ্ক্রোনাস ড্রাইভার যা FastAPI-এর সাথে খুব ভালোভাবে কাজ করে।

Step 2: MongoDB সংযোগ স্থাপন করা

MongoDB সংযোগ স্থাপন করতে Motor ব্যবহার করা হয়। প্রথমে MongoDB এর URI (যেমন mongodb://localhost:27017) ব্যবহার করে ডাটাবেসের সাথে সংযোগ স্থাপন করতে হবে।

উদাহরণ: MongoDB সংযোগ কনফিগারেশন

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient
from pydantic import BaseModel
from typing import List

app = FastAPI()

# MongoDB URI এবং ডাটাবেস কনফিগারেশন
MONGO_DETAILS = "mongodb://localhost:27017"  # MongoDB URI
client = AsyncIOMotorClient(MONGO_DETAILS)
database = client.test_db  # MongoDB ডাটাবেস
collection = database.items  # MongoDB কালেকশন

# Pydantic Model
class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float | None = None

এখানে:

• Motor Client ব্যবহার করে MongoDB এর সাথে সংযোগ স্থাপন করা হয়েছে।
• test_db MongoDB ডাটাবেস এবং items MongoDB কালেকশন তৈরি করা হয়েছে।

Step 3: Pydantic Model এবং MongoDB Document Integration

FastAPI তে MongoDB ডেটা ইন্টিগ্রেট করার জন্য Pydantic Model ব্যবহার করা হয়, যার মাধ্যমে ডাটা যাচাই এবং ডাটা ট্রান্সফার করা হয়। MongoDB ডকুমেন্ট গুলি Pydantic Model এর মাধ্যমে ম্যানিপুলেট করা হয়।

উদাহরণ: Pydantic Model এবং MongoDB ডকুমেন্ট হ্যান্ডলিং

from motor.motor_asyncio import AsyncIOMotorClient
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

MONGO_DETAILS = "mongodb://localhost:27017"
client = AsyncIOMotorClient(MONGO_DETAILS)
database = client.test_db
collection = database.items

# Pydantic Model
class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float | None = None

# MongoDB Document হিসেবে Mongodb এর ডাটা পাওয়া
def item_helper(item) -> dict:
    return {
        "name": item["name"],
        "description": item["description"],
        "price": item["price"],
        "tax": item["tax"],
    }

# একটি নতুন item ডকুমেন্ট MongoDB তে ইনসার্ট করা
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    item_data = item.dict()  # Pydantic Model থেকে dict কনভার্ট
    result = await collection.insert_one(item_data)  # MongoDB তে ডাটা ইনসার্ট
    created_item = await collection.find_one({"_id": result.inserted_id})
    return item_helper(created_item)  # MongoDB ডকুমেন্টকে Pydantic Model হিসেবে ফেরত দেয়া

এখানে:

• item_helper() ফাংশনটি MongoDB ডকুমেন্টকে Pydantic Model ডেটাতে কনভার্ট করে।
• create_item() রাউটটি একটি নতুন item MongoDB ডাটাবেসে ইনসার্ট করে এবং সেই ডকুমেন্টটি Pydantic Model হিসেবে ফেরত দেয়।

Step 4: MongoDB থেকে ডাটা রিট্রিভ করা

MongoDB থেকে ডাটা রিট্রিভ করতে এবং সেই ডাটাকে Pydantic Model এর সাথে কনভার্ট করার জন্য find() মেথড ব্যবহার করা হয়।

উদাহরণ: MongoDB থেকে ডাটা রিট্রিভ করা

@app.get("/items/", response_model=List[Item])
async def get_items():
    items = []
    async for item in collection.find():  # MongoDB থেকে ডাটা রিট্রিভ
        items.append(item_helper(item))  # MongoDB ডকুমেন্টকে Pydantic Model হিসেবে কনভার্ট
    return items

এখানে:

• find() ব্যবহার করে MongoDB থেকে সব ডকুমেন্ট রিট্রিভ করা হয়েছে।
• item_helper() ফাংশনটি MongoDB ডকুমেন্টকে Pydantic Model এর সাথে কনভার্ট করে।

রিকোয়েস্ট উদাহরণ:

GET /items/

রেসপন্স:

[
  {
    "name": "Laptop",
    "description": "A high-performance laptop",
    "price": 1500.00,
    "tax": 120.00
  },
  {
    "name": "Phone",
    "description": "A modern smartphone",
    "price": 799.99,
    "tax": 60.00
  }
]

Step 5: MongoDB ডকুমেন্ট আপডেট করা

MongoDB ডকুমেন্ট আপডেট করার জন্য update_one() বা update_many() মেথড ব্যবহার করা হয়।

উদাহরণ: MongoDB ডকুমেন্ট আপডেট করা

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    updated_item = await collection.update_one(
        {"_id": item_id}, {"$set": item.dict()}  # MongoDB ডকুমেন্ট আপডেট
    )
    if updated_item.matched_count == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

এখানে:

• update_one() ব্যবহার করে MongoDB ডকুমেন্ট আপডেট করা হয়েছে।

Step 6: MongoDB ডকুমেন্ট মুছে ফেলা

MongoDB ডকুমেন্ট মুছে ফেলার জন্য delete_one() মেথড ব্যবহার করা হয়।

উদাহরণ: MongoDB ডকুমেন্ট মুছে ফেলা

@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
    deleted_item = await collection.delete_one({"_id": item_id})  # MongoDB ডকুমেন্ট মুছে ফেলা
    if deleted_item.deleted_count == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"message": "Item deleted successfully"}

এখানে:

• delete_one() ব্যবহার করে MongoDB ডকুমেন্ট মুছে ফেলা হয়েছে।

FastAPI তে Pydantic Models ব্যবহার করে MongoDB এর ডকুমেন্ট হ্যান্ডল করা খুবই সহজ এবং কার্যকর। MongoDB-এ ডাটা সংরক্ষণ, রিট্রিভ, আপডেট এবং মুছে ফেলার জন্য FastAPI ও Pydantic মডেল ব্যবহার করা যেতে পারে। Motor লাইব্রেরি MongoDB-এর অ্যাসিঙ্ক্রোনাস ড্রাইভার হিসেবে ব্যবহৃত হয়, যা FastAPI এর সাথে খুব ভালোভাবে কাজ করে।