Databases লাইব্রেরি ব্যবহার করে ডাটাবেস ম্যানেজমেন্ট

FastAPI তে ডাটাবেস ম্যানেজমেন্ট খুবই সহজ, বিশেষত Databases লাইব্রেরি ব্যবহার করে। Databases একটি asynchronous ডাটাবেস লাইব্রেরি, যা SQLAlchemy এবং Tortoise ORM এর মতো সিনক্রোনাস লাইব্রেরি থেকে দ্রুত এবং স্কেলেবল ডাটাবেস ইন্টিগ্রেশন প্রদান করে। এটি PostgreSQL, MySQL, SQLite সহ অন্যান্য ডাটাবেস সমর্থন করে।

এখানে, আমরা Databases লাইব্রেরি ব্যবহার করে FastAPI তে ডাটাবেস ম্যানেজমেন্টের মূল ধারণাগুলি দেখব।

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

প্রথমে databases এবং ডাটাবেসের জন্য উপযুক্ত ড্রাইভার ইনস্টল করতে হবে। উদাহরণস্বরূপ, PostgreSQL ব্যবহারের জন্য asyncpg ইনস্টল করতে হবে।

PostgreSQL এর জন্য:

pip install databases asyncpg

SQLite এর জন্য:

pip install databases

Step 2: Databases লাইব্রেরি কনফিগারেশন

Databases লাইব্রেরি ব্যবহার করার জন্য, আপনাকে ডাটাবেসের URL প্রদান করতে হবে এবং Database ক্লাসের মাধ্যমে ডাটাবেস কনফিগার করতে হবে।

উদাহরণ: PostgreSQL ডাটাবেস কনফিগারেশন

from fastapi import FastAPI
from databases import Database

# PostgreSQL ডাটাবেস URL
DATABASE_URL = "postgresql://user:password@localhost/mydatabase"

# FastAPI অ্যাপ তৈরি
app = FastAPI()

# Databases লাইব্রেরি দিয়ে ডাটাবেস কনফিগারেশন
database = Database(DATABASE_URL)

# ডাটাবেস কনেকশন ওপেন এবং ক্লোজ করার জন্য উপযুক্ত পদ্ধতি
@app.on_event("startup")
async def startup():
    # ডাটাবেস কনেক্ট
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    # ডাটাবেস ডিসকানেক্ট
    await database.disconnect()

এখানে:

• Database(DATABASE_URL): ডাটাবেসের URL ব্যবহার করে ডাটাবেস সংযোগ তৈরি করা হয়েছে।
• await database.connect(): অ্যাপ স্টার্ট হলে ডাটাবেস সংযোগ করা হবে।
• await database.disconnect(): অ্যাপ শাটডাউন হলে ডাটাবেস সংযোগ বন্ধ করা হবে।

Step 3: ডাটাবেস টেবিল তৈরি করা

এখন, SQLAlchemy বা অন্যান্য ORM লাইব্রেরি ব্যবহার না করে আমরা সরাসরি SQL কমান্ড ব্যবহার করে ডাটাবেস টেবিল তৈরি করব।

উদাহরণ: টেবিল তৈরি করা

from databases import Database

# SQLAlchemy স্টাইলের মডেল বা টেবিল
CREATE_TABLE_QUERY = """
CREATE TABLE IF NOT EXISTS items (
    id SERIAL PRIMARY KEY,
    name TEXT,
    description TEXT,
    price NUMERIC
);
"""

@app.on_event("startup")
async def startup():
    # টেবিল তৈরি করা
    await database.connect()
    await database.execute(CREATE_TABLE_QUERY)

এখানে, CREATE_TABLE_QUERY স্ট্রিং ব্যবহার করে SQL টেবিল তৈরি করা হয়েছে। execute() ফাংশনটি SQL কমান্ড চালানোর জন্য ব্যবহার করা হয়।

Step 4: ডাটা ইনসার্ট করা এবং রিট্রিভ করা

Databases লাইব্রেরি ব্যবহার করে আপনি ডাটাবেসে ডাটা ইনসার্ট করতে পারেন এবং সেখান থেকে ডাটা রিট্রিভ করতে পারেন।

উদাহরণ: ডাটা ইনসার্ট এবং রিট্রিভ

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from databases import Database

app = FastAPI()

# পিডেন্টিক মডেল
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float

# ইনসার্ট করার জন্য SQL কমান্ড
INSERT_ITEM_QUERY = """
INSERT INTO items(name, description, price) 
VALUES (:name, :description, :price) RETURNING id;
"""

# রিট্রিভ করার জন্য SQL কমান্ড
SELECT_ITEM_QUERY = "SELECT * FROM items WHERE id = :id;"

@app.post("/items/")
async def create_item(item: Item):
    # ডাটা ইনসার্ট
    values = {"name": item.name, "description": item.description, "price": item.price}
    last_record_id = await database.execute(INSERT_ITEM_QUERY, values)
    return {"id": last_record_id}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # ডাটা রিট্রিভ
    query = SELECT_ITEM_QUERY
    values = {"id": item_id}
    result = await database.fetch_one(query, values)
    if result is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return result

ব্যাখ্যা:

• create_item: Item মডেল থেকে ইনপুট গ্রহণ করে ডাটাবেসে নতুন রেকর্ড ইনসার্ট করা হয়।
• read_item: নির্দিষ্ট item_id দিয়ে ডাটাবেস থেকে রেকর্ড রিট্রিভ করা হয়।

Step 5: ডাটাবেস কোয়েরি অপটিমাইজেশন

Databases লাইব্রেরি async অপারেশন সমর্থন করে, যা অ্যাসিঙ্ক্রোনাস ডাটাবেস কোয়েরি প্রক্রিয়া দ্রুত এবং স্কেলেবল করে তোলে। উদাহরণস্বরূপ:

# অ্যাসিঙ্ক্রোনাস কোয়েরি ব্যবহার
SELECT_ALL_ITEMS_QUERY = "SELECT * FROM items;"

@app.get("/items/")
async def get_all_items():
    items = await database.fetch_all(SELECT_ALL_ITEMS_QUERY)
    return items

এখানে, fetch_all() ফাংশনটি একটি অ্যাসিঙ্ক্রোনাস ফাংশন, যা ডাটাবেস থেকে একাধিক রেকর্ড ফেরত দেয়।

Step 6: SQLAlchemy এর সাথে Databases লাইব্রেরি ব্যবহার

FastAPI তে SQLAlchemy ORM এর সাথে Databases লাইব্রেরি ব্যবহার করে আপনি async ডাটাবেস ম্যানেজমেন্ট আরও সহজ এবং দ্রুত করতে পারেন।

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# SQLAlchemy setup
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/mydatabase"
engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# SQLAlchemy Model
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    description = Column(String, index=True)
    price = Column(Integer)

# Dependency for session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Databases লাইব্রেরি ব্যবহার করে FastAPI তে ডাটাবেস ম্যানেজমেন্ট অত্যন্ত সহজ এবং কার্যকর। এটি asynchronous কোড চালায়, যা আপনাকে উচ্চ পারফরম্যান্স এবং স্কেলেবিলিটি প্রদান করে। আপনি SQLAlchemy বা অন্য ORM ব্যবহার না করেও Databases লাইব্রেরি দিয়ে সরাসরি SQL কোয়েরি ব্যবহার করে ডাটাবেস পরিচালনা করতে পারেন।