প্রজেক্ট, পর্ব ১: একটা ডেটা এক্সপ্লোরার API বানানো

এবার পুরো টিউটোরিয়ালটা একটা প্রজেক্টে কাজে লাগানোর পালা। এই পাতা আর পরের পাতা মিলিয়ে আপনি বানাবেন, টেস্ট করবেন আর প্রকাশ করবেন BitByte Data Explorer API: একটা ছোট FastAPI ব্যাকএন্ড, যা আরেকটা API-কে (BitByte Practice API) কল করে, যা ফেরত আসে তা যাচাই করে, পরিষ্কার উত্তরের আকারে সাজায়, আর আপস্ট্রিমের (upstream) গোলমাল স্পষ্ট এরর মেসেজের আড়ালে রাখে। এই পাতায় আপনি পরিকল্পনা করবেন আর একটা ফিচার ব্রাঞ্চে সেটা বানাবেন; পরের পাতায় টেস্ট লিখবেন, ডকুমেন্টেশন করবেন আর একটা পুল রিকোয়েস্টের মাধ্যমে মার্জ করবেন।

AI-এর কাজে এই গড়নটা সব জায়গায়। চ্যাটবটের ব্যাকএন্ড কল করে একটা LLM প্রোভাইডারকে; RAG সার্ভিস কল করে ভেক্টর ডেটাবেস আর এমবেডিং API-কে; একটা এজেন্টের "টুল" আসলে ছোট ছোট এন্ডপয়েন্ট, যারা অন্য সিস্টেমকে কল করে। প্রতিটা ক্ষেত্রে আপনার ব্যাকএন্ড বসে থাকে একজন ক্লায়েন্ট আর এমন একটা আপস্ট্রিম সার্ভিসের মাঝখানে, যেটা আপনার নিয়ন্ত্রণে নেই। আর তার কাজও একই: কখনো আটকে না থাকা, কখনো আবর্জনা ডেটা সামনে না পাঠানো, আর সবসময় এমন উত্তর দেওয়া যা ক্লায়েন্ট বুঝতে পারে।

যা শিখবেন

• কোড লেখার আগে একটা ছোট ব্রিফকে এন্ডপয়েন্টের তালিকা, রেসপন্সের আকার আর এররের নিয়মে বদলানো।
• uv আর Git দিয়ে প্রজেক্ট সাজানো, আর একটা ফিচার ব্রাঞ্চে ছোট ছোট কমিটে কাজ করা।
• আপস্ট্রিমের সব কল একটা মডিউলে রাখা: প্রতিটা রিকোয়েস্টে টাইমআউট, পেজিনেশন, ফেরত আসা ডেটার ভ্যালিডেশন, আর সব ধরনের গোলমালের জন্য একটাই এরর টাইপ।
• টাইপ করা প্যারামিটার, রেসপন্স মডেল আর সঠিক স্ট্যাটাস কোড (404, 422, 502, 504) দিয়ে পরিষ্কার FastAPI এন্ডপয়েন্ট বানানো।
• curl দিয়ে তৈরি API পরীক্ষা করা, আর ইচ্ছে করে ভেঙে দেখা যে এররের পথগুলো ঠিকমতো কাজ করছে।

কাজের বিবরণ (ব্রিফ)

BitByte টিম তাদের ওয়েবসাইটের জন্য, আর পরে একটা AI স্টাডি অ্যাসিস্ট্যান্টের জন্য, একটা ছোট "ডেটা এক্সপ্লোরার" সার্ভিস চায়। অ্যাসিস্ট্যান্ট সরাসরি Practice API-এর সাথে কথা বলবে না; বরং এক জায়গা থেকে ছোট, অনুমানযোগ্য উত্তর পাবে। প্রোডাক্ট ওনার ব্রিফটা লিখেছেন এভাবে:

"আমাদের কোর্সগুলোর তালিকা দেখাও, লেভেল আর রেটিং দিয়ে ছাঁকো, একটা কোর্স আলাদা করে দেখাও, প্রতি লেভেলের সারাংশ দাও, আর একটা শহরের আজকের আবহাওয়া এক লাইনের পরামর্শসহ দেখাও। ডেটার উৎস ধীর বা বন্ধ থাকলে ক্র্যাশ না করে সেটা পরিষ্কার করে জানাও।"

আগে এটাকে একটা চুক্তিতে (contract) বদলে নিন। এই টেবিলটাই আপনি বানাবেন, আর পরের পাতার README-তেও এটাই লেখা থাকবে:

আর যে নিয়মগুলো সব এন্ডপয়েন্টের জন্য খাটে, অর্থাৎ "নন-ফাংশনাল রিকোয়ারমেন্ট":

• আপস্ট্রিমের প্রতিটা কলে টাইমআউট। ধীর আপস্ট্রিমের কারণে আমাদের সার্ভার আটকে থাকবে না। বেশি দেরি হলে → 504 Gateway Timeout।
• আপস্ট্রিমের গোলমাল হয়ে যায় 502 Bad Gateway, সাথে ছোট একটা detail মেসেজ: পৌঁছানো যাচ্ছে না, 5xx, JSON নয় এমন বডি, বা অপ্রত্যাশিত আকারের ডেটা। ক্লায়েন্ট কখনো Python ট্রেসব্যাক বা 500 দেখবে না।
• আপস্ট্রিমের ডেটা ব্যবহারের আগে Pydantic দিয়ে যাচাই হয়। অন্য সিস্টেম থেকে আসা ডেটাও ইনপুট, ঠিক ব্যবহারকারীর রিকোয়েস্টের মতো।
• কনফিগারেশন আসে এনভায়রনমেন্ট ভেরিয়েবল থেকে (PRACTICE_API_URL, UPSTREAM_TIMEOUT), যাতে একই কোড ল্যাপটপে আর সার্ভারে চলে।
• কাজ হয় একটা ফিচার ব্রাঞ্চে, ছোট ছোট কমিটে, রিভিউয়ের জন্য তৈরি অবস্থায়।

