নির্ভরযোগ্য API কল: টাইমআউট, রিট্রাই, পেজিনেশন ও LLM API

mkdir -p ~/projects/reliable-client
cd ~/projects/reliable-client
python -m venv .venv
source .venv/bin/activate
pip install -q requests
export PRACTICE_BASE_URL="http://localhost:8800"
export PRACTICE_TOKEN="token-nadia-001"

import json
import os
import random
import time

import requests

BASE = os.environ["PRACTICE_BASE_URL"]
TIMEOUT = (3.05, 10)

for attempt in range(1, 5):
    r = requests.get(f"{BASE}/v1/flaky", params={"key": "demo-1", "fail_times": 2}, timeout=TIMEOUT)
    print(attempt, r.status_code, r.json())
    if r.ok:
        break

1 503 {'detail': 'Service temporarily unavailable'}
2 503 {'detail': 'Service temporarily unavailable'}
3 200 {'ok': True, 'attempt': 3}

def backoff_delay(attempt, base=0.1, cap=5.0):
    """Seconds to wait after the given failed attempt (1, 2, 3 ...)."""
    return min(cap, base * 2 ** (attempt - 1))

print([backoff_delay(n) for n in range(1, 8)])

[0.1, 0.2, 0.4, 0.8, 1.6, 3.2, 5.0]

RETRY_STATUSES = {429, 500, 502, 503, 504}


def request_with_retries(method, url, *, session=None, max_attempts=4, base=0.1, cap=5.0,
                         jitter=True, on_retry=None, **kwargs):
    """Send a request, retrying transient failures. Returns the last response."""
    caller = session or requests
    kwargs.setdefault("timeout", TIMEOUT)
    for attempt in range(1, max_attempts + 1):
        last = attempt == max_attempts
        retry_after = None
        try:
            response = caller.request(method, url, **kwargs)
        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as error:
            if last:
                raise
            reason = type(error).__name__
        else:
            if response.status_code not in RETRY_STATUSES or last:
                return response
            reason = str(response.status_code)
            header = response.headers.get("Retry-After", "")
            retry_after = int(header) if header.isdigit() else None

        delay = backoff_delay(attempt, base, cap)
        if jitter:
            delay = random.uniform(0, delay)
        if retry_after is not None:
            delay = max(delay, min(retry_after, 30))   # সার্ভারের কথা মানুন, তবে অনন্তকাল নয়
        if on_retry:
            on_retry(attempt, reason, delay)
        time.sleep(delay)

log = []
r = request_with_retries(
    "GET", f"{BASE}/v1/flaky", params={"key": "demo-2", "fail_times": 2},
    jitter=False, on_retry=lambda n, why, delay: log.append((n, why, delay)),
)
print(r.status_code, r.json())
print(log)

200 {'ok': True, 'attempt': 3}
[(1, '503', 0.1), (2, '503', 0.2)]

r = request_with_retries("GET", f"{BASE}/v1/flaky", params={"key": "demo-3", "fail_times": 10},
                         max_attempts=3, jitter=False)
print("last status:", r.status_code)
try:
    r.raise_for_status()
except requests.exceptions.HTTPError as error:
    print("gave up:", error.response.status_code)

try:
    request_with_retries("GET", "http://localhost:8001/health", max_attempts=2, base=0.05)
except requests.exceptions.ConnectionError:
    print("gave up: ConnectionError")

last status: 503
gave up: 503
gave up: ConnectionError

headers = {"X-Client-Id": "nadia-demo"}
for i in range(1, 5):
    r = requests.get(f"{BASE}/v1/limited", headers=headers, timeout=TIMEOUT)
    print(i, r.status_code, r.json())

wait = int(r.headers["Retry-After"])
print("Retry-After is a whole number of seconds up to 10:", 1 <= wait <= 10)

1 200 {'ok': True, 'remaining': 2}
2 200 {'ok': True, 'remaining': 1}
3 200 {'ok': True, 'remaining': 0}
4 429 {'detail': 'Too many requests'}
Retry-After is a whole number of seconds up to 10: True

headers = {"X-Client-Id": "nadia-retry"}
for _ in range(3):
    requests.get(f"{BASE}/v1/limited", headers=headers, timeout=TIMEOUT)

