FastAPI এর সঙ্গে Database Integration

FastAPI তে Database Integration খুবই সহজ এবং কার্যকর। এখানে, আমরা SQLAlchemy, Tortoise ORM, এবং Databases প্যাকেজ ব্যবহার করে FastAPI-এর সঙ্গে ডাটাবেস সংযোগ করতে পারি। এখানে মূলত SQLAlchemy এবং SQLite ডাটাবেসের সঙ্গে কীভাবে কাজ করা যায়, তা দেখানো হবে।

SQLAlchemy একটি জনপ্রিয় ORM (Object-Relational Mapping) লাইব্রেরি, যা Python ডেভেলপারদের জন্য SQL ডাটাবেসের সঙ্গে কাজ করার জন্য সহজ এবং শক্তিশালী একটি উপায়।

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

FastAPI-এর সঙ্গে ডাটাবেস ইন্টিগ্রেশন করতে SQLAlchemy এবং Databases প্যাকেজ ইনস্টল করা হয়।

pip install sqlalchemy databases sqlite

এছাড়া, আপনি Pydantic (যা FastAPI ডাটা ভ্যালিডেশনের জন্য ব্যবহার করে) এবং uvicorn (FastAPI অ্যাপ চালানোর জন্য) ইনস্টল করতে পারেন।

pip install pydantic uvicorn

Step 2: ডাটাবেস কনফিগারেশন

প্রথমে SQLAlchemy দিয়ে ডাটাবেস সংযোগ কনফিগার করতে হবে।

Example: SQLite ডাটাবেস কনফিগারেশন

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

# ডাটাবেস কনফিগারেশন
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

# SQLAlchemy ডাটাবেস ইঞ্জিন তৈরি করা
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base ক্লাস তৈরি
Base = declarative_base()

# ডাটাবেস মডেল তৈরি
class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String)
    price = Column(Integer)

এখানে, Item হলো একটি SQLAlchemy মডেল, যা ডাটাবেসে একটি টেবিল হিসেবে ব্যবহৃত হবে। SessionLocal একটি ডাটাবেস সেশন তৈরি করতে ব্যবহৃত হয়।

Step 3: ডাটাবেস সেশন তৈরি এবং টেবিল তৈরি করা

ডাটাবেস সেশন তৈরি করতে এবং টেবিলগুলো তৈরি করতে create_all ব্যবহার করা হয়।

# ডাটাবেস সেশন তৈরি
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# ডাটাবেস টেবিল তৈরি
Base.metadata.create_all(bind=engine)

এখানে, get_db() ফাংশনটি ডাটাবেস সেশন প্রদান করে, যা FastAPI রাউটগুলোর মধ্যে Dependency Injection হিসেবে ব্যবহৃত হবে। create_all() ফাংশনটি টেবিল তৈরি করবে যদি সেগুলি ইতিমধ্যে না থাকে।

Step 4: FastAPI রাউট তৈরি করা

এখন FastAPI এর রাউট তৈরি করা যাক, যাতে ডাটাবেসের মধ্যে ডাটা যোগ করা, পড়া, আপডেট করা এবং মুছে ফেলা যায়।

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from pydantic import BaseModel

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

# Pydantic মডেল তৈরি (Response Model)
class ItemCreate(BaseModel):
    name: str
    description: str
    price: int

class ItemResponse(BaseModel):
    id: int
    name: str
    description: str
    price: int

    class Config:
        orm_mode = True

# Item তৈরি (POST রিকোয়েস্ট)
@app.post("/items/", response_model=ItemResponse)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(name=item.name, description=item.description, price=item.price)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

# Item পড়া (GET রিকোয়েস্ট)
@app.get("/items/{item_id}", response_model=ItemResponse)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    return db_item

# সকল Item দেখানো (GET রিকোয়েস্ট)
@app.get("/items/", response_model=list[ItemResponse])
def read_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    items = db.query(Item).offset(skip).limit(limit).all()
    return items

এখানে, create_item ফাংশনটি POST রিকোয়েস্টের মাধ্যমে ডাটাবেসে নতুন আইটেম যোগ করে এবং read_item ফাংশনটি GET রিকোয়েস্টের মাধ্যমে একটি নির্দিষ্ট আইটেম পড়ে।

Step 5: Uvicorn দিয়ে FastAPI অ্যাপ চালানো

