FastAPI — async Python API, Pydantic আর স্বয়ংক্রিয় docs

১ · FastAPI কোথায় বসে: ASGI, Starlette আর Pydantic

এতদিন বাজারঘরের backend লিখেছেন Express-এ (পর্ব ২৫), input যাচাই করেছেন Zod দিয়ে (পর্ব ২৭), আর OpenAPI document লিখেছেন আলাদা করে (পর্ব ৩৫)। FastAPI এই তিনটি কাজ একটি জায়গা থেকে করে: Python-এর type hint. আপনি function-এর parameter-এ type লেখেন, আর সেই একই লেখা থেকে FastAPI তিনটি জিনিস পায়: ইনপুট রূপান্তর ও যাচাই, editor-এর autocomplete, আর API-র লিখিত চুক্তি।

WSGI বনাম ASGI: server আর app-এর মধ্যের চুক্তি

Python-এ web server (যে socket থেকে bytes পড়ে) আর web app (যে আপনার কোড চালায়) আলাদা। দুজনের মধ্যে একটি মানা চুক্তি লাগে। পুরোনো চুক্তির নাম WSGI (PEP 333, ২০০৩; Python 3-এর জন্য PEP 3333, ২০১০)। এতে app একটি সাধারণ function: একটি request নেয়, পুরো উত্তর ফেরত দেয়, আর ততক্ষণ সেই worker আটকে থাকে। Django আর Flask এই পথে বড় হয়েছে।

নতুন চুক্তি ASGI, যা Django Channels-এর কাজ থেকে (Andrew Godwin-এর নেতৃত্বে, ২০১৬ থেকে ২০১৯-এর মধ্যে) দাঁড়িয়েছে। এতে app একটি async function, যা তিনটি জিনিস পায়: scope (কোন path, কোন header), receive (client-এর পাঠানো বার্তা পড়ার জন্য) আর send (উত্তর পাঠানোর জন্য)। একটি connection অনেকক্ষণ খোলা রেখে অনেক বার্তা আদান-প্রদান করা যায়, তাই WebSocket (পর্ব ৬০) আর streaming (পর্ব ৬১) স্বাভাবিকভাবে চলে, আর অপেক্ষার সময় একই process অন্য request সামলায়।

Node.js-এ এই ভাগ ছিল না, কারণ Node শুরু থেকেই event loop-এ চলে (পর্ব ২৪)। Python-এ async/await এসেছে Python 3.5-এ (২০১৫), আর তার সাথে পুরো web জগতের জন্য নতুন চুক্তি লেগেছে। নিচের ছবিতে বাজারঘরের একটি request কোন স্তর পেরিয়ে যায়, তা দেখুন।

এখানে একটি কথা পরিষ্কার রাখুন: FastAPI নিজে কোনো socket খোলে না। uvicorn হলো server, FastAPI হলো app. Express-এ দুটি একসাথে আসে, Python-এ আলাদা। Python ভাষা, type hint আর async-এর মূল ধারণা এই কোর্সে আবার শেখানো হবে না। দরকার হলে Python কোর্সের type hint আর mypy আর asyncio পর্ব দেখুন। FastAPI-র hello world আর প্রথম CRUD আছে ওয়েব ব্যাকএন্ড — FastAPI-র শুরু পর্বে। এখানে আমরা একজন ফুল-স্ট্যাক ডেভেলপারের চোখে দেখব: Express-এর সাথে তুলনা, নিরাপত্তা, মাপা আচরণ আর সিদ্ধান্ত।

২ · প্রথম path operation: path, query আর type hint

FastAPI-তে একটি route-কে বলে path operation: একটি path (/products/{product_id}) আর একটি HTTP method (GET) মিলে একটি কাজ। Decorator দিয়ে function-টিকে সেই কাজের সাথে জোড়া হয়। Express-এর app.get('/products/:id', handler)-এর সাথে মিলিয়ে দেখুন।

from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Query, status

app = FastAPI(title='বাজারঘর API', version='1.0.0')

def pagination(limit: int = Query(default=20, le=100),
               offset: int = Query(default=0, ge=0)):
    return {'limit': limit, 'offset': offset}

@app.get('/products/{product_id}', response_model=ProductOut)
def get_product(product_id: int, db: DB):
    product = db.get(Product, product_id)
    if product is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, 'Product not found')
    return product

FastAPI নিয়মটি এভাবে পড়ে: path-এ {product_id} আছে, আর function-এ একই নামের parameter আছে, তাই এটি path parameter. Path-এ নেই এমন সাধারণ parameter, যেমন limit, হয়ে যায় query parameter (?limit=2). আর int লেখা মানে: URL-এ সবকিছু string হয়ে আসে, FastAPI তাকে সংখ্যায় রূপান্তর করবে, না পারলে 422. পর্ব ২৭-এ Zod-এ z.coerce.number() নিজে লিখতে হয়েছিল, এখানে type hint-ই সেটি করে। Query(le=100) একটি সীমা যোগ করে, যেন কেউ এক অনুরোধে লাখ সারি না চায় (পর্ব ৩৪)।

সত্যিকারের run · FastAPI 0.162.0, uvicorn 0.54.0 · ~/bazarghar-এ, 127.0.0.1-এ

$ uvicorn main:app --host 127.0.0.1 --port 8769
INFO:     Started server process [9904]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8769 (Press CTRL+C to quit)

$ curl -s "127.0.0.1:8769/products?limit=2"
[{"id":1,"name":"Miniket Rice 5kg","price":420,"stock":30,"category":{"id":1,"name":"Grocery"}},
 {"id":2,"name":"Soybean Oil 1L","price":190,"stock":0,"category":{"id":1,"name":"Grocery"}}]

$ curl -s 127.0.0.1:8769/products/3
{"id":3,"name":"Sylhet Tea 400g","price":260,"stock":18,"category":{"id":2,"name":"Tea"}}

$ curl -s -w '\nHTTP %{http_code}\n' 127.0.0.1:8769/products/abc
{"detail":[{"type":"int_parsing","loc":["path","product_id"],"msg":"Input should be a valid integer, unable to parse string as an integer","input":"abc"}]}
HTTP 422