waits = []
r = request_with_retries("GET", f"{BASE}/v1/limited", headers=headers,
                         on_retry=lambda n, why, delay: waits.append((why, delay >= 1)))
print(r.status_code, r.json())
print(waits)

200 {'ok': True, 'remaining': 2}
[('429', True)]

session = requests.Session()
session.headers.update({"X-Client-Id": "nadia-session", "Accept": "application/json"})
session.headers["Authorization"] = f"Bearer {os.environ['PRACTICE_TOKEN']}"

echoed = session.get(f"{BASE}/v1/echo", timeout=TIMEOUT).json()["headers"]
print(echoed["accept"], "authorization" in echoed)

session.close()

application/json True

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

retry = Retry(total=3, backoff_factor=0.05, status_forcelist=[429, 500, 502, 503, 504],
              allowed_methods=["GET"], respect_retry_after_header=True)
auto = requests.Session()
auto.mount("http://", HTTPAdapter(max_retries=retry))
auto.mount("https://", HTTPAdapter(max_retries=retry))

r = auto.get(f"{BASE}/v1/flaky", params={"key": "demo-4", "fail_times": 2}, timeout=TIMEOUT)
print(r.status_code, r.json())

try:
    auto.get(f"{BASE}/v1/flaky", params={"key": "demo-5", "fail_times": 10}, timeout=TIMEOUT)
except requests.exceptions.RetryError:
    print("RetryError: out of retries")

200 {'ok': True, 'attempt': 3}
RetryError: out of retries

import csv

courses = []
page = 1
while page:
    r = request_with_retries("GET", f"{BASE}/v1/courses", params={"page": page, "per_page": 4})
    r.raise_for_status()
    data = r.json()
    courses.extend(data["items"])
    page = data["next_page"]
print(len(courses), "courses fetched")

with open("courses.json", "w", encoding="utf-8") as f:
    json.dump(courses, f, indent=2, ensure_ascii=False)

with open("courses.csv", "w", encoding="utf-8", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["id", "title", "level", "hours", "rating"], lineterminator="\n")
    writer.writeheader()
    writer.writerows(courses)

10 courses fetched

head -3 courses.csv
wc -l courses.csv

id,title,level,hours,rating
1,How AI Works,beginner,3,4.7
2,Python for AI,beginner,12,4.8
11 courses.csv

with open("courses.json", encoding="utf-8") as f:
    again = json.load(f)
print(again == courses)

with open("courses.csv", encoding="utf-8", newline="") as f:
    rows = list(csv.DictReader(f))
print(rows[0])
print(type(rows[0]["hours"]).__name__)

True
{'id': '1', 'title': 'How AI Works', 'level': 'beginner', 'hours': '3', 'rating': '4.7'}
str

# file: practice_client.py
"""A small, reusable client for the BitByte Practice API."""
import os
import random
import time
from typing import NamedTuple

import requests

RETRY_STATUSES = {429, 500, 502, 503, 504}


class APIError(Exception):
    """The API answered with an error we will not (or can no longer) retry."""

    def __init__(self, message, status=None):
        super().__init__(message)
        self.status = status


class ChatReply(NamedTuple):
    text: str
    usage: dict