অ্যাপটি চালানোর জন্য Uvicorn ব্যবহার করা হয়।

uvicorn main:app --reload

এখানে, main হল আপনার Python ফাইলের নাম এবং app হল FastAPI অ্যাপের অবজেক্ট।

Step 6: টেস্ট রিকোয়েস্ট

POST রিকোয়েস্ট: নতুন Item তৈরি করা

URL: /items/
POST Body:

{
  "name": "Laptop",
  "description": "A high-performance laptop",
  "price": 1500
}

GET রিকোয়েস্ট: Item পড়া

URL: /items/1

GET রিকোয়েস্ট: সকল Item পড়া

URL: /items/?skip=0&limit=10

Step 7: Dependency Injection এবং Database Session

FastAPI তে Dependency Injection ব্যবহার করে ডাটাবেস সেশন ইনজেক্ট করা হয়েছে। get_db ফাংশনটি ডাটাবেস সেশন তৈরি করে এবং রাউটে Depends(get_db) দিয়ে সেশনটি ইনজেক্ট করা হয়।

from fastapi import Depends
from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

FastAPI তে Database Integration খুবই সহজ এবং কার্যকর। আপনি SQLAlchemy বা অন্যান্য ORM ব্যবহার করে ডাটাবেসের সঙ্গে সংযোগ স্থাপন করতে পারেন এবং FastAPI এর মাধ্যমে বিভিন্ন রাউট তৈরি করে ডাটাবেসে ডাটা পরিচালনা করতে পারেন। FastAPI এর Dependency Injection এর মাধ্যমে ডাটাবেস সেশনগুলি খুব সহজেই রাউটগুলোর মধ্যে ইনজেক্ট করা যায়।

FastAPI এবং SQLAlchemy একসাথে ব্যবহার করে আপনি একটি পাইথন-বেসড ওয়েব অ্যাপ্লিকেশন তৈরি করতে পারেন, যেখানে SQLAlchemy ডাটাবেস ইন্টারঅ্যাকশনের জন্য ব্যবহৃত হয় এবং FastAPI HTTP এন্ডপয়েন্ট তৈরি এবং পরিচালনা করে। SQLAlchemy একটি শক্তিশালী ORM (Object-Relational Mapping) টুল যা SQL ডাটাবেসের সাথে কাজ করতে সহায়ক, এবং FastAPI তে এটি সহজে ইন্টিগ্রেট করা যায়।

এখানে FastAPI এবং SQLAlchemy এর ইন্টিগ্রেশন কিভাবে কাজ করে এবং কীভাবে ডাটাবেস পরিচালনা করা যায় তা বিস্তারিতভাবে আলোচনা করা হলো।

Step 1: SQLAlchemy ইনস্টলেশন

প্রথমে আপনাকে SQLAlchemy এবং Databases (অ্যাসিঙ্ক্রোনাস ডাটাবেস সাপোর্টের জন্য) ইনস্টল করতে হবে:

pip install sqlalchemy databases

এছাড়া, আপনি যদি SQLite ডাটাবেস ব্যবহার করতে চান, তবে এই কমান্ডটি ব্যবহার করুন:

pip install sqlite

Step 2: FastAPI এবং SQLAlchemy সেটআপ

এখানে একটি সিম্পল FastAPI অ্যাপ্লিকেশন তৈরি করা হয়েছে যেখানে SQLAlchemy ব্যবহৃত হচ্ছে।

SQLAlchemy ডাটাবেস কনফিগারেশন:

SQLAlchemy ব্যবহারের জন্য প্রথমে ডাটাবেস কনফিগারেশন তৈরি করতে হবে। এই কনফিগারেশনের মধ্যে ডাটাবেস ইঞ্জিন এবং সেশন তৈরি করা হয়।

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

# ডাটাবেস কনফিগারেশন
DATABASE_URL = "sqlite:///./test.db"

# ডাটাবেস ইঞ্জিন তৈরি
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

# সেশন তৈরি
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# ডাটাবেস মডেল তৈরি করার জন্য বেস ক্লাস
Base = declarative_base()

এখানে:

• DATABASE_URL: ডাটাবেস সংযোগের URL।
• engine: SQLAlchemy ইঞ্জিন, যা ডাটাবেসের সাথে সংযোগ স্থাপন করবে।
• SessionLocal: সেশন ম্যানেজমেন্টের জন্য SQLAlchemy-র সেশন ফ্যাক্টরি।
• Base: SQLAlchemy মডেল তৈরি করতে ব্যবহৃত বেস ক্লাস।

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