$ curl -s -w '\nHTTP %{http_code}\n' "127.0.0.1:8769/products?limit=101"
{"detail":[{"type":"less_than_equal","loc":["query","limit"],"msg":"Input should be less than or equal to 100","input":"101","ctx":{"le":100}}]}
HTTP 422

Handler-এর ভেতরে একটিও if লেখা হয়নি, তবু abc আর 101 দুটিই থেমেছে। Error-এর loc বলে দেয় সমস্যা কোথায়: ["path","product_id"] মানে path-এ, ["query","limit"] মানে query-তে। Frontend এই loc দেখে ঠিক input-এর পাশে বার্তা দেখাতে পারে। প্রথম লাইনের JSON-এ category একটি nested object, যা আসছে SQLAlchemy-র একটি সম্পর্ক থেকে, অংশ ৮-এ দেখবেন।

নিচের ছোট মডেলটি দেখায় type hint থেকে যাচাই আসলে কোথা থেকে আসে। এটি Pydantic নয়, শুধু ধারণা দেখানোর জন্য Python-এর dataclasses আর typing.get_type_hints দিয়ে বানানো একটি খেলনা রূপ। Python runtime-এ class-এর type hint পড়তে পারে, আর FastAPI ও Pydantic ঠিক এই সুবিধাটিই কাজে লাগায়।

# শেখানোর মডেল: type hint পড়ে যাচাই, Pydantic-এর ছোট খেলনা রূপ
from dataclasses import dataclass, fields, MISSING
from typing import get_type_hints

@dataclass
class ProductIn:
    name: str
    price: int
    stock: int = 0

def validate(model, data):
    hints, errors, clean = get_type_hints(model), [], {}
    allowed = {f.name: f for f in fields(model)}
    for key in data.keys() - allowed.keys():
        errors.append(('extra_forbidden', key, 'Extra inputs are not permitted'))
    for name, f in allowed.items():
        if name not in data:
            if f.default is MISSING:
                errors.append(('missing', name, 'Field required'))
            continue
        value, typ = data[name], hints[name]
        try:
            clean[name] = typ(value)          # lax mode: "650" -> 650
        except (TypeError, ValueError):
            errors.append((f'{typ.__name__}_parsing', name, f'Input should be a valid {typ.__name__}'))
    if errors:
        return 422, errors
    return 201, model(**clean)

print(get_type_hints(ProductIn))
print(validate(ProductIn, {'name': 'Ghee 500g', 'price': '650'}))
print(validate(ProductIn, {'name': 'Ghee', 'price': 'ten', 'is_admin': True}))
print(validate(ProductIn, {'price': 95}))

৩ · Pydantic v2: request model, 422 আর “নরম” রূপান্তর

Request body-র জন্য একটি Pydantic model লিখুন। Function-এর parameter-এ সেই model দিলে FastAPI বোঝে, এটি JSON body থেকে আসবে।

from pydantic import BaseModel, ConfigDict, Field

class ProductIn(BaseModel):
    model_config = ConfigDict(extra='forbid')        # অচেনা field মানে 422
    name: str = Field(min_length=1, max_length=120)
    price: int = Field(gt=0, description='দাম, টাকায়')
    stock: int = Field(default=0, ge=0)
    category_id: int

@app.post('/products', response_model=ProductOut, status_code=201)
def create_product(item: ProductIn, db: DB, user: CurrentUser):
    ...

সত্যিকারের run · admin-এর token দিয়ে (token কীভাবে আসে, অংশ ৯-এ) · $AT হলো সেই token · পরের দুটি command-এ একই header, সংক্ষেপে ...

$ curl -s -X POST 127.0.0.1:8769/products -H "Authorization: Bearer $AT" \
    -H 'Content-Type: application/json' -d '{"name":"","price":0,"category_id":1}'
{"detail":[{"type":"string_too_short","loc":["body","name"],"msg":"String should have at least 1 character","input":"","ctx":{"min_length":1}},{"type":"greater_than","loc":["body","price"],"msg":"Input should be greater than 0","input":0,"ctx":{"gt":0}}]}
HTTP 422

$ curl ... -d '{"name":"Ghee 500g","price":650,"category_id":1,"cost_price":1}'
{"detail":[{"type":"extra_forbidden","loc":["body","cost_price"],"msg":"Extra inputs are not permitted","input":1}]}
HTTP 422

$ curl ... -d '{"name":"Ghee 500g","price":"650","category_id":1}'
{"id":8,"name":"Ghee 500g","price":650,"stock":0,"category":{"id":1,"name":"Grocery"}}
HTTP 201

তিনটি উত্তর তিনটি শিক্ষা দেয়। প্রথমটিতে দুটি ভুল একসাথে, প্রতিটির type, loc, msg, যে input এসেছিল সেটি, আর নিয়মের মান (ctx). দ্বিতীয়টিতে ক্রেতা চুপিচুপি cost_price (ক্রয়মূল্য, যা শুধু দোকানের জানার কথা) পাঠাতে চেয়েছিলেন। এটি mass assignment-এর চেষ্টা, পর্ব ২৭-এ Zod-এর .strict() দিয়ে যা আটকেছিলেন। extra='forbid' একই কাজ করে।

তৃতীয়টি অবাক করতে পারে: "650" একটি string, তবু গৃহীত হলো, আর ডেটাবেসে গেল সংখ্যা 650. Pydantic-এর ডিফল্ট হলো lax mode, মানে নিরাপদ মনে হলে সে coercion করে: সংখ্যার মতো দেখতে string থেকে সংখ্যা। কড়া হতে চাইলে strict=True. আর মনে রাখুন, extra-র ডিফল্ট forbid নয়, ignore: অচেনা field চুপচাপ ফেলে দেওয়া হয়।

সত্যিকারের run · Pydantic 2.14.2 · একই body দুই রকম model-এ

body = {'name': 'Ghee 500g', 'price': '650', 'is_admin': True}
default: Loose(name='Ghee 500g', price=650)
strict : int_type ('price',) Input should be a valid integer
strict : extra_forbidden ('is_admin',) Extra inputs are not permitted