ডিজাইন

তিনটা ফাইল, প্রত্যেকের একটাই কাজ। রাউটগুলো জানে না HTTP কল কীভাবে হয়, আর আপস্ট্রিম ক্লায়েন্ট FastAPI সম্পর্কে কিছুই জানে না:

data-explorer/
├── pyproject.toml      ডিপেন্ডেন্সি আর FastAPI-র entrypoint (uv সামলায়)
├── uv.lock             প্রতিটা প্যাকেজের হুবহু ভার্সন
├── .gitignore
└── app/
    ├── __init__.py     app/-কে প্যাকেজ বানায়
    ├── schemas.py      Pydantic মডেল: আমরা যা পাঠাই, আপস্ট্রিম থেকে যা নিই
    ├── upstream.py     Practice API-এর প্রতিটা কল: টাইমআউট, এরর, পেজিনেশন, ভ্যালিডেশন
    └── main.py         FastAPI অ্যাপ: রাউট, প্যারামিটার, স্ট্যাটাস কোড

  ক্লায়েন্ট (ব্রাউজার, curl, AI অ্যাসিস্ট্যান্ট)
        │  GET /weather/sylhet
        ▼
  Data Explorer API  :8000   main.py ──► upstream.py
        │  GET /v1/weather?city=sylhet   (টাইমআউট 5 s)
        ▼
  BitByte Practice API  :8800

আপস্ট্রিমের প্রতিটা ধরনের সমস্যার জন্য আমাদের API-এর একটা নির্দিষ্ট উত্তর:

ধাপ ১: প্রজেক্ট আর রিপোজিটরি

নিশ্চিত হোন যে Practice API নিজের টার্মিনালে চলছে (fastapi run practice_api.py --port 8800, ওয়েব API ও REST: ডকুমেন্টেশন পড়া আর curl দিয়ে পরীক্ষা পাতা থেকে)। তারপর প্রজেক্টের ফোল্ডার, একটা Git রিপোজিটরি আর একটা uv প্রজেক্ট বানান:

mkdir -p ~/projects/data-explorer
cd ~/projects/data-explorer
git init
uv init --bare --name data-explorer
uv add -q "fastapi[standard]" requests
cat pyproject.toml

Initialized empty Git repository in /home/nadia/projects/data-explorer/.git/
Initialized project `data-explorer`
[project]
name = "data-explorer"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "fastapi[standard]>=0.143.0",
    "requests>=2.34.2",
]

uv init --bare শুধু pyproject.toml লেখে (নমুনা main.py বা README নয়), আর uv add .venv বানায়, প্যাকেজগুলো ইনস্টল করে আর হুবহু ভার্সনগুলো uv.lock-এ লিখে রাখে। আপনার requires-python লাইনে দেখাবে আপনার কাছে যে Python আছে সেটা (কোর্সের স্ট্যান্ডার্ড 3.14; 3.12 বা নতুন যেকোনোটা চলবে)। uv-র বদলে pip দিয়ে একই সেটআপ:

python -m venv .venv
source .venv/bin/activate          # Windows PowerShell: .venv\Scripts\Activate.ps1
pip install "fastapi[standard]" requests
pip freeze > requirements.txt

FastAPI কমান্ডকে বলে দিন অ্যাপটা কোথায় আছে, যাতে ফাইলের নাম ছাড়াই fastapi run লিখতে পারেন। FastAPI-র নিজের ডকুমেন্টেশন এখন এই সেটিংটাই সুপারিশ করে:

cat >> pyproject.toml <<'EOF'

[tool.fastapi]
entrypoint = "app.main:app"
EOF
mkdir app
touch app/__init__.py

প্রথম কমিটের আগে .gitignore (দেখুন .gitignore: সিক্রেট আর অপ্রয়োজনীয় ফাইল বাইরে রাখা)। ভার্চুয়াল এনভায়রনমেন্ট, ক্যাশ, লগ আর আসল .env ফাইল কখনো Git-এ যায় না:

# file: .gitignore
# Python
__pycache__/
*.pyc

# Virtual environment (rebuilt from uv.lock with `uv sync`)
.venv/

# Secrets and local settings
.env

# Logs from local runs
*.log

কাঠামোটা main-এ কমিট করুন, তারপর আসল কাজের জন্য একটা ফিচার ব্রাঞ্চ শুরু করুন:

git add .
git status --short
git commit -q -m "Set up the Data Explorer project with uv"
git switch -c feature/data-explorer

A  .gitignore
A  app/__init__.py
A  pyproject.toml
A  uv.lock
Switched to a new branch 'feature/data-explorer'

খেয়াল করুন git status --short কী দেখাল: .venv নেই, যেমনটা থাকার কথা। কিন্তু uv.lock কমিট হয়েছে: এটা দিয়েই একজন সহকর্মী, একটা CI সার্ভার বা একটা Docker বিল্ড হুবহু একই ভার্সনগুলো ইনস্টল করতে পারে (uv sync)।

ধাপ ২: আগে স্কিমা

লজিকের আগে ডেটার আকারগুলো লিখে ফেলুন। Course দুবার কাজে লাগে: Practice API যা পাঠায় তা যাচাই করতে, আর আমরা যা পাঠাই তার বর্ণনা দিতে। UpstreamWeather-এ Practice API-এর ফিল্ডের নাম; Weather-এ আমাদের নিজেদের।