এখন একটি ডাটাবেস মডেল তৈরি করা যাক। উদাহরণস্বরূপ, একটি User মডেল তৈরি করা যাক:

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

এখানে:

• User ক্লাসটি SQLAlchemy মডেল, যেখানে তিনটি কলাম (id, name, email) ডিফাইন করা হয়েছে।
• index=True: ফিল্ডগুলির জন্য ইনডেক্স তৈরি করবে।
• unique=True: email ফিল্ডটি ইউনিক হতে হবে, অর্থাৎ, একাধিক ব্যবহারকারী একই ইমেইল ব্যবহার করতে পারবেন না।

Step 3: FastAPI এন্ডপয়েন্ট তৈরি

এখন FastAPI অ্যাপ তৈরি করে ডাটাবেসের সাথে ইন্টিগ্রেট করা যাক।

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel
from typing import List

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

# Pydantic Model for Request and Response
class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        orm_mode = True

# Dependency to get the database session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Create a new user
@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# Get all users
@app.get("/users/", response_model=List[UserResponse])
def get_users(db: Session = Depends(get_db)):
    users = db.query(User).all()
    return users

এখানে:

• UserCreate: Pydantic মডেল যা ইনপুট হিসেবে name এবং email নেয়।
• UserResponse: Pydantic মডেল যা রেসপন্স হিসেবে id, name, এবং email প্রদান করে। এটি orm_mode = True এর মাধ্যমে SQLAlchemy মডেল থেকে রেসপন্স তৈরি করে।
• get_db: ডাটাবেস সেশন রিটার্ন করার জন্য একটি Dependency ফাংশন।
• create_user: নতুন ব্যবহারকারী তৈরি করার জন্য POST এন্ডপয়েন্ট।
• get_users: সব ব্যবহারকারীদের ডাটাবেস থেকে আনা এবং রেসপন্স করা GET এন্ডপয়েন্ট।

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

এখন, SQLAlchemy মডেল অনুযায়ী ডাটাবেসে টেবিল তৈরি করা দরকার।

Base.metadata.create_all(bind=engine)

এই কোডটি ডাটাবেসে users টেবিল তৈরি করবে, যদি তা ইতিমধ্যে না থাকে।

Step 5: অ্যাপ চালানো

এখন, FastAPI অ্যাপ চালাতে নিচের কমান্ড ব্যবহার করুন:

uvicorn main:app --reload

এখানে, main হল আপনার Python ফাইলের নাম (যেমন main.py), এবং app হল FastAPI অ্যাপের অবজেক্ট।

Step 6: টেস্টিং

আপনি ব্রাউজারে Swagger UI-এ গিয়ে আপনার API টেস্ট করতে পারেন:
http://127.0.0.1:8000/docs

এখানে:

• POST /users/: নতুন ব্যবহারকারী তৈরি করতে ব্যবহার করুন।
• GET /users/: সব ব্যবহারকারীর তালিকা দেখতে ব্যবহার করুন।

FastAPI এবং SQLAlchemy একত্রে ব্যবহার করে আপনি খুব সহজে একটি অ্যাপ্লিকেশন তৈরি করতে পারেন যা ডাটাবেস ইন্টারঅ্যাকশন, ভ্যালিডেশন এবং API এন্ডপয়েন্টগুলোকে দ্রুত পরিচালনা করতে সহায়ক। SQLAlchemy-র ORM মডেল ব্যবহার করে ডাটাবেসের সাথে সংযোগ স্থাপন এবং FastAPI-র সাহায্যে API তৈরি করা অত্যন্ত সুবিধাজনক ও কার্যকর।

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 কোয়েরি ব্যবহার করে ডাটাবেস পরিচালনা করতে পারেন।

FastAPI ব্যবহার করে CRUD (Create, Read, Update, Delete) অপারেশন তৈরি করা সহজ এবং দ্রুত। FastAPI এর সাহায্যে আমরা API তৈরি করতে পারি যা ডাটা তৈরি, পড়া, আপডেট এবং মুছে ফেলার কাজগুলো সঞ্চালন করতে পারে। এখানে, আমরা দেখব কিভাবে FastAPI তে CRUD অপারেশন তৈরি করা যায়।