ডিফল্ট model-এ is_admin নীরবে বাদ গেছে। এটি নিরাপদ (ডেটায় ঢোকেনি), কিন্তু client জানতেও পারল না যে তার পাঠানো একটি field কাজে লাগেনি। কোনটি বেছে নেবেন, তা দলের সিদ্ধান্ত। বাজারঘরে লেখার API-তে আমরা forbid রেখেছি, কারণ ভুল বানানের field (stok) নীরবে হারালে bug খুঁজে পাওয়া কঠিন।

পর্ব ৩৩-এ (Status code আর error format) দেখেছেন, “JSON পড়া যায় কিন্তু নিয়ম ভাঙে” মানে 422, আর ভুলের ছাঁচ সবখানে এক হওয়া উচিত। FastAPI-র {"detail":[...]} ছাঁচ RFC 9457-এর problem details নয়। দরকার হলে RequestValidationError-এর জন্য নিজের exception handler লিখে ছাঁচ বদলানো যায়, তবে তখন OpenAPI-তে লেখা 422-এর বর্ণনাও মিলিয়ে নিতে হয়।

৪ · response_model: যা পাঠাবেন না, তা আটকানো

ঢোকার দরজা পাহারা দেওয়া হলো। এবার বেরোনোর দরজা। ডেটাবেসের একটি সারিতে প্রায়ই এমন তথ্য থাকে যা client-এর জানার কথা নয়: ব্যবহারকারীর password hash, admin কি না, পণ্যের ক্রয়মূল্য। Handler যদি পুরো সারি ফেরত দেয়, সবই চলে যায়। একে বলে excessive data exposure.

পরীক্ষার জন্য বাজারঘরে দুটি route রাখা হলো, একই কাজ, একটি পার্থক্য। প্রথমটি AI-এর লেখা কোডে প্রায়ই দেখা “সব column-কে dict বানিয়ে ফেরত দাও” ধাঁচের, কোনো response_model নেই। দ্বিতীয়টিতে response_model=UserOut, যেখানে শুধু id আর email.

class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)   # ORM object থেকে পড়া যাবে
    id: int
    email: str

@app.get('/users/me', response_model=UserOut)
def me(user: CurrentUser):
    return user

@app.get('/users/me/raw')                              # ভুলটি: response_model নেই
def me_raw(user: CurrentUser):
    return {c.name: getattr(user, c.name) for c in User.__table__.columns}

সত্যিকারের run · একজন সাধারণ ক্রেতার token দিয়ে ($RT) · hash-টি আসল argon2id hash, একটি পরীক্ষার password-এর

$ curl -s 127.0.0.1:8769/users/me -H "Authorization: Bearer $RT"
{"id":2,"email":"rina@bazarghar.test"}

$ curl -s 127.0.0.1:8769/users/me/raw -H "Authorization: Bearer $RT"
{"id":2,"email":"rina@bazarghar.test","password_hash":"$argon2id$v=19$m=65536,t=3,p=4$uTtXniGM+dcOLUfoeD6nLQ$yDOg7eV8NYsFZWYRnEThcrKKUZ59cReFFzqGpD12Aio","is_admin":false}

দ্বিতীয় উত্তরে hash-এর সাথে hash-এর সব setting-ও আছে (memory ৬৫,৫৩৬ KiB, ৩ বার, ৪ lane). Argon2 ধীর বলে সঙ্গে সঙ্গে password বের করা কঠিন, কিন্তু hash বাইরে গেলে আক্রমণকারী নিজের যন্ত্রে যত খুশি অনুমান চালাতে পারেন, কোনো rate limit ছাড়া। আর is_admin জানিয়ে দেয় কোন account-এ আক্রমণ লাভজনক। এই ফাঁস ধরার মতো কোনো error হয়নি, status 200. তাই এটি ধরা পড়ে শুধু review বা test-এ।

বাজারঘরের Product table-এ cost_price আছে, কিন্তু ProductOut-এ নেই। তাই আগের সব product-এর উত্তরে ক্রয়মূল্য নেই, যদিও handler পুরো ORM object ফেরত দিয়েছে। Return type দিয়েও (-> ProductOut) একই কাজ হয়, আর দুটি একসাথে দিলে response_model প্রাধান্য পায়। এই সমস্যার পূর্ণ ছবি, যাকে OWASP API Security Top 10 (২০২৩) “Broken Object Property Level Authorization” বলে, পাবেন API নিরাপত্তা পর্বে।

৫ · স্বয়ংক্রিয় docs: OpenAPI 3.1 আর /docs

পর্ব ৩৫-এ (Versioning আর OpenAPI) OpenAPI document লিখতে আলাদা tool লেগেছিল, আর কোড বদলালে document পিছিয়ে পড়ার ভয় ছিল। FastAPI-তে document তৈরি হয় চলার সময়ে, কোড থেকেই। কিছু না লিখেই বাজারঘর API-তে দুটি ঠিকানা আছে: /openapi.json আর /docs.

সত্যিকারের run · /openapi.json থেকে বাছাই করা অংশ, Python দিয়ে ছাপানো

3.1.0 {'title': 'বাজারঘর API', 'version': '1.0.0'}
POST /token Login ['200', '422']
GET /users/me Me ['200']
GET /users/me/raw Me Raw ['200']
GET /products List Products ['200', '422']
POST /products Create Product ['201', '422']
GET /products/{product_id} Get Product ['200', '422']
POST /orders Place Order ['202']

components.schemas.ProductIn:
  "name":  {"type": "string", "maxLength": 120, "minLength": 1, "title": "Name"}
  "price": {"type": "integer", "exclusiveMinimum": 0.0, "title": "Price", "description": "দাম, টাকায়"}
  "stock": {"type": "integer", "minimum": 0.0, "title": "Stock", "default": 0}
  "category_id": {"type": "integer", "title": "Category Id"}
  "additionalProperties": false, "required": ["name", "price", "category_id"]

securitySchemes: {"OAuth2PasswordBearer": {"type": "oauth2",
  "flows": {"password": {"scopes": {}, "tokenUrl": "token"}}}}

$ curl -s 127.0.0.1:8769/docs | grep -E "title|swagger-ui-bundle"
    <title>বাজারঘর API - Swagger UI</title>
    <script src="https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui-bundle.js"></script>