# file: app/schemas.py
"""The shapes of the data this API sends out and accepts from upstream."""
from typing import Literal

from pydantic import BaseModel, Field

Level = Literal["beginner", "intermediate", "advanced"]


class Course(BaseModel):
    id: int = Field(ge=1)
    title: str = Field(min_length=1)
    level: Level
    hours: int = Field(ge=0)
    rating: float = Field(ge=0, le=5)


class CourseList(BaseModel):
    count: int
    items: list[Course]


class LevelSummary(BaseModel):
    level: Level
    courses: int
    total_hours: int
    average_rating: float


class UpstreamWeather(BaseModel):
    """What the Practice API sends for /v1/weather (its names, not ours)."""
    city: str
    temp_c: float
    humidity: int
    condition: str


class Weather(BaseModel):
    city: str
    temperature_c: float
    humidity_pct: int = Field(ge=0, le=100)
    condition: str
    advice: str

লেভেলের জন্য Literal কেন? এটা একটা বদ্ধ তালিকা: আপনার কোড চলার আগেই FastAPI ?level=expert-কে 422 দিয়ে ফিরিয়ে দেবে, অনুমোদিত মানগুলো /docs-এ দেখা যাবে, আর আপস্ট্রিম থেকে অচেনা লেভেলের কোর্স এলে সেটা চুপচাপ ঢুকে না পড়ে ভ্যালিডেশনে আটকে যাবে। একটা ভালো আর একটা খারাপ রেকর্ড দিয়ে মডেলটা পরীক্ষা করুন:

from pydantic import ValidationError
from app.schemas import Course

print(Course.model_validate({"id": 8, "title": "RAG Systems", "level": "advanced", "hours": 10, "rating": 4.9}))
try:
    Course.model_validate({"id": 8, "title": "RAG Systems", "level": "expert", "hours": "ten", "rating": 4.9})
except ValidationError as exc:
    print(exc.error_count(), "errors:", [e["loc"][0] for e in exc.errors()])

id=8 title='RAG Systems' level='advanced' hours=10 rating=4.9
2 errors: ['level', 'hours']

git add app/schemas.py
git commit -q -m "Add response and upstream schemas"

ধাপ ৩: আপস্ট্রিম ক্লায়েন্ট

Practice API যে আছে, সেটা শুধু এই ফাইলটাই জানে। সেটা অন্য ঠিকানায় সরে গেলে, ফিল্ডের নাম বদলালে বা কোনো আসল ডেটা প্রোভাইডার তার জায়গা নিলে, বদলাতে হবে শুধু এই একটা ফাইল।

# file: app/upstream.py
"""Everything that talks to the BitByte Practice API lives in this one file."""
import os

import requests
from pydantic import BaseModel, ValidationError

from app.schemas import Course, UpstreamWeather

BASE_URL = os.environ.get("PRACTICE_API_URL", "http://localhost:8800")
TIMEOUT_SECONDS = float(os.environ.get("UPSTREAM_TIMEOUT", "5"))
MAX_PAGES = 50

session = requests.Session()
session.headers["User-Agent"] = "bitbyte-data-explorer/0.1"


class UpstreamError(Exception):
    """The Practice API was unreachable, too slow, or sent something we cannot use."""

    def __init__(self, message: str, status_code: int = 502):
        super().__init__(message)
        self.message = message
        self.status_code = status_code


def get_json(path: str, params: dict | None = None) -> dict | None:
    """GET a Practice API path and return the JSON body, or None for a 404."""
    try:
        response = session.get(f"{BASE_URL}{path}", params=params, timeout=TIMEOUT_SECONDS)
    except requests.Timeout:
        raise UpstreamError(f"Practice API did not answer within {TIMEOUT_SECONDS:g} s", 504)
    except requests.ConnectionError:
        raise UpstreamError("Practice API is unreachable")
    if response.status_code == 404:
        return None
    if not response.ok:
        raise UpstreamError(f"Practice API failed with status {response.status_code}")
    try:
        return response.json()
    except ValueError:
        raise UpstreamError("Practice API sent a body that is not JSON")


def parse(model: type[BaseModel], data: dict):
    """Validate upstream data before we trust it."""
    try:
        return model.model_validate(data)
    except ValidationError:
        raise UpstreamError(f"Practice API sent {model.__name__} data in an unexpected shape")


def fetch_courses(level: str | None = None) -> list[Course]:
    """All courses (optionally of one level), following next_page to the end."""
    courses, page = [], 1
    while page is not None:
        if page > MAX_PAGES:
            raise UpstreamError("Practice API returned too many pages")
        data = get_json("/v1/courses", {"level": level, "page": page, "per_page": 20})
        if data is None:
            raise UpstreamError("Practice API has no /v1/courses endpoint")
        courses += [parse(Course, item) for item in data["items"]]
        page = data["next_page"]
    return courses


def fetch_course(course_id: int) -> Course | None:
    data = get_json(f"/v1/courses/{course_id}")
    return None if data is None else parse(Course, data)


def fetch_weather(city: str) -> UpstreamWeather | None:
    data = get_json("/v1/weather", {"city": city})
    return None if data is None else parse(UpstreamWeather, data)

এই ফাইলের সিদ্ধান্তগুলো, একে একে:

• সেটিং আসে এনভায়রনমেন্ট থেকে, যুক্তিসঙ্গত ডিফল্টসহ। সার্ভারে আপস্ট্রিম থাকবে অন্য ঠিকানায়; সেদিকে নির্দেশ করতে কারও কোড এডিট করা উচিত নয়।
• পুরো অ্যাপের জন্য একটাই requests.Session: এটা কানেকশন আবার ব্যবহার করে (দ্রুততর), আর প্রতিটা কলে আমাদের User-Agent পাঠায়, যাতে আপস্ট্রিমের পরিচালকেরা দেখতে পান কে কল করছে (দেখুন নির্ভরযোগ্য API কল: টাইমআউট, রিট্রাই, পেজিনেশন ও LLM API)।
• প্রতিটা রিকোয়েস্টে timeout=। এটা ছাড়া requests অনন্তকাল অপেক্ষা করে, আর প্রতিটা আটকে থাকা রিকোয়েস্ট আপনার সার্ভারের একটা করে ওয়ার্কার থ্রেড দখল করে রাখে।
• একটাই এক্সেপশন টাইপ, UpstreamError, যার ভেতরে থাকে আমাদের API কোন স্ট্যাটাস কোড দিয়ে উত্তর দেবে। যারা কল করে তারা পাঁচটা requests এক্সেপশন নয়, একটাই জিনিস ধরে।
• এখানে 404 কোনো এরর নয়। "এমন কোর্স নেই" একটা স্বাভাবিক উত্তর, তাই get_json ফেরত দেয় None, আর ক্লায়েন্টকে কী বলা হবে তা ঠিক করে রাউট।
• নিরাপত্তা-সীমাসহ পেজিনেশন। লুপটা next_page ধরে এগোয় যতক্ষণ না সেটা null হয়; MAX_PAGES নিশ্চিত করে যে বাগওয়ালা কোনো আপস্ট্রিম আপনাকে চিরকাল লুপে আটকে রাখতে পারবে না।
• parse ভেতরে আসা সবকিছু যাচাই করে। খেয়াল করুন, এরর মেসেজে মডেলের নাম আছে, কিন্তু আপস্ট্রিমের ডেটা নেই: এরর মেসেজ শেষ পর্যন্ত লগে যায়, ব্যবহারকারীর চোখের সামনেও আসে।

এনভায়রনমেন্ট ভেরিয়েবল দিয়ে অ্যাপটাকে Practice API-এর দিকে নির্দেশ করুন (ঠিকানাটা ডিফল্টের মতোই, কিন্তু এখন দেখতে পাচ্ছেন সেটা কোথা থেকে আসে), আর কোনো রাউট বানানোর আগেই Python থেকে ক্লায়েন্টটা চালিয়ে দেখুন:

export PRACTICE_API_URL=http://localhost:8800
export UPSTREAM_TIMEOUT=5

from app import upstream

courses = upstream.fetch_courses()
print(len(courses), "courses; first:", courses[0])
print(upstream.fetch_course(99))
print(upstream.fetch_weather("Sylhet"))

10 courses; first: id=1 title='How AI Works' level='beginner' hours=3 rating=4.7
None
city='Sylhet' temp_c=27.4 humidity=85 condition='Rain'

Practice API এক পাতায় সর্বোচ্চ ২০টা আর ডিফল্টে ৫টা কোর্স পাঠায়, তবু ১০টা কোর্সই এসেছে: fetch_courses প্রতি পাতায় ২০টা চেয়েছে আর next_page ধরে এগিয়েছে। কোর্স ৯৯ এসেছে None হিসেবে, এক্সেপশন হিসেবে নয়।

এবার ব্যর্থতার পথগুলো। Practice API-এর /v1/slow এন্ডপয়েন্ট আপনি যতক্ষণ বলবেন ততক্ষণ অপেক্ষা করে; টাইমআউট এক সেকেন্ডে নামিয়ে আনুন আর তাকে দুই সেকেন্ড অপেক্ষা করতে বলুন:

upstream.TIMEOUT_SECONDS = 1
try:
    upstream.get_json("/v1/slow", {"seconds": 2})
except upstream.UpstreamError as exc:
    print(exc.status_code, exc.message)
upstream.TIMEOUT_SECONDS = 5

try:
    upstream.parse(upstream.Course, {"id": 1, "title": "How AI Works", "level": "beginner"})
except upstream.UpstreamError as exc:
    print(exc.status_code, exc.message)

504 Practice API did not answer within 1 s
502 Practice API sent Course data in an unexpected shape

এক সেকেন্ড পর ক্লায়েন্ট হাল ছেড়ে দিয়েছে আর টাইমআউটটাকে স্পষ্ট মেসেজসহ একটা 504 এররে বদলে দিয়েছে। hours আর rating ছাড়া একটা কোর্স আধখালি অবস্থায় সামনে না গিয়ে প্রত্যাখ্যাত হয়েছে। ক্লায়েন্টটা কমিট করুন:

git add app/upstream.py
git commit -q -m "Add Practice API client with timeouts, pagination and validation"

ধাপ ৪: এন্ডপয়েন্টগুলো

# file: app/main.py
"""BitByte Data Explorer API: clean, validated endpoints over the BitByte Practice API."""
from typing import Annotated, Literal

from fastapi import FastAPI, HTTPException, Path, Query, Request
from fastapi.responses import JSONResponse

from app import upstream
from app.schemas import Course, CourseList, Level, LevelSummary, Weather

app = FastAPI(title="BitByte Data Explorer API", version="0.1.0")


@app.exception_handler(upstream.UpstreamError)
def upstream_failed(request: Request, exc: upstream.UpstreamError) -> JSONResponse:
    return JSONResponse({"detail": exc.message}, status_code=exc.status_code)


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/courses")
def list_courses(
    level: Level | None = None,
    min_rating: Annotated[float, Query(ge=0, le=5)] = 0,
    sort: Literal["id", "rating", "hours"] = "id",
) -> CourseList:
    courses = [c for c in upstream.fetch_courses(level) if c.rating >= min_rating]
    courses.sort(key=lambda c: getattr(c, sort), reverse=sort != "id")
    return CourseList(count=len(courses), items=courses)


@app.get("/courses/{course_id}")
def get_course(course_id: Annotated[int, Path(ge=1)]) -> Course:
    course = upstream.fetch_course(course_id)
    if course is None:
        raise HTTPException(404, f"Course {course_id} not found")
    return course


@app.get("/summary")
def summary() -> list[LevelSummary]:
    by_level: dict[str, list[Course]] = {}
    for course in upstream.fetch_courses():
        by_level.setdefault(course.level, []).append(course)
    return [
        LevelSummary(
            level=level,
            courses=len(group),
            total_hours=sum(c.hours for c in group),
            average_rating=round(sum(c.rating for c in group) / len(group), 2),
        )
        for level, group in by_level.items()
    ]


def advice_for(temp_c: float, condition: str) -> str:
    if condition == "Rain":
        return "Carry an umbrella."
    if temp_c >= 30:
        return "Hot day: drink water and avoid the midday sun."
    return "Good weather to go out."


@app.get("/weather/{city}")
def weather(city: Annotated[str, Path(min_length=2, max_length=40)]) -> Weather:
    data = upstream.fetch_weather(city)
    if data is None:
        raise HTTPException(404, f"No weather data for {city}")
    return Weather(
        city=data.city,
        temperature_c=data.temp_c,
        humidity_pct=data.humidity,
        condition=data.condition,
        advice=advice_for(data.temp_c, data.condition),
    )

যা খেয়াল করবেন:

• @app.exception_handler(upstream.UpstreamError) ক্লায়েন্টের এররটা যেখানেই রেইজ হোক ধরে ফেলে, আর সঠিক স্ট্যাটাসসহ একটা JSON রেসপন্সে বদলে দেয়। এটা না থাকলে একটা UpstreamError হয়ে যেত খালি একটা 500 "Internal Server Error"। একটা হ্যান্ডলারই সব রাউট সামলায়, এখনকার আর পরের সবগুলো।
• রাউটগুলো upstream.fetch_courses(...) কল করে মডিউলের মাধ্যমে, from app.upstream import fetch_courses দিয়ে নয়। এই ছোট অভ্যাসটার জন্যই পরের পাতার টেস্টগুলো ফাংশনটাকে একটা নকল (fake) ফাংশন দিয়ে বদলে দিতে পারবে।
• প্যারামিটারগুলো টাইপ করা আর সীমা বাঁধা Annotated, Query আর Path দিয়ে (দেখুন Pydantic মডেল, ভ্যালিডেশন ও এরর রেসপন্স): min_rating হতে হবে ০–৫, course_id অন্তত ১, city ২–৪০ অক্ষর, sort তিনটা শব্দের একটা। ভুল ইনপুট কখনো আপস্ট্রিম পর্যন্ত পৌঁছায় না।
• রিটার্ন টাইপগুলোই রেসপন্স মডেল। FastAPI ফলাফল যাচাই ও সিরিয়ালাইজ করে আর /docs-এ ডকুমেন্ট করে; response_model= লাগে না।
• নতুন করে সাজানোটা আমাদের: temp_c হয় temperature_c, humidity হয় humidity_pct, আর advice নতুন যোগ হয়। আপস্ট্রিম কোনো ফিল্ডের নাম বদলালে বদলাবে শুধু UpstreamWeather আর এই কয়েকটা লাইন; আমাদের ক্লায়েন্টরা কিছুই টের পাবে না।
• সাধারণ def রাউট, কারণ requests ব্লক করে। FastAPI def রাউটগুলো একটা থ্রেড পুলে চালায়, তাই একটা ধীর আপস্ট্রিম কল পুরো সার্ভার থামিয়ে দেয় না (ব্যাকএন্ড কী করে: রিকোয়েস্ট, রাউট ও রেসপন্স পাতার sync বনাম async অংশ)।

git add app/main.py
git commit -q -m "Add courses, summary and weather endpoints"

ধাপ ৫: চালিয়ে ঘুরে দেখুন

সার্ভারটা ব্যাকগ্রাউন্ডে চালু করুন আর সাড়া দেওয়া পর্যন্ত অপেক্ষা করুন (নিজের টার্মিনালে আপনি দ্বিতীয় একটা উইন্ডোতে সহজভাবে fastapi dev চালাতে পারেন, আর Ctrl+C দিয়ে থামাতে পারেন):

fastapi run --port 8000 > server.log 2>&1 &
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health

{"status":"ok"}

অ্যাডভান্সড কোর্সগুলো, সবচেয়ে ভালো রেটিং আগে:

curl -s "http://localhost:8000/courses?level=advanced&sort=rating" | python -m json.tool

{
    "count": 3,
    "items": [
        {
            "id": 8,
            "title": "RAG Systems",
            "level": "advanced",
            "hours": 10,
            "rating": 4.9
        },
        {
            "id": 9,
            "title": "AI Agents",
            "level": "advanced",
            "hours": 12,
            "rating": 4.8
        },
        {
            "id": 10,
            "title": "LLMOps",
            "level": "advanced",
            "hours": 8,
            "rating": 4.3
        }
    ]
}

৪.৭ বা তার বেশি রেটিংয়ের কোর্স, লেভেলভিত্তিক সারাংশ, আর দুটো শহরের আবহাওয়া:

curl -s "http://localhost:8000/courses?min_rating=4.7" | python -c "import json, sys; d = json.load(sys.stdin); print(d['count'], [c['title'] for c in d['items']])"
curl -s http://localhost:8000/summary; echo
curl -s http://localhost:8000/weather/Sylhet; echo
curl -s http://localhost:8000/weather/rajshahi; echo

5 ['How AI Works', 'Python for AI', 'LLM APIs', 'RAG Systems', 'AI Agents']
[{"level":"beginner","courses":5,"total_hours":42,"average_rating":4.64},{"level":"intermediate","courses":2,"total_hours":11,"average_rating":4.55},{"level":"advanced","courses":3,"total_hours":30,"average_rating":4.67}]
{"city":"Sylhet","temperature_c":27.4,"humidity_pct":85,"condition":"Rain","advice":"Carry an umbrella."}
{"city":"Rajshahi","temperature_c":33.6,"humidity_pct":58,"condition":"Sunny","advice":"Hot day: drink water and avoid the midday sun."}

সারাংশটা হিসাব করেছে আমাদের API, আপস্ট্রিম নয়: ৫টা বিগিনার কোর্স, ৩ + ১২ + ৮ + ১০ + ৯ = ৪২ ঘণ্টা। ব্রিফের "নতুন করে সাজানো" অংশ এটাই: ক্লায়েন্ট একটা প্রশ্ন করে একটা ছোট উত্তর পায়, সব কোর্স ডাউনলোড করে নিজে অঙ্ক কষতে হয় না। AI অ্যাসিস্ট্যান্টের জন্য এর গুরুত্ব দ্বিগুণ: তার কনটেক্সটে কম টোকেন যায়, আর সংখ্যা ভুল যোগ করার সুযোগও কমে।

ব্রাউজারে http://localhost:8000/docs খুলুন: প্রতিটা এন্ডপয়েন্ট তার প্যারামিটার, level আর sort-এর অনুমোদিত মান আর রেসপন্সের আকারসহ সেখানে আছে, সবই আপনার টাইপ হিন্ট থেকে তৈরি। একই তথ্য JSON হিসেবে:

curl -s http://localhost:8000/openapi.json | python -c "import json, sys; print(list(json.load(sys.stdin)['paths']))"

['/health', '/courses', '/courses/{course_id}', '/summary', '/weather/{city}']

ধাপ ৬: ইচ্ছে করে ভাঙুন

যে এরর-পথ কখনো কাজ করতে দেখেননি, সেটা নিছক অনুমান। প্রতিটা পরীক্ষা করুন। প্রথমে ক্লায়েন্টের ভুল আর অনুপস্থিত ডেটা:

curl -s -w '  [%{http_code}]\n' http://localhost:8000/courses/99
curl -s -w '  [%{http_code}]\n' http://localhost:8000/weather/Comilla
curl -s -w '  [%{http_code}]\n' http://localhost:8000/courses/0
curl -s -w '  [%{http_code}]\n' "http://localhost:8000/courses?level=expert"

{"detail":"Course 99 not found"}  [404]
{"detail":"No weather data for Comilla"}  [404]
{"detail":[{"type":"greater_than_equal","loc":["path","course_id"],"msg":"Input should be greater than or equal to 1","input":"0","ctx":{"ge":1}}]}  [422]
{"detail":[{"type":"literal_error","loc":["query","level"],"msg":"Input should be 'beginner', 'intermediate' or 'advanced'","input":"expert","ctx":{"expected":"'beginner', 'intermediate' or 'advanced'"}}]}  [422]

-w ' [%{http_code}]\n' দিলে curl বডির পরে স্ট্যাটাস কোডটা প্রিন্ট করে। 404-গুলোতে আছে আমাদের নিজেদের ছোট মেসেজ; 422-গুলো আসে FastAPI-র ভ্যালিডেশন থেকে, আর ঠিক কোন প্যারামিটার ভুল (loc) আর কোন মানগুলো চলে, তা দেখিয়ে দেয়। এগুলোর কোনোটার জন্যই Practice API-কে কল করা হয়নি।

এবার এমন একটা আপস্ট্রিম, যেটা বন্ধ। অ্যাপের দ্বিতীয় একটা কপি পোর্ট 8001-এ চালু করুন, যার PRACTICE_API_URL এমন একটা পোর্টের দিকে, যেখানে কেউ শুনছে না:

PRACTICE_API_URL=http://127.0.0.1:9 fastapi run --port 8001 > broken.log 2>&1 &
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8001/health; echo
curl -s -w '  [%{http_code}]\n' http://localhost:8001/courses
curl -s -w '  [%{http_code}]\n' http://localhost:8001/weather/dhaka

{"status":"ok"}
{"detail":"Practice API is unreachable"}  [502]
{"detail":"Practice API is unreachable"}  [502]

/health এখনো ok বলছে, কারণ এটা আপস্ট্রিমের ওপর নির্ভর করে না। লাইভনেস চেকের ("প্রসেসটা কি বেঁচে আছে?") জন্য এটাই ঠিক; অ্যাডভান্সড টিউটোরিয়ালে আলাদা একটা রেডিনেস চেক যোগ হবে, যা ডিপেন্ডেন্সিগুলোও পরীক্ষা করে। ডেটার এন্ডপয়েন্টগুলো একটা পরিষ্কার বাক্যসহ 502 দিচ্ছে, আর আমাদের সার্ভার চালু আছে। ভাঙা কপিটার লগ দেখুন:

grep -oE '"GET .*' broken.log

"GET /health HTTP/1.1" 200 OK
"GET /courses HTTP/1.1" 502 Bad Gateway
"GET /weather/dhaka HTTP/1.1" 502 Bad Gateway

কোনো ট্রেসব্যাক নেই, শুধু তিনটা রিকোয়েস্ট আর তাদের স্ট্যাটাস কোড। যিনি লগ পড়ছেন বা যে মনিটরিং অ্যালার্ট বাজছে, 502 তাঁকে বলে দেয়: "আমাদের কোড ঠিক আছে, পেছনের জিনিসটা ঠিক নেই।"

Git-এ আপনি এখন কোথায়

git status --short
git log --oneline --graph --decorate --all

* 3a8f75a (HEAD -> feature/data-explorer) Add courses, summary and weather endpoints
* 089f152 Add Practice API client with timeouts, pagination and validation
* 6bba991 Add response and upstream schemas
* 04b26f9 (main) Set up the Data Explorer project with uv

git status --short কিছুই প্রিন্ট করেনি: ওয়ার্কিং ট্রি পরিষ্কার, আর server.log ও broken.log বাইরে আছে কারণ *.log ইগনোর করা। main-এ এখনো শুধু কাঠামোটা; ফিচার ব্রাঞ্চে আছে তিনটা কমিট, প্রত্যেকটা আলাদাভাবে রিভিউ করার মতো ছোট, আর প্রত্যেকটার মেসেজ বলে দেয় সেটা কী যোগ করছে। পরের পাতার পুল রিকোয়েস্ট একজন রিভিউয়ারকে ঠিক এটাই দেখাবে। আপনার হ্যাশগুলো ওপরেরগুলোর থেকে আলাদা হবে: কমিটের হ্যাশে কমিট করার সময়টাও ঢোকে।

বাস্তবে কোথায় দেখবেন

• LLM ব্যাকএন্ড: Practice API-এর জায়গায় একটা LLM প্রোভাইডার বসান, তাহলেই এটা প্রতিটা চ্যাট ব্যাকএন্ডের কাঠামো: একটা ক্লায়েন্ট মডিউল, প্রতিটা কলে টাইমআউট, প্রোভাইডারের এরর 502/504-এ বদলানো, আর ব্যবহারকারীর কাছে পৌঁছানোর আগে রেসপন্স যাচাই।
• AI এজেন্টের টুল: এজেন্টরা (আর MCP সার্ভার) /summary বা /weather/{city}-এর মতো ছোট, টাইপ করা এন্ডপয়েন্ট কল করে। পরিষ্কার এররসহ ছোট, অনুমানযোগ্য উত্তরই একটা এজেন্টকে নির্ভরযোগ্য করে।
• "Backend for frontend" সার্ভিস: একটা মোবাইল অ্যাপ বা ওয়েবসাইট নিজে পাঁচটা আপস্ট্রিম API কল করে না বললেই চলে; একটা ব্যাকএন্ড সেগুলো কল করে, ডেটা জোড়া দেয় ও সাজায়, আর তাদের কি (key) লুকিয়ে রাখে।
• ডেটা পাইপলাইন: একই ক্লায়েন্ট কোড (সেশন, টাইমআউট, পেজিনেশন, ভ্যালিডেশন) দিয়েই ETL জব SaaS API থেকে ডেটা টেনে ডেটাবেস বা ট্রেনিং সেটে নেয়।

সাধারণ ভুল

• আপস্ট্রিম কলে টাইমআউট না দেওয়া। একটা আটকে যাওয়া আপস্ট্রিম প্রতি রিকোয়েস্টে একটা করে ওয়ার্কার দখল করে, যতক্ষণ না পুরো সার্ভার সাড়া দেওয়া বন্ধ করে। সমাধান: প্রতিটা কলে একটা সেটিং থেকে নেওয়া timeout=।
• আপস্ট্রিমের এরর সরাসরি সামনে পাঠানো। আপস্ট্রিমের 500 ফেরত দিলে (বা requests-এর এক্সেপশনকে নিজের 500 হতে দিলে) ক্লায়েন্ট কিছুই জানতে পারে না, আর দোষটা কার তাও চাপা পড়ে। সমাধান: নিজের মেসেজসহ 502/504-এ বদলান, আর যেখানে "পাওয়া যায়নি" স্বাভাবিক উত্তর সেখানে 404।
• আপস্ট্রিমের ডেটায় অন্ধ বিশ্বাস। নাম বদলানো বা অনুপস্থিত একটা ফিল্ড রাউটের গভীরে KeyError হয়ে যায়, বা উত্তরে ভুল সংখ্যা বসায়। সমাধান: সীমানাতেই Pydantic মডেল দিয়ে যাচাই করুন, আর পরিষ্কারভাবে ব্যর্থ হোন।
• শুধু প্রথম পাতা পড়া। Practice API ডিফল্টে ৫টা কোর্স পাঠায়; next_page উপেক্ষা করা কোড চুপচাপ অর্ধেক ডেটার রিপোর্ট দেয়। সমাধান: পাতার সীমাসহ লুপ চালান, যতক্ষণ না next_page null হয়।
• সরাসরি main-এ একটা বিশাল কমিটে কাজ করা। কেউ সেটা রিভিউ করতে পারে না, আর টুকরো টুকরো করে কিছু ফেরানোও যায় না। সমাধান: একটা ফিচার ব্রাঞ্চ, আর প্রতিটা ধাপে একটা কমিট।

নিজে চেষ্টা করুন

উত্তরগুলো চলমান app-এ নতুন রাউট যোগ করে আর FastAPI-র TestClient দিয়ে সেগুলো কল করে (pytest দিয়ে আপনার API টেস্ট থেকে), তাই সার্ভার রিস্টার্ট করতে হয় না। নিজের প্রজেক্টে একই ফাংশনগুলো app/main.py-তে বসান।

1. সহজ: GET /levels যোগ করুন, যা আপস্ট্রিমের কোর্সগুলো থেকে হিসাব করে বিদ্যমান লেভেলগুলোর সাজানো তালিকা ফেরত দেয়।
2. মাঝারি: GET /weather (শহর ছাড়া) যোগ করুন, যা Dhaka, Chattogram, Sylhet, Khulna আর Rajshahi-র আবহাওয়া ফেরত দেয়, সবচেয়ে গরমটা আগে। আগের weather() ফাংশনটাই আবার ব্যবহার করুন।
3. কঠিন: আপস্ট্রিম মাঝে মাঝে কিছুক্ষণের জন্য 503 দেয়। upstream.session-এ urllib3-এর Retry-সহ একটা HTTPAdapter বসান, যা 429, 502, 503 আর 504-এ ছোট ব্যাকঅফ দিয়ে GET রিকোয়েস্ট সর্বোচ্চ ৩ বার আবার চেষ্টা করে; তারপর Practice API-এর /v1/flaky?key=p20&fail_times=2 দিয়ে upstream.get_json-এর মাধ্যমে সেটা প্রমাণ করুন। ডিফল্ট হিসেবে শুধু GET রিট্রাই করা কেন যুক্তিসঙ্গত?

উত্তর

# 1. GET /levels
from fastapi.testclient import TestClient
from app.main import app, weather

@app.get("/levels")
def levels() -> list[str]:
    return sorted({course.level for course in upstream.fetch_courses()})

client = TestClient(app)
print(client.get("/levels").json())

['advanced', 'beginner', 'intermediate']

# 2. GET /weather for all five cities, hottest first
from app.schemas import Weather

CITIES = ["Dhaka", "Chattogram", "Sylhet", "Khulna", "Rajshahi"]

@app.get("/weather")
def all_weather() -> list[Weather]:
    reports = [weather(city) for city in CITIES]
    return sorted(reports, key=lambda w: w.temperature_c, reverse=True)

for w in client.get("/weather").json():
    print(f"{w['city']:11} {w['temperature_c']:5} {w['advice']}")

Rajshahi     33.6 Hot day: drink water and avoid the midday sun.
Khulna       32.1 Hot day: drink water and avoid the midday sun.
Dhaka        31.5 Hot day: drink water and avoid the midday sun.
Chattogram   29.8 Carry an umbrella.
Sylhet       27.4 Carry an umbrella.

# 3. Retries for temporary upstream failures
from requests.adapters import HTTPAdapter
from urllib3.util import Retry

retry = Retry(total=3, backoff_factor=0.2, status_forcelist=[429, 502, 503, 504],
              allowed_methods=["GET"])
upstream.session.mount("http://", HTTPAdapter(max_retries=retry))
upstream.session.mount("https://", HTTPAdapter(max_retries=retry))

print(upstream.get_json("/v1/flaky", {"key": "p20", "fail_times": 2}))

{'ok': True, 'attempt': 3}

প্রথম দুটো কলে এসেছে 503, অ্যাডাপ্টার অপেক্ষা করে আবার চেষ্টা করেছে, আর তৃতীয় চেষ্টায় সফল হয়েছে; আপনার কোড দেখেছে শুধু সফলতাটা। GET বারবার করা নিরাপদ, কারণ এটা কিছু বদলায় না। কিছু তৈরি করে এমন POST (একটা রিভিউ, একটা পেমেন্ট, টাকা লাগে এমন একটা LLM কল) রিট্রাই করলে কাজটা দুবার হয়ে যেতে পারে, তাই সেগুলোর আগে আইডেমপোটেন্সি কি (idempotency key) লাগে, যা অ্যাডভান্সড টিউটোরিয়ালে বানানো হবে। মোট সময়ের কথাও মাথায় রাখুন: রিট্রাই সবচেয়ে খারাপ ক্ষেত্রের অপেক্ষাকে কয়েক গুণ করে দেয়, তাই টাইমআউট আর রিট্রাইয়ের সংখ্যা মিলে যেন আপনার নিজের ক্লায়েন্টের ধৈর্যের সীমার মধ্যে থাকে।

সারসংক্ষেপ

• প্রজেক্ট শুরু করুন চুক্তি লিখে: এন্ডপয়েন্ট, রেসপন্সের আকার, আর প্রতিটা ধরনের এররে কী হবে।
• আপস্ট্রিমের প্রতিটা কল একটা মডিউলে রাখুন: একটা শেয়ার করা সেশন, প্রতিটা রিকোয়েস্টে টাইমআউট, সীমাসহ পেজিনেশন, আর ফেরত আসা ডেটার Pydantic ভ্যালিডেশন।
• আপস্ট্রিমের গোলমাল একটা এক্সেপশন টাইপে, আর একটা এক্সেপশন হ্যান্ডলার দিয়ে ছোট মেসেজসহ 502 বা 504-এ বদলান; "পাওয়া যায়নি"-কে স্বাভাবিক উত্তর ধরে 404 দিন।
• প্রতিটা প্যারামিটার Annotated, Query, Path আর Literal দিয়ে টাইপ ও সীমাবদ্ধ করুন, যাতে ভুল ইনপুট আপস্ট্রিম কলের খরচ হওয়ার আগেই 422 পায়।
• ফিচার ব্রাঞ্চে ছোট ছোট কমিটে বানান, আর ইচ্ছে করে ভেঙে এরর-পথগুলো পরীক্ষা করুন।

এরপর: প্রজেক্ট, পর্ব ২: টেস্ট, ডকুমেন্টেশন ও প্রকাশ পাতায় যোগ হবে স্বয়ংক্রিয় টেস্ট, যেখানে আপস্ট্রিমের জায়গায় থাকবে নকল (fake) ফাংশন; সাথে একটা README, একটা রিভিউ, একটা রিফ্যাক্টর, আর পুল রিকোয়েস্ট ওয়ার্কফ্লোর মাধ্যমে main-এ মার্জ।