কিভাবে FastAPI দিয়ে একটি REST API তৈরি করবেন

FastAPI নতুন Python API-এর জন্য ডিফল্ট পছন্দ হয়ে উঠেছে, এবং সঙ্গত কারণে: স্বয়ংক্রিয় OpenAPI ডক্স, বাস্তব অ্যাসিঙ্ক সমর্থন, এবং Pydantic দ্বারা চালিত বৈধকরণের অনুরোধ যা আপনার ব্যবসার যুক্তিকে স্পর্শ করার আগেই খারাপ ইনপুট ধরে। এই টিউটোরিয়ালটি একটি খালি ফোল্ডার থেকে প্রমাণীকরণ-প্রস্তুত কাঠামো সহ একটি ডাটাবেস-ব্যাকড CRUD পরিষেবাতে একটি কার্যকরী REST API তৈরি করে।

সূচিপত্র

• প্রকল্প সেটআপ
• আপনার প্রথম রুট
• Pydantic সঙ্গে অনুরোধ এবং প্রতিক্রিয়া মডেল
• SQLModel এর সাথে একটি ডাটাবেস যোগ করা হচ্ছে
• সম্পূর্ণ CRUD এন্ডপয়েন্ট তৈরি করা
• ত্রুটি হ্যান্ডলিং
• স্বয়ংক্রিয় ডক্স
• প্রোডাকশনে চলছে
• FAQ

প্রকল্প সেটআপ

একটি ভার্চুয়াল পরিবেশ তৈরি করুন এবং মূল প্যাকেজগুলি ইনস্টল করুন:

mkdir fastapi-todo && cd fastapi-todo
python3 -m venv venv
source venv/bin/activate

pip install fastapi uvicorn[standard] sqlmodel

uvicornASGI সার্ভার যা আসলে আপনার অ্যাপ চালায়;sqlmodelএকটি ডাটাবেস স্তরের জন্য SQLAlchemy এবং Pydantic একত্রিত করে যা আপনার API স্কিমাগুলির সাথে মডেলগুলি ভাগ করে।

আপনার প্রথম রুট

তৈরি করুনmain.py:

from fastapi import FastAPI

app = FastAPI(title="Todo API")

@app.get("/")
def read_root():
    return {"message": "Todo API is running"}

হট রিলোড সহ এটি চালান:

uvicorn main:app --reload

দেখুনhttp://127.0.0.1:8000এবং আপনি JSON প্রতিক্রিয়া দেখতে পাবেন। দেখুন/docsএবং আপনার ইতিমধ্যেই একটি ইন্টারেক্টিভ সোয়াগার UI আছে — আপনার ফাংশন স্বাক্ষর ছাড়া আর কিছুই থেকে তৈরি নয়।

Pydantic সঙ্গে অনুরোধ এবং প্রতিক্রিয়া মডেল

ম্যানুয়ালি চেক করার পরিবর্তেrequest.json()অনুপস্থিত ক্ষেত্রগুলির জন্য, একটি মডেল সংজ্ঞায়িত করুন এবং FastAPI কে আপনার জন্য যাচাই করতে দিন:

from pydantic import BaseModel

class TodoCreate(BaseModel):
    title: str
    done: bool = False

@app.post("/todos")
def create_todo(todo: TodoCreate):
    return {"title": todo.title, "done": todo.done}

অনুপস্থিত একটি অনুরোধ পাঠানtitleএবং FastAPI স্বয়ংক্রিয়ভাবে একটি স্ট্রাকচার্ড 422 ত্রুটি প্রদান করে — কোনো অতিরিক্ত কোডের প্রয়োজন নেই। এই বৈধতা স্তরটি ফ্লাস্ক থেকে দলগুলি স্থানান্তরিত করার একক বৃহত্তম কারণ।

SQLModel এর সাথে একটি ডাটাবেস যোগ করা হচ্ছে

SQLModel একটি ক্লাসকে আপনার ডাটাবেস টেবিল এবং আপনার API স্কিমা উভয় হিসাবে পরিবেশন করতে দেয়:

from sqlmodel import SQLModel, Field, create_engine, Session

class Todo(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    done: bool = False

engine = create_engine("sqlite:///todos.db")

def init_db():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

কলinit_db()FastAPI এর লাইফস্প্যান হ্যান্ডলার ব্যবহার করে স্টার্টআপে:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    yield

app = FastAPI(title="Todo API", lifespan=lifespan)

সম্পূর্ণ CRUD এন্ডপয়েন্ট তৈরি করা

মডেল এবং সেশন নির্ভরতার সাথে, চারটি CRUD অপারেশন সংক্ষিপ্ত এবং পাঠযোগ্য:

from fastapi import Depends, HTTPException
from sqlmodel import select

@app.post("/todos", response_model=Todo)
def create_todo(todo: Todo, session: Session = Depends(get_session)):
    session.add(todo)
    session.commit()
    session.refresh(todo)
    return todo

@app.get("/todos", response_model=list[Todo])
def list_todos(session: Session = Depends(get_session)):
    return session.exec(select(Todo)).all()

@app.get("/todos/{todo_id}", response_model=Todo)
def get_todo(todo_id: int, session: Session = Depends(get_session)):
    todo = session.get(Todo, todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo

@app.put("/todos/{todo_id}", response_model=Todo)
def update_todo(todo_id: int, data: Todo, session: Session = Depends(get_session)):
    todo = session.get(Todo, todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    todo.title = data.title
    todo.done = data.done
    session.add(todo)
    session.commit()
    session.refresh(todo)
    return todo

@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int, session: Session = Depends(get_session)):
    todo = session.get(Todo, todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    session.delete(todo)
    session.commit()
    return {"ok": True}

নোটিশDepends(get_session)– ফাস্টএপিআই এর নির্ভরতা ইনজেকশন সিস্টেম। এটা চলেget_sessionপ্রতিটি অনুরোধের জন্য, ফলাফলটি আপনার ফাংশনে হস্তান্তর করুন এবং পরে এটি পরিষ্কার করুন। একই প্যাটার্ন প্রতিটি হ্যান্ডলারে বয়লারপ্লেটের পুনরাবৃত্তি না করে প্রমাণীকরণ, হার সীমিতকরণ এবং পৃষ্ঠায় স্থানান্তর করে।

ত্রুটি হ্যান্ডলিং

HTTPExceptionবেশিরভাগ ক্ষেত্রেই কভার করে, কিন্তু ডোমেন-নির্দিষ্ট ত্রুটির জন্য, একটি কাস্টম ব্যতিক্রম হ্যান্ডলার নিবন্ধন করুন যাতে প্রতিক্রিয়াগুলি API জুড়ে সামঞ্জস্যপূর্ণ থাকে:

from fastapi.responses import JSONResponse
from fastapi import Request

class TodoLimitError(Exception):
    pass

@app.exception_handler(TodoLimitError)
def limit_handler(request: Request, exc: TodoLimitError):
    return JSONResponse(status_code=400, content={"error": "todo limit reached"})

স্বয়ংক্রিয় ডক্স

আপনার লেখা প্রতিটি রুট ইতিমধ্যেই/docsএ নথিভুক্ত করা হয়েছে৷ (Swagger) এবং/redoc(ReDoc), আপনার Pydantic মডেল এবং টাইপ ইঙ্গিত থেকে উত্পন্ন. একটি ডকস্ট্রিং বাField(description=...)দিয়ে বিবরণ যোগ করুন এবং সেগুলি স্বয়ংক্রিয়ভাবে জেনারেট করা স্কিমাতে প্রদর্শিত হয় — হাতে রক্ষণাবেক্ষণের জন্য আলাদা কোনো OpenAPI YAML নেই।

প্রোডাকশনে চলছে

উৎপাদনের জন্য, একাধিক কর্মীদের সাথে গুনিকর্নের পিছনে ইউভিকর্ন চালান, অথবা ইউভিকর্নের নিজস্ব মাল্টি-ওয়ার্কার মোড ব্যবহার করুন:

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

এটিকে Nginx বা TLS সমাপ্তির জন্য একটি পরিচালিত লোড ব্যালেন্সারের পিছনে রাখুন এবংcreate_engineপরিবর্তন করে PostgreSQL এর জন্য SQLite অদলবদল করুন। সংযোগ স্ট্রিং – SQLModel এর API পরিবর্তন হয় না।

আপনি যদি এই APIটিকে একটি আধুনিক ফ্রন্টএন্ডের সাথে যুক্ত করেন, আমাদেরNext.js 15 এবং PostgreSQL গাইডক্লায়েন্ট সাইড কভার করে, এবং জাভাস্ক্রিপ্টে অ্যাসিঙ্ক ফান্ডামেন্টালের জন্য দেখুনআমাদের অ্যাসিঙ্ক/অপেক্ষা গভীর ডুব.

FAQ

ফাস্টএপিআই কি ফ্লাস্কের চেয়ে দ্রুত?
ফাস্টএপিআই-এর অ্যাসিঙ্ক সমর্থন এবং স্টারলেট ফাউন্ডেশন এটিকে একযোগে লোডের অধীনে যথেষ্ট দ্রুত করে তোলে, বিশেষ করে ডাটাবেস কল বা বহিরাগত API অনুরোধের মতো আই/ও-বাউন্ড এন্ডপয়েন্টের জন্য।

আমার কি প্রতিটি রুটের জন্য async def দরকার?
না। ফাস্টএপিআই সিঙ্ক্রোনাস চলেdefএকটি থ্রেড পুলে স্বয়ংক্রিয়ভাবে রুট. ব্যবহার করুনasync defযখন আপনি প্রকৃত একযোগে সুবিধা পেতে async লাইব্রেরি (async DB ড্রাইভার, httpx) কল করছেন।

Pydantic এবং SQLModel এর মধ্যে পার্থক্য কি?
Pydantic অনুরোধ এবং প্রতিক্রিয়ার জন্য ডেটা আকার যাচাই করে। SQLModel Pydantic মডেলগুলিকে প্রসারিত করে যাতে একই ক্লাস ডুপ্লিকেট স্কিমা সংজ্ঞা এড়িয়ে ডাটাবেস টেবিলে ম্যাপ করতে পারে।

আমি কিভাবে প্রমাণীকরণ যোগ করব?
ফাস্টএপিআই ব্যবহার করুনOAuth2PasswordBearerনির্ভরতা ইনজেকশন সহ – একটিget_current_userনির্ভরতা একটি JWT যাচাই করে এবং প্রমাণীকরণের যুক্তিকে এক জায়গায় রেখে সুরক্ষা প্রয়োজন এমন যেকোনো রুটে যোগ করা হয়।

ফাস্টএপিআই কি ওয়েবসকেটগুলি পরিচালনা করতে পারে?
হ্যাঁ, দেশীয়ভাবে।@app.websocket("/ws")দিয়ে একটি রুট সংজ্ঞায়িত করুন এবং একটিasync defহ্যান্ডলার যে অপেক্ষা করছেwebsocket.receive_text()একটি লুপে

নির্মাণের জন্য প্রস্তুত?

উপরের কোডটিকে একটি নতুন প্রকল্পে ক্লোন করুন, চালান ||||৷ , এবং নির্ধারিত তারিখ বা অগ্রাধিকারের মতো ক্ষেত্রগুলির সাথে টোডো মডেলটি প্রসারিত করা শুরু করুন৷ আপনি ডাটাবেস সেশনের জন্য যে নির্ভরতা ইনজেকশন প্যাটার্ন ব্যবহার করেছেন সেটিই আপনি প্রমাণীকরণ, ক্যাশিং এবং রেট সীমিত করার জন্য পরে পৌঁছাবেন।uvicorn main:app --reloadটিপি