লক্ষ করুন, কোডের প্রতিটি সিদ্ধান্ত document-এ চলে এসেছে। Field(gt=0) হলো exclusiveMinimum, extra='forbid' হলো additionalProperties: false, description='দাম, টাকায়' বাংলাসহ এসেছে, আর OAuth2PasswordBearer থেকে login-এর বর্ণনা। /users/me-তে 422 নেই, কারণ সেখানে যাচাই করার মতো কোনো parameter নেই। একটি ছোট খুঁত: সীমার মান 0.0 হিসেবে এসেছে, পূর্ণসংখ্যা হিসেবে নয়। কাজে প্রভাব নেই, কিন্তু document থেকে কোড বানানোর tool এ রকম খুঁটিনাটিতে আটকাতে পারে, তাই নিজের tool দিয়ে একবার যাচাই করুন।

@app.post('/products', response_model=ProductOut,
          status_code=201)
def create_product(item: ProductIn, db: DB,
                   user: CurrentUser): ...
decorator route-টি app-এর তালিকায় রাখে

for route in app.routes:
    body   ← ProductIn.model_json_schema()
    201    ← ProductOut
    422    ← HTTPValidationError
    security ← OAuth2PasswordBearer
ফল app.openapi_schema-তে জমা থাকে

{"openapi": "3.1.0",
 "info": {"title": "বাজারঘর API", ...},
 "paths": {"/products": {"post": {...}}},
 "components": {"schemas": {"ProductIn": {
   ..., "additionalProperties": false}}}}

<title>বাজারঘর API - Swagger UI</title>
<script src="https://cdn.jsdelivr.net/npm/
  swagger-ui-dist@5/swagger-ui-bundle.js">
url: '/openapi.json'
ছোট একটি HTML, ভারী অংশ CDN থেকে

Authorize  →  POST /token (form)
POST /products  {"name": "", "price": 0, ...}
422  string_too_short · greater_than
form-টি schema থেকে, request-টি আসল

নিচের খেলনা মডেলটি দ্বিতীয় ধাপের মূল ভাবনা দেখায়: একটি dataclass-এর type hint আর metadata পড়ে JSON Schema বানানো, তারপর সেটিকে OpenAPI-র কাঠামোয় বসানো।

# শেখানোর মডেল: dataclass থেকে OpenAPI-ধাঁচের JSON Schema
import json
from dataclasses import dataclass, field, fields, MISSING
from typing import get_type_hints

JSON_TYPE = {str: 'string', int: 'integer', float: 'number', bool: 'boolean'}

@dataclass
class ProductIn:
    name: str = field(metadata={'minLength': 1, 'maxLength': 120})
    price: int = field(metadata={'exclusiveMinimum': 0, 'description': 'price in taka'})
    stock: int = field(default=0, metadata={'minimum': 0})

def schema_of(model):
    hints, props, required = get_type_hints(model), {}, []
    for f in fields(model):
        prop = {'type': JSON_TYPE[hints[f.name]], **f.metadata}
        if f.default is MISSING:
            required.append(f.name)
        else:
            prop['default'] = f.default
        props[f.name] = prop
    return {'type': 'object', 'properties': props, 'required': required,
            'additionalProperties': False, 'title': model.__name__}

doc = {'openapi': '3.1.0', 'info': {'title': 'Bazarghar API', 'version': '1.0.0'},
       'paths': {'/products': {'post': {
           'requestBody': {'content': {'application/json': {
               'schema': {'$ref': '#/components/schemas/ProductIn'}}}},
           'responses': {'201': {'description': 'Created'},
                         '422': {'description': 'Validation Error'}}}}},
       'components': {'schemas': {'ProductIn': schema_of(ProductIn)}}}
print(json.dumps(doc['components'], indent=1))
print('paths:', [(m.upper(), p) for p, ops in doc['paths'].items() for m in ops])

৬ · Depends: ডেটাবেস session, বর্তমান ব্যবহারকারী আর test

পর্ব ২৯-এ (Backend-এর গঠন) দেখেছেন, dependency injection আসলে “দরকারি জিনিস parameter হিসেবে দেওয়া”। Express-এ সেটি আপনি হাতে জুড়েছিলেন। FastAPI-তে এর জন্য একটি ভাষা আছে: Depends. Parameter-এর পাশে লিখে দিন কোন function মানটি বানাবে, বাকিটা FastAPI করবে।

def get_db():
    with Session(engine) as session:     # request শেষে session বন্ধ
        yield session

DB = Annotated[Session, Depends(get_db)]

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)], db: DB) -> User:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])  # ভুলে 401
    user = db.get(User, int(payload['sub']))
    ...
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]

@app.post('/products', response_model=ProductOut, status_code=201)
def create_product(item: ProductIn, db: DB, user: CurrentUser):
    if not user.is_admin:
        raise HTTPException(status.HTTP_403_FORBIDDEN, 'Admins only')
    ...

তিনটি জিনিস লক্ষ করুন। এক, get_db-এ return নয়, yield: এটি একটি yield dependency. yield-এর আগের অংশ request-এর আগে চলে, পরের অংশ (এখানে with-এর শেষ, মানে session বন্ধ) কাজ শেষে। দুই, dependency-র নিজেরও dependency থাকতে পারে: get_current_user নিজেই DB চায়। তিন, Annotated দিয়ে একবার নাম দিলে (DB, CurrentUser) প্রতিটি route-এ শুধু ছোট নামটি লিখলেই হয়।

নিচের খেলনা resolver দেখায় FastAPI কীভাবে parameter পড়ে dependency ডাকে। এটি আসল কোড নয়, তবে মূল তিনটি নিয়ম এখানে আছে: signature পড়া, এক request-এ একবারই বানানো (cache), আর yield-এর পরের অংশ শেষে চালানো।

# শেখানোর মডেল: Depends-এর মতো একটি ছোট resolver
import inspect

class Depends:
    def __init__(self, fn): self.fn = fn