Step 1: ফাইল স্ট্রাকচার তৈরি করা

প্রথমে, একটি প্রাথমিক ফোল্ডার স্ট্রাকচার তৈরি করা যাক। উদাহরণস্বরূপ:

my_fastapi_project/
├── app/
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   ├── crud.py
├── requirements.txt

Step 2: Pydantic Models তৈরি করা (Schemas)

FastAPI তে CRUD অপারেশন করার জন্য প্রথমে আমাদের Pydantic models (schemas) তৈরি করতে হবে, যা ডাটা ভ্যালিডেশন এবং টাইপ এনোটেশন নিশ্চিত করে।

উদাহরণ: schemas.py ফাইল

from pydantic import BaseModel
from typing import Optional

class ItemBase(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

class ItemCreate(ItemBase):
    pass

class ItemUpdate(ItemBase):
    pass

class Item(ItemBase):
    id: int

    class Config:
        orm_mode = True

এখানে:

• ItemBase: একটি বেস মডেল যা সাধারণ ফিল্ডসমূহ ধারণ করে।
• ItemCreate: নতুন আইটেম তৈরি করার জন্য ব্যবহৃত মডেল।
• ItemUpdate: আইটেম আপডেট করার জন্য ব্যবহৃত মডেল।
• Item: ডাটা রিটার্ন করার জন্য ব্যবহৃত মডেল, এতে id ফিল্ডও অন্তর্ভুক্ত করা হয়েছে।

Step 3: Models তৈরি করা (Database Models)

এখন, ডাটাবেস মডেল তৈরি করা হবে। উদাহরণস্বরূপ, আমরা SQLAlchemy ব্যবহার করে একটি ডাটাবেস মডেল তৈরি করব।

উদাহরণ: models.py ফাইল

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True, nullable=True)
    price = Column(Float)
    tax = Column(Float, nullable=True)

এখানে, Item একটি SQLAlchemy মডেল যা items নামের টেবিলকে প্রতিনিধিত্ব করে। এতে id, name, description, price এবং tax ফিল্ড রয়েছে।

Step 4: CRUD অপারেশন তৈরি করা

এখন আমরা CRUD অপারেশন তৈরি করব, যেখানে Create, Read, Update এবং Delete অপারেশন সংজ্ঞায়িত করা হবে।

উদাহরণ: crud.py ফাইল

from sqlalchemy.orm import Session
from . import models, schemas

# Create
def create_item(db: Session, item: schemas.ItemCreate):
    db_item = models.Item(name=item.name, description=item.description, price=item.price, tax=item.tax)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

# Read
def get_item(db: Session, item_id: int):
    return db.query(models.Item).filter(models.Item.id == item_id).first()

def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()

# Update
def update_item(db: Session, item_id: int, item: schemas.ItemUpdate):
    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if db_item:
        db_item.name = item.name
        db_item.description = item.description
        db_item.price = item.price
        db_item.tax = item.tax
        db.commit()
        db.refresh(db_item)
    return db_item

# Delete
def delete_item(db: Session, item_id: int):
    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if db_item:
        db.delete(db_item)
        db.commit()
    return db_item

এখানে:

• create_item: নতুন আইটেম তৈরি করে এবং ডাটাবেসে সেভ করে।
• get_item: একটি নির্দিষ্ট আইটেম আইডি দিয়ে আইটেমটি খুঁজে বের করে।
• get_items: সমস্ত আইটেমের একটি তালিকা রিটার্ন করে।
• update_item: একটি নির্দিষ্ট আইটেম আপডেট করে।
• delete_item: একটি নির্দিষ্ট আইটেম মুছে ফেলে।

Step 5: FastAPI এ রাউটিং (API Endpoints)

এখন আমরা FastAPI এ CRUD অপারেশনগুলোর জন্য রাউট তৈরি করব। প্রতিটি অপারেশনের জন্য আলাদা এন্ডপয়েন্ট তৈরি করা হবে।

উদাহরণ: main.py ফাইল

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from . import crud, models, schemas

# Database setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# FastAPI instance
app = FastAPI()

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

# Create item
@app.post("/items/", response_model=schemas.Item)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    return crud.create_item(db=db, item=item)

# Read item by ID
@app.get("/items/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = crud.get_item(db=db, item_id=item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

# Read all items
@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db=db, skip=skip, limit=limit)
    return items