class PracticeClient:
    def __init__(self, base_url=None, token=None, timeout=(3.05, 30), max_attempts=4,
                 backoff=0.1, jitter=0.25, sleep=time.sleep):
        self.base_url = (base_url or os.environ["PRACTICE_BASE_URL"]).rstrip("/")
        self.timeout = timeout
        self.max_attempts = max_attempts
        self.backoff = backoff
        self.jitter = jitter
        self._sleep = sleep
        self.session = requests.Session()
        self.session.headers["Accept"] = "application/json"
        token = token or os.environ.get("PRACTICE_TOKEN")
        if token:
            self.session.headers["Authorization"] = f"Bearer {token}"
        self.calls = 0
        self.usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}

    def __enter__(self):
        return self

    def __exit__(self, *exc_info):
        self.session.close()

    def _delay(self, attempt, response=None):
        delay = min(5.0, self.backoff * 2 ** (attempt - 1))
        delay *= 1 + random.uniform(0, self.jitter)
        header = response.headers.get("Retry-After", "") if response is not None else ""
        if header.isdigit():
            delay = max(delay, min(int(header), 30))
        return delay

    @staticmethod
    def _detail(response):
        try:
            return response.json()["detail"]
        except (ValueError, KeyError, TypeError):
            return response.text[:100]

    def request(self, method, path, **kwargs):
        """Send a request with retries; return the parsed JSON or raise APIError."""
        url = f"{self.base_url}{path}"
        kwargs.setdefault("timeout", self.timeout)
        for attempt in range(1, self.max_attempts + 1):
            last = attempt == self.max_attempts
            try:
                response = self.session.request(method, url, **kwargs)
            except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as error:
                if last:
                    raise APIError(f"{method} {path} failed: {type(error).__name__}") from error
                self._sleep(self._delay(attempt))
                continue
            if response.status_code in RETRY_STATUSES and not last:
                self._sleep(self._delay(attempt, response))
                continue
            if not response.ok:
                message = f"{method} {path} -> {response.status_code}: {self._detail(response)}"
                raise APIError(message, response.status_code)
            return response.json()

    def get(self, path, **params):
        return self.request("GET", path, params=params)

    def courses(self, **filters):
        """Yield every course, following next_page."""
        page = 1
        while page:
            data = self.get("/v1/courses", page=page, per_page=4, **filters)
            yield from data["items"]
            page = data["next_page"]

    def chat(self, messages, model="practice-llm-1"):
        """Send a conversation (non-streaming) and return a ChatReply."""
        data = self.request("POST", "/v1/chat", json={"model": model, "messages": messages})
        usage = data["usage"]
        self.calls += 1
        for name in self.usage:
            self.usage[name] += usage[name]
        return ChatReply(data["message"]["content"], usage)

from practice_client import APIError, PracticeClient

waits = []
client = PracticeClient(jitter=0, sleep=waits.append)

print(client.get("/health"))
print(client.get("/v1/flaky", key="demo-6", fail_times=2))
print("waited between retries:", waits)

try:
    client.get("/v1/courses/99")
except APIError as error:
    print(error, "| status", error.status, "| waits unchanged:", waits)

print([c["title"] for c in client.courses(level="intermediate")])

{'status': 'ok'}
{'ok': True, 'attempt': 3}
waited between retries: [0.1, 0.2]
GET /v1/courses/99 -> 404: Course 99 not found | status 404 | waits unchanged: [0.1, 0.2]
['Prompt Engineering', 'LLM APIs']

token = os.environ["PRACTICE_TOKEN"]
body = {"model": "practice-llm-1", "messages": [{"role": "user", "content": "What is an API?"}]}
r = requests.post(f"{BASE}/v1/chat", json=body, headers={"Authorization": f"Bearer {token}"},
                  timeout=(3.05, 60))
print(r.status_code)
print(json.dumps(r.json(), indent=2))

200
{
  "model": "practice-llm-1",
  "message": {
    "role": "assistant",
    "content": "This is a practice answer about: What is an API. A real model would say more."
  },
  "usage": {
    "prompt_tokens": 4,
    "completion_tokens": 16,
    "total_tokens": 20
  }
}

for bad in (None, "token-wrong"):
    anonymous = PracticeClient(token=bad, sleep=waits.append)
    anonymous.session.headers.pop("Authorization", None)
    if bad:
        anonymous.session.headers["Authorization"] = f"Bearer {bad}"
    try:
        anonymous.chat([{"role": "user", "content": "hi"}])
    except APIError as error:
        print(error.status, error)

401 POST /v1/chat -> 401: Missing bearer token
401 POST /v1/chat -> 401: Invalid or expired token

history = [{"role": "system", "content": "You are a concise tutor."}]
for question in ["What is an API?", "Why use timeouts?", "What is a retry?"]:
    history.append({"role": "user", "content": question})
    reply = client.chat(history)
    history.append({"role": "assistant", "content": reply.text})
    print(reply.usage["prompt_tokens"], reply.usage["completion_tokens"], "|", reply.text)

print(client.calls, "calls, totals:", client.usage)

9 16 | This is a practice answer about: What is an API. A real model would say more.
28 15 | This is a practice answer about: Why use timeouts. A real model would say more.
47 16 | This is a practice answer about: What is a retry. A real model would say more.
3 calls, totals: {'prompt_tokens': 84, 'completion_tokens': 47, 'total_tokens': 131}

PRICE_PER_MILLION = {"input": 2.00, "output": 10.00}   # উদাহরণ হিসেবে USD-তে দাম


def estimate_cost(usage, price=PRICE_PER_MILLION):
    return (usage["prompt_tokens"] * price["input"]
            + usage["completion_tokens"] * price["output"]) / 1_000_000


per_call = estimate_cost(client.usage) / client.calls
print(f"cost of the conversation: ${estimate_cost(client.usage):.6f}")
print(f"if 1000 users make 30 such calls a month: ${per_call * 1000 * 30:.2f}")

cost of the conversation: $0.000638
if 1000 users make 30 such calls a month: $6.38

# pip install anthropic       (এনভায়রনমেন্টে ANTHROPIC_API_KEY লাগবে)
import anthropic

client = anthropic.Anthropic(timeout=60.0, max_retries=2)   # ANTHROPIC_API_KEY নিজেই পড়ে

message = client.messages.create(
    model="claude-sonnet-5-5",   # মডেলের নাম বদলায়: প্রোভাইডারের মডেল তালিকা দেখুন
    max_tokens=1024,             # উত্তরের দৈর্ঘ্যের সীমা সবসময় দিন
    system="You are a concise tutor.",
    messages=[{"role": "user", "content": "What is an API?"}],
)
print(message.content[0].text)
print(message.usage.input_tokens, message.usage.output_tokens)

# pip install openai          (এনভায়রনমেন্টে OPENAI_API_KEY লাগবে)
from openai import OpenAI

client = OpenAI(timeout=60.0, max_retries=2)   # OPENAI_API_KEY নিজেই পড়ে

response = client.responses.create(
    model=os.environ["OPENAI_MODEL"],   # প্রোভাইডারের ডকুমেন্টেশন থেকে একটা বর্তমান মডেল বেছে নিন
    instructions="You are a concise tutor.",
    input="What is an API?",
    max_output_tokens=1024,
)
print(response.output_text)
print(response.usage.input_tokens, response.usage.output_tokens)

# 1. বিলম্ব
print([backoff_delay(n, base=0.5, cap=4) for n in range(1, 6)])

# 2. লম্বা বিগিনার কোর্স
long_ones = [c for c in client.courses(level="beginner") if c["hours"] > 8]
for c in sorted(long_ones, key=lambda c: c["hours"]):
    print(c["hours"], c["title"])

[0.5, 1.0, 2.0, 4, 4]
9 Git, APIs and Backend
10 SQL and Data Handling
12 Python for AI

# 3. হাল ছাড়া বনাম চালিয়ে যাওয়া
a = request_with_retries("GET", f"{BASE}/v1/flaky", params={"key": "ex-1", "fail_times": 3},
                         max_attempts=3, jitter=False, base=0.05)
b = request_with_retries("GET", f"{BASE}/v1/flaky", params={"key": "ex-2", "fail_times": 3},
                         max_attempts=5, jitter=False, base=0.05)
print(a.status_code, b.status_code, b.json())

# 4. উত্তর সেভ করা
records = []
for question in ["What is a token?", "What is a timeout?", "What is JSON?"]:
    reply = client.chat([{"role": "user", "content": question}])
    records.append({"question": question, "answer": reply.text, **reply.usage})
with open("answers.json", "w", encoding="utf-8") as f:
    json.dump(records, f, indent=2, ensure_ascii=False)
with open("answers.json", encoding="utf-8") as f:
    saved = json.load(f)
print(len(saved), sum(r["total_tokens"] for r in saved))

503 200 {'ok': True, 'attempt': 4}
3 58

# 5. টোকেন বাজেট
class BudgetClient(PracticeClient):
    def __init__(self, max_total_tokens, **kwargs):
        super().__init__(**kwargs)
        self.max_total_tokens = max_total_tokens

    def chat(self, messages, model="practice-llm-1"):
        if self.usage["total_tokens"] >= self.max_total_tokens:
            raise APIError("token budget exceeded")
        return super().chat(messages, model)


budgeted = BudgetClient(40)
for number in range(1, 10):
    try:
        reply = budgeted.chat([{"role": "user", "content": f"Question number {number}"}])
    except APIError as error:
        print(f"stopped before call {number}:", error, budgeted.usage["total_tokens"])
        break

stopped before call 4: token budget exceeded 54