def solve(fn, request, cache, cleanups):
    kwargs = {}
    for name, p in inspect.signature(fn).parameters.items():
        if isinstance(p.default, Depends):
            dep = p.default.fn
            if dep not in cache:                       # একই request-এ একবারই
                value = solve(dep, request, cache, cleanups)
                if inspect.isgenerator(value):         # yield থাকলে: setup, পরে cleanup
                    cleanups.append(value)
                    value = next(value)
                cache[dep] = value
            kwargs[name] = cache[dep]
        else:
            kwargs[name] = request[name]
    return fn(**kwargs)

def get_db():
    print('  open session'); yield 'session#1'; print('  close session')

def get_current_user(db=Depends(get_db)):
    print(f'  find user with {db}'); return {'id': 2, 'email': 'rina@bazarghar.test'}

def place_order(product_id, db=Depends(get_db), user=Depends(get_current_user)):
    return f'order for product {product_id} by user {user["id"]} using {db}'

cleanups = []
print('result:', solve(place_order, {'product_id': 3}, {}, cleanups))
for gen in reversed(cleanups):
    next(gen, None)                                    # yield-এর পরের অংশ চলে

এবার পুরো ছবি। নিচের stepper-এ একটি আসল request, admin-এর token আর string-এ লেখা দাম "650" নিয়ে, ধাপে ধাপে চলছে। ক্রমটি আসল FastAPI-তে একটি ছোট probe চালিয়ে মিলিয়ে নেওয়া হয়েছে।

1POST /products  Bearer …  {"price":"650",…}2route match → create_product3Depends: oauth2_scheme, get_db, get_current_user4ProductIn.model_validate(body)5create_product(item, db, user) → Product6response_model=ProductOut → JSON 201

Test-এ dependency বদলানো

Depends-এর সবচেয়ে বড় উপহার test-এ। app.dependency_overrides একটি dictionary: আসল dependency-কে key করে নকল function দিলে পুরো app-এ সেটিই চলে। আসল token বানাতে হয় না, আসল ডেটাবেসও লাগে না। নিচে pytest দিয়ে চারটি test, মেমরির SQLite আর নকল ব্যবহারকারী নিয়ে। Test-এর তত্ত্ব আছে Test double পর্বে, আর Python-এ pytest-এর মূল কথা pytest পর্বে।

from fastapi.testclient import TestClient
from main import app, get_current_user, get_db

def as_user(is_admin):
    return lambda: User(id=1, email='t@bazarghar.test', password_hash='x', is_admin=is_admin)

def test_buyer_gets_403(client):                  # client fixture: get_db বদলানো
    app.dependency_overrides[get_current_user] = as_user(False)
    r = client.post('/products', json={'name': 'Green Tea', 'price': 180, 'category_id': 1})
    assert r.status_code == 403

def test_extra_field_rejected(client):
    app.dependency_overrides[get_current_user] = as_user(True)
    r = client.post('/products', json={'name': 'Ghee', 'price': 650,
                                       'category_id': 1, 'cost_price': 1})
    assert r.json()['detail'][0]['type'] == 'extra_forbidden'

সত্যিকারের run · pytest 9.2.0, httpx 0.28.1 · path ছোট করা হয়েছে

$ python -m pytest -q test_api.py
....                                                                     [100%]
============================== warnings summary ===============================
venv\Lib\site-packages\fastapi\testclient.py:1
  ~/bazarghar/venv/…/fastapi/testclient.py:1: StarletteDeprecationWarning: Using `httpx` with `starlette.testclient` is deprecated; install `httpx2` instead.
    from starlette.testclient import TestClient as TestClient  # noqa

4 passed, 1 warning in 0.57s

$ pip install httpx2==2.16.0
$ python -m pytest -q test_api.py
....                                                                     [100%]
4 passed in 0.42s

এই warning একটি ভালো শিক্ষা। TestClient আসলে Starlette-এর। লেখার সময়ে FastAPI-র testing পাতা এখনো বলে “TestClient ব্যবহার করতে httpx install করুন”, কিন্তু Starlette 1.10-এর source আগে httpx2 খোঁজে, না পেলে httpx নিয়ে warning দেয়। httpx2 (2.16.0) install করার পরে warning চলে গেল, test একই। Docs আর আসল আচরণ মেলেনি, তাই সিদ্ধান্ত নিন আপনার installed version-এর আচরণ দেখে, আর warning-কে উপেক্ষা না করে পড়ুন। আজ যা warning, পরের version-এ তা error হতে পারে।

৭ · async def, def আর আটকে যাওয়া event loop

FastAPI-তে path operation দুইভাবে লেখা যায়, আর দুটির চলার জায়গা আলাদা। async def চলে event loop-এ, যে একটি thread-এ অনেক request পালা করে চালায়। সাধারণ def FastAPI পাঠায় একটি thread pool-এ, যাতে সে loop আটকাতে না পারে। পর্ব ২৪-এ (Node.js server) Node-এর loop আটকানো দেখেছেন। Python-এ ফাঁদটি আরও সহজে পাতা যায়, কারণ async def লিখলেই কোড async হয় না।

বাজারঘরের ল্যাবে তিনটি route বানানো হলো, প্রতিটি ১ সেকেন্ডের “ধীর কাজ” করে, আর একটি /ping যা কিছুই করে না।

@app.get('/async-await')
async def async_await():
    await asyncio.sleep(1)          # অপেক্ষা, কিন্তু loop অন্যদের চালায়
    return {'ok': True}

@app.get('/async-blocking')
async def async_blocking():
    time.sleep(1)                   # পুরো event loop থেমে যায়
    return {'ok': True}

@app.get('/sync-def')
def sync_def():
    time.sleep(1)                   # সাধারণ def: FastAPI thread pool-এ চালায়
    return {'ok': True}

একটি client প্রতিটি route-এ একসাথে ৫টি request পাঠাল, আর ০.১ সেকেন্ড পরে একটি /ping.

সত্যিকারের run · uvicorn 0.54.0, একটি process · একটি local কম্পিউটারে, দুবার চালানো, দুবারই একই ছবি

/async-await     5 requests at once -> finished at 1.02 1.02 1.02 1.02 1.02 s; /ping sent at 0.10 s answered at 0.11 s
/async-blocking  5 requests at once -> finished at 1.01 5.02 5.02 5.02 5.02 s; /ping sent at 0.10 s answered at 5.02 s
/sync-def        5 requests at once -> finished at 1.02 1.02 1.02 1.02 1.02 s; /ping sent at 0.10 s answered at 0.11 s

1t=0.00  5 × GET /async-blocking এলো2t=0.10  GET /ping এলো, loop ব্যস্ত3t=1.00  req1 শেষ, req2 time.sleep শুরু4t=1→5   req2…req5 একে একে5t=5.02  বাকি সব উত্তর আর /ping6তুলনা: await বা def

Thread pool-এরও সীমা আছে

তাহলে কি সব def লিখলেই হলো? Thread pool অসীম নয়। FastAPI সাধারণ def চালায় AnyIO-র thread pool-এ, যার ডিফল্ট সীমা ৪০টি thread. ল্যাবে ৫০টি request একসাথে পাঠিয়ে মাপা হলো।

সত্যিকারের run · প্রতিটি request ১ সেকেন্ডের কাজ · একটি local কম্পিউটারে, দুবার

/sync-def    50 at once: 40 done by 1.5 s, last at 2.05 s
/async-await 50 at once: 50 done by 1.5 s, last at 1.32 s
/sync-def    50 at once: 40 done by 1.5 s, last at 2.04 s
/async-await 50 at once: 50 done by 1.5 s, last at 1.33 s

হিসাবটি সোজা। ৪০টি thread প্রথম সেকেন্ডে ৪০টি request শেষ করল, বাকি ১০টি দ্বিতীয় দফায়: ⌈৫০ ÷ ৪০⌉ = ২ দফা × ১ সেকেন্ড = প্রায় ২ সেকেন্ড। await-এর পথে কোনো thread লাগে না, তাই ৫০টিই প্রথম দফায়, শেষটি ১.৩ সেকেন্ডে (বাড়তি সময়টুকু ৫০টি connection সামলানোর)। এগুলো একটি local কম্পিউটারের মাপ, সাধারণ benchmark নয়। কিন্তু সূত্রটি সাধারণ: সাধারণ def-এ একসাথে চলা blocking কাজের সংখ্যা thread pool-এর আকারে বাঁধা।

Python-এর asyncio-র মূল ধারণা, gather আর TaskGroup, পাবেন asyncio পর্বে। নিচের খেলনা মডেলটি server ছাড়াই একই ঘটনা দেখায়: একটি “heartbeat” task যার প্রতি ০.০৫ সেকেন্ডে টিক দেওয়ার কথা, আর তিনটি handler.

# শেখানোর মডেল: এক event loop, তিনটি request, আর একটি heartbeat
import asyncio, time

async def handler(name, blocking, t0, log):
    if blocking:
        time.sleep(0.1)              # পুরো loop থেমে যায়
    else:
        await asyncio.sleep(0.1)     # অপেক্ষার সময় loop অন্যদের চালায়
    log.append(f'{name} done at {time.perf_counter() - t0:.1f} s')

async def heartbeat(t0, ticks):
    for _ in range(4):
        ticks.append(round(time.perf_counter() - t0, 2))
        await asyncio.sleep(0.05)

async def run(blocking):
    t0, log, ticks = time.perf_counter(), [], []
    await asyncio.gather(heartbeat(t0, ticks),
                         *(handler(f'req{i}', blocking, t0, log) for i in (1, 2, 3)))
    label = 'time.sleep   ' if blocking else 'asyncio.sleep'
    print(f'{label}: ' + ', '.join(log))
    gaps = [b - a for a, b in zip(ticks, ticks[1:])]
    print(f'{"":15}heartbeat should tick every 0.05 s, largest gap {max(gaps):.2f} s')

asyncio.run(run(blocking=False))
asyncio.run(run(blocking=True))

৮ · SQLAlchemy 2: model, সম্পর্ক আর query গোনা

FastAPI কোনো ORM সাথে আনে না, এটি Django-র সাথে বড় পার্থক্য। Python জগতে সবচেয়ে প্রচলিত পছন্দ SQLAlchemy (Mike Bayer, ২০০৬ থেকে), যার 2.0 সংস্করণ (২০২৩) type hint-ভিত্তিক নতুন লেখার ধরন এনেছে। লেখার সময়ে এই ল্যাবে SQLAlchemy 2.1.6, ডেটাবেস SQLite. FastAPI-র লেখকের আরেকটি library SQLModel একই class-কে Pydantic model আর SQLAlchemy table দুটিই বানায়। এটি সুবিধাজনক, তবে লেখার সময়ে তার version 0.0.48, আর অংশ ৪-এর কারণে ঢোকা-বেরোনোর model আলাদা রাখার নিয়ম সেখানেও মানতে হয়। এখানে আমরা সরাসরি SQLAlchemy ব্যবহার করেছি। ORM-এর ধারণা পর্ব ৩৮-এ, SQLAlchemy-র বিস্তারিত Python কোর্সের SQLAlchemy পর্বে।

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Category(Base):
    __tablename__ = 'categories'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(80), unique=True)
    products: Mapped[list['Product']] = relationship(back_populates='category')

class Product(Base):
    __tablename__ = 'products'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120))
    price: Mapped[int]
    stock: Mapped[int] = mapped_column(default=0)
    cost_price: Mapped[int] = mapped_column(default=0)    # শুধু দোকানের জন্য
    category_id: Mapped[int] = mapped_column(ForeignKey('categories.id'))
    category: Mapped[Category] = relationship(back_populates='products')

এখানেও type hint কাজ করছে: Mapped[int] থেকে SQLAlchemy column-এর type বোঝে, আর Mapped[list['Product']] থেকে বোঝে এটি এক-থেকে-অনেক সম্পর্ক (একটি category, অনেক product). এটিই পর্ব ৪১-এর (সম্পর্ক) shop-এর মডেল, Prisma-র বদলে Python-এ। এই মডিউলের অন্য framework গুলোর মতো একই ডেটা দেওয়া হলো: ৩টি category, ৬টি product. তারপর সব product তার category-র নামসহ পড়া হলো, তিনভাবে, আর প্রতিবার SQL গোনা হলো।

সত্যিকারের run · SQLAlchemy 2.1.6, SQLite · প্রতিবার নতুন session · SQL-এর প্রথম ৭০ অক্ষর

$ python seed.py
Seeded 3 categories, 6 products, 2 users
lazy (default): 6 products, 4 queries
    SELECT products.id, products.name, products.price, products.stock, pro
    SELECT categories.id, categories.name
    SELECT categories.id, categories.name
    SELECT categories.id, categories.name
selectinload: 6 products, 2 queries
    SELECT products.id, products.name, products.price, products.stock, pro
    SELECT categories.id, categories.name
joinedload: 6 products, 1 queries
    SELECT products.id, products.name, products.price, products.stock, pro

এখানে একটি মজার পার্থক্য আছে। পর্ব ৪২-এ (N+1 সমস্যা) আর এই মডিউলের অন্য ORM-এর run-এ একই ডেটায় lazy পথ দিয়েছিল ৭টি query: ১টি product-এর জন্য, আর প্রতিটি product-এর জন্য ১টি। SQLAlchemy দিল ৪টি: ১ + ৩. কারণ তার identity map. একটি many-to-one সম্পর্কে (product থেকে category) SQLAlchemy আগে দেখে, এই primary key-র category কি এই session-এ আগেই আনা হয়েছে? হলে আর query নেই। তাই ৬টি product-এর জন্য ৩টি আলাদা category-ই আনা হলো। এটি N+1-এর সমাধান নয়, শুধু ভাগ্য: ১০০টি category-তে ১০০টি query হবে।

আসল সমাধান eager loading. selectinload দ্বিতীয় একটি query-তে WHERE id IN (…) দিয়ে সব category একবারে আনে: মোট ২টি, product যতই হোক। joinedload একটি JOIN দিয়ে ১টিতে করে, many-to-one-এ ভালো, কিন্তু one-to-many-তে সারি বহুগুণ হতে পারে। বাজারঘরের /products route তাই selectinload ব্যবহার করে, আর server log-এ নিজেই query গোনে:

@app.get('/products', response_model=list[ProductOut])
def list_products(db: DB, page: Annotated[dict, Depends(pagination)]):
    stmt = (select(Product).options(selectinload(Product.category))
            .order_by(Product.id).limit(page['limit']).offset(page['offset']))
    return db.scalars(stmt).all()

সত্যিকারের run · uvicorn log থেকে

list_products: 2 SQL queries
INFO:     127.0.0.1:8387 - "GET /products?limit=2 HTTP/1.1" 200 OK

Migration-এর জন্য SQLAlchemy-র সঙ্গী tool হলো Alembic, যা পর্ব ৪০-এর (Migration আর seed) কাজটি করে। এই ল্যাবে শুধু create_all ব্যবহার হয়েছে, যা নতুন table বানায় কিন্তু পুরোনো table বদলায় না, তাই আসল project-এ এটি যথেষ্ট নয়। Async route-এ ডেটাবেস চাইলে SQLAlchemy-র AsyncSession আর async driver (যেমন PostgreSQL-এর জন্য asyncpg) লাগে। সেটি এখানে চালানো হয়নি।

৯ · Login, background কাজ, AI মডেল আর কখন FastAPI

OAuth2 password flow আর JWT

FastAPI-র security tool গুলো OAuth2-এর নামে, কারণ OpenAPI-তে সেভাবে লিখলে /docs-এ “Authorize” বোতাম আসে। বাজারঘরের login হলো OAuth2 password flow: client /token-এ form হিসেবে username আর password পাঠায়, উত্তরে পায় একটি JWT access token, তারপর প্রতিটি request-এ Authorization: Bearer …. JWT-এর গঠন, মেয়াদ, refresh আর browser-এ রাখার প্রশ্ন পর্ব ৪৬-এ (JWT login) আছে, তাই এখানে শুধু FastAPI-র অংশ।

import jwt                                   # PyJWT
from pwdlib import PasswordHash
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

SECRET_KEY = 'example-secret-change-me-32-bytes-long'   # আসলে environment থেকে
ALGORITHM = 'HS256'
password_hash = PasswordHash.recommended()               # Argon2
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')

@app.post('/token')
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()], db: DB):
    user = db.scalar(select(User).where(User.email == form.username))
    if user is None or not password_hash.verify(form.password, user.password_hash):
        raise HTTPException(401, 'Wrong email or password',
                            headers={'WWW-Authenticate': 'Bearer'})
    exp = datetime.now(timezone.utc) + timedelta(minutes=15)
    token = jwt.encode({'sub': str(user.id), 'exp': exp}, SECRET_KEY, algorithm=ALGORITHM)
    return {'access_token': token, 'token_type': 'bearer'}

সত্যিকারের run · PyJWT 2.15.1, pwdlib 0.3.1, python-multipart 0.0.33 · পরীক্ষার account আর পরীক্ষার password · কিছু header বাদ, status পাশে লেখা, token ছোট করা

$ curl -s -i -X POST 127.0.0.1:8769/products -H 'Content-Type: application/json' -d '{"name":"Green Tea","price":180,"category_id":2}'
HTTP/1.1 401 Unauthorized
www-authenticate: Bearer
{"detail":"Not authenticated"}

$ curl -s -X POST 127.0.0.1:8769/token -d 'username=rina@bazarghar.test&password=wrong'
{"detail":"Wrong email or password"}      HTTP 401

$ curl -s -X POST 127.0.0.1:8769/token -d 'username=rina@bazarghar.test&password=example-pass-2'
{"access_token":"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIyIiwiZXhwIjoxNzkzNjQ3NDQ4fQ.xxuv…","token_type":"bearer"}

$ curl -s -X POST 127.0.0.1:8769/products -H "Authorization: Bearer $RT" ...
{"detail":"Admins only"}                  HTTP 403

$ curl -s 127.0.0.1:8769/users/me -H "Authorization: Bearer abc.def.ghi"
{"detail":"Invalid token"}                HTTP 401

401 আর 403-এর পার্থক্য এখানে স্পষ্ট (পর্ব ৩৩): token নেই বা ভুল মানে “আপনি কে জানি না”, 401. Rina-র token ঠিক কিন্তু তিনি admin নন, মানে “জানি, কিন্তু অনুমতি নেই”, 403. Login-এর form পড়তে python-multipart package লাগে, না থাকলে FastAPI চালুর সময়েই বলে দেয়। এই ধাঁচের password flow নিজের প্রথম-পক্ষের app-এর জন্য, যেখানে আপনার নিজের login পাতা। অন্য কারও app-কে ব্যবহারকারীর password দেওয়ার জন্য OAuth-এর এই grant এখন নিরুৎসাহিত, সেখানে authorization code flow (পর্ব ৪৭)। নিরাপত্তার পূর্ণ তত্ত্ব Session আর token পর্বে।

BackgroundTasks বনাম আসল queue

অর্ডারের পরে invoice পাঠানো ক্রেতার অপেক্ষার অংশ হওয়া উচিত নয় (পর্ব ৬২)। FastAPI-তে সবচেয়ে সহজ উপায় BackgroundTasks: উত্তর পাঠানোর পরে একই process-এ একটি function চালানো।

def send_invoice(order_id: int):
    time.sleep(1)                      # ধরে নিন: PDF বানানো, mail পাঠানো
    log.info('invoice for order %d sent', order_id)

@app.post('/orders', status_code=202)
def place_order(background: BackgroundTasks, user: CurrentUser):
    order_id = 501
    background.add_task(send_invoice, order_id)
    log.info('order %d accepted, responding now', order_id)
    return {'order_id': order_id, 'status': 'accepted'}

সত্যিকারের run · client-এর মাপ আর server log

$ curl -s -w '\nHTTP %{http_code}  time %{time_total}s\n' -X POST 127.0.0.1:8769/orders -H "Authorization: Bearer $RT"
{"order_id":501,"status":"accepted"}
HTTP 202  time 0.003008s

order 501 accepted, responding now
INFO:     127.0.0.1:8470 - "POST /orders HTTP/1.1" 202 Accepted
invoice for order 501 sent

ক্রেতা ৩ ms-এ উত্তর পেলেন, আর log-এর ক্রম বলে invoice গেছে উত্তরের পরে। কিন্তু এটি পর্ব ৬২-এর fire-and-forget-এর আত্মীয়: কাজটি process-এর মেমরিতে। Deploy বা crash-এ হারাবে, retry নেই, আর কাজটি সেই server-এরই CPU খায়। Log লেখা, ছোট cache গরম করা, এমন হারালে-ক্ষতি-নেই কাজে BackgroundTasks ঠিক আছে। Invoice, টাকা, ক্রেতাকে দেওয়া কথার জন্য আসল job queue লাগে। Python-এ প্রচলিত উদাহরণ Celery, ARQ বা Dramatiq, বা PostgreSQL-ভিত্তিক queue. কোনটি, তা আপনার পরিকাঠামোর ওপর।

AI মডেল serve করা: একটি প্রচলিত ব্যবহার

অনেক দল FastAPI বেছে নেয় একটি সাধারণ কারণে: তাদের machine learning মডেল আগে থেকেই Python-এ (scikit-learn, PyTorch ইত্যাদি)। মডেলের চারপাশে একটি typed HTTP API, input যাচাই Pydantic-এ, আর docs নিজে থেকে, এটি model serving-এর একটি সহজ শুরু। বাজারঘরে যেমন “এই পণ্যের সাথে আর কী কেনা হয়” সুপারিশের মডেল একটি আলাদা FastAPI সেবা হতে পারে, আর মূল Next.js অ্যাপ তাকে HTTP-তে ডাকতে পারে।

কিন্তু অংশ ৭-এর ফাঁদ এখানে সবচেয়ে বড়। মডেল চালানো (inference) সাধারণত CPU-bound বা GPU-র কাজ, await-যোগ্য অপেক্ষা নয়। async def-এর ভেতরে model.predict(...) লিখলে প্রতিটি অনুমানের সময় পুরো server থেমে যায়। সাধারণ পথ হলো: def route (thread pool), বা আলাদা worker process, বা মডেল চালানোর জন্য বিশেষ serving tool, আর বড় মডেলের উত্তর stream করা (পর্ব ৬১)। মডেলটি চালুর সময় একবার মেমরিতে আনুন (FastAPI-র lifespan), প্রতি request-এ নয়। কোন পথ আপনার জন্য, তা মডেলের আকার আর ভিড় মেপে ঠিক করুন।

কখন FastAPI বেছে নেবেন

FastAPI-র নিজের পাতায় গতির বড় দাবি আছে, কিন্তু বাজারঘরের মতো অ্যাপে সময়ের বেশিরভাগ যায় ডেটাবেসে আর network-এ, framework-এর ভেতরে নয়। তাই framework বাছাইয়ে গতির চেয়ে বড় প্রশ্ন: আপনার দল কোন ভাষায় দক্ষ, কোন library লাগবে, আর কত কিছু নিজে জুড়তে রাজি।

১০ · AI-এর যুগে: FastAPI কোড যাচাই

FastAPI-র কোড ছোট আর পড়তে সহজ, তাই AI খুব দ্রুত একটি “চলা” API লিখে দেয়। সমস্যা হলো, এই পর্বের প্রায় প্রতিটি ভুল কোনো error দেয় না: password hash ফাঁস হলে status 200, loop আটকালেও উত্তর আসে, শুধু দেরিতে। AI-এর সাধারণ ভুলগুলো:

AI-এর লেখা FastAPI কোড হাতে পেলে

• pip show দিয়ে installed version দেখে সেই version-এর docs-এ প্রতিটি import আর option পেয়েছি
• প্রতিটি route-এ response_model আছে, আর উত্তরের JSON-এ গোপন কোনো key নেই
• ঢোকার model-এ অচেনা field পাঠিয়ে 422 দেখেছি
• প্রতিটি async def-এর ভেতরে কোনো blocking call নেই, একসাথে request পাঠিয়ে মেপেছি
• Token ছাড়া, অন্যের token দিয়ে আর ঠিক token দিয়ে তিনটি test লিখেছি
• তালিকার route-এ SQL query গুনেছি, N+1 নেই
• pytest-এর warning পড়েছি, শুধু “passed” দেখে থামিনি
• Secret, database URL আর CORS-এর তালিকা environment থেকে আসে, কোডে নেই