# Update item
@app.put("/items/{item_id}", response_model=schemas.Item)
def update_item(item_id: int, item: schemas.ItemUpdate, db: Session = Depends(get_db)):
    db_item = crud.update_item(db=db, item_id=item_id, item=item)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

# Delete item
@app.delete("/items/{item_id}", response_model=schemas.Item)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    db_item = crud.delete_item(db=db, item_id=item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

এখানে:

• /items/: নতুন আইটেম তৈরি করার জন্য।
• /items/{item_id}: নির্দিষ্ট আইটেমের ডিটেইল দেখতে।
• /items/: সমস্ত আইটেমের তালিকা দেখতে।
• /items/{item_id}: নির্দিষ্ট আইটেম আপডেট বা ডিলিট করতে।

Step 6: Requirements.txt

প্রজেক্টে ব্যবহৃত লাইব্রেরিগুলির তালিকা তৈরি করতে requirements.txt ফাইল তৈরি করা হয়।

fastapi
uvicorn
sqlalchemy
pydantic

Step 7: সার্ভার চালানো

FastAPI অ্যাপ্লিকেশন চালাতে uvicorn ব্যবহার করা হয়।

uvicorn app.main:app --reload

FastAPI ব্যবহার করে CRUD (Create, Read, Update, Delete) অপারেশন তৈরি করা অত্যন্ত সহজ এবং দ্রুত। আমরা এখানে SQLAlchemy ব্যবহার করেছি ডাটাবেস মডেল তৈরি করার জন্য এবং Pydantic মডেল ব্যবহার করেছি ডাটা ভ্যালিডেশন এবং টাইপ এনোটেশন নিশ্চিত করার জন্য। এইভাবে FastAPI তে CRUD অপারেশন করতে আপনি খুবই সহজে একটি পূর্ণাঙ্গ API তৈরি করতে পারেন।

FastAPI তে PostgreSQL, MySQL, বা SQLite এর মতো ডাটাবেস ব্যবহার করা খুবই সহজ এবং কার্যকর। FastAPI SQLAlchemy এবং Tortoise ORM এর মতো লাইব্রেরির সাথে ইন্টিগ্রেশন সমর্থন করে, যা ডাটাবেসের সাথে সহজে কাজ করার সুযোগ দেয়। এখানে আমরা দেখব কিভাবে FastAPI-র সঙ্গে বিভিন্ন ধরনের ডাটাবেস (PostgreSQL, MySQL, SQLite) ব্যবহার করা যায়।

১. PostgreSQL ব্যবহার FastAPI-তে

Step 1: PostgreSQL সেটআপ

প্রথমে PostgreSQL ডাটাবেস ইনস্টল এবং কনফিগার করুন, এবং ডাটাবেস তৈরি করুন।

sudo apt-get install postgresql postgresql-contrib

ডাটাবেস তৈরি করুন:

sudo -u postgres psql
CREATE DATABASE fastapi_db;
CREATE USER fastapi_user WITH PASSWORD 'yourpassword';
ALTER ROLE fastapi_user SET client_encoding TO 'utf8';
ALTER ROLE fastapi_user SET default_transaction_isolation TO 'read committed';
ALTER ROLE fastapi_user SET timezone TO 'UTC';
GRANT ALL PRIVILEGES ON DATABASE fastapi_db TO fastapi_user;

Step 2: SQLAlchemy এবং Psycopg2 ইনস্টলেশন

pip install sqlalchemy psycopg2

• sqlalchemy: ডাটাবেস ORM হিসেবে ব্যবহৃত হয়।
• psycopg2: PostgreSQL এর সাথে সংযোগ স্থাপনের জন্য প্রয়োজনীয় ড্রাইভার।

Step 3: FastAPI অ্যাপ কনফিগারেশন

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

app = FastAPI()

# PostgreSQL ডাটাবেস URL
SQLALCHEMY_DATABASE_URL = "postgresql://fastapi_user:yourpassword@localhost/fastapi_db"

# SQLAlchemy engine এবং session সেটআপ
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# মডেল তৈরি
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    price = Column(Integer)

# ডাটাবেস তৈরি
Base.metadata.create_all(bind=engine)

@app.post("/items/")
def create_item(name: str, price: int):
    db = SessionLocal()
    db_item = Item(name=name, price=price)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    db.close()
    return db_item

এখানে:

• SQLALCHEMY_DATABASE_URL: PostgreSQL ডাটাবেসের URL।
• SessionLocal: ডাটাবেস সেশন তৈরি করতে ব্যবহৃত।
• Base: SQLAlchemy মডেলকে বেস ক্লাস হিসাবে ডিফাইন করা হয়েছে।

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

POST /items/

{
  "name": "Laptop",
  "price": 1500
}

রেসপন্স:

{
  "id": 1,
  "name": "Laptop",
  "price": 1500
}

২. MySQL ব্যবহার FastAPI-তে

Step 1: MySQL সেটআপ

প্রথমে MySQL ইনস্টল এবং কনফিগার করুন, এবং ডাটাবেস তৈরি করুন।

sudo apt-get install mysql-server
sudo mysql_secure_installation

ডাটাবেস তৈরি করুন:

mysql -u root -p
CREATE DATABASE fastapi_db;
CREATE USER 'fastapi_user'@'localhost' IDENTIFIED BY 'yourpassword';
GRANT ALL PRIVILEGES ON fastapi_db.* TO 'fastapi_user'@'localhost';
FLUSH PRIVILEGES;

Step 2: SQLAlchemy এবং MySQL-connector ইনস্টলেশন

pip install sqlalchemy mysql-connector-python

• mysql-connector-python: MySQL ডাটাবেসের সাথে সংযোগ স্থাপনের জন্য ব্যবহৃত ড্রাইভার।

Step 3: FastAPI অ্যাপ কনফিগারেশন

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

app = FastAPI()

# MySQL ডাটাবেস URL
SQLALCHEMY_DATABASE_URL = "mysql+mysqlconnector://fastapi_user:yourpassword@localhost/fastapi_db"

# SQLAlchemy engine এবং session সেটআপ
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# মডেল তৈরি
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    price = Column(Integer)

# ডাটাবেস তৈরি
Base.metadata.create_all(bind=engine)

@app.post("/items/")
def create_item(name: str, price: int):
    db = SessionLocal()
    db_item = Item(name=name, price=price)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    db.close()
    return db_item

এখানে, mysql+mysqlconnector MySQL ডাটাবেসের সাথে সংযোগ স্থাপনের জন্য ব্যবহৃত হচ্ছে।

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

POST /items/

{
  "name": "Smartphone",
  "price": 800
}

রেসপন্স:

{
  "id": 1,
  "name": "Smartphone",
  "price": 800
}

৩. SQLite ব্যবহার FastAPI-তে

Step 1: SQLite সেটআপ

SQLite ইনস্টল করার জন্য কোনো আলাদা সেটআপের প্রয়োজন নেই, কারণ এটি একটি ফাইল-ভিত্তিক ডাটাবেস, যা Python-এর সঙ্গে একযোগে আসে।

Step 2: SQLAlchemy এবং SQLite ব্যবহার

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

app = FastAPI()

# SQLite ডাটাবেস URL
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"  # `./test.db` ফাইল ব্যবহার করা হবে

# SQLAlchemy engine এবং session সেটআপ
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# মডেল তৈরি
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    price = Column(Integer)

# ডাটাবেস তৈরি
Base.metadata.create_all(bind=engine)

@app.post("/items/")
def create_item(name: str, price: int):
    db = SessionLocal()
    db_item = Item(name=name, price=price)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    db.close()
    return db_item

এখানে, sqlite:///./test.db ব্যবহার করে একটি SQLite ডাটাবেস তৈরি করা হয়েছে। এটি ফাইল-ভিত্তিক ডাটাবেস, তাই কোনো সার্ভার সেটআপের প্রয়োজন নেই।

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

POST /items/

{
  "name": "Tablet",
  "price": 600
}

রেসপন্স:

{
  "id": 1,
  "name": "Tablet",
  "price": 600
}

FastAPI তে PostgreSQL, MySQL, এবং SQLite ডাটাবেস ব্যবহারের মাধ্যমে আপনি স্কেলেবল এবং শক্তিশালী API তৈরি করতে পারেন। SQLAlchemy ব্যবহার করে ডাটাবেস মডেল, সংযোগ এবং সেশন পরিচালনা করা সহজ হয়। FastAPI এর সাথে এই ডাটাবেস ইন্টিগ্রেশন অ্যাসিঙ্ক্রোনাস অপারেশন এবং দ্রুত পারফরম্যান্স নিশ্চিত করতে সাহায্য